diff --git a/electron/bridges/boundedSshExec.cjs b/electron/bridges/boundedSshExec.cjs index ab67bc61c..5aef2b71e 100644 --- a/electron/bridges/boundedSshExec.cjs +++ b/electron/bridges/boundedSshExec.cjs @@ -158,7 +158,7 @@ function executeBoundedSshCommand(sshClient, command, options = {}) { terminate: true, // Before the callback arrives ssh2 owns an uncancellable channel-open // request. Closing the physical transport is the only public cleanup. - invalidateTransport: !streamRef, + invalidateTransport: !streamRef && options.invalidateTransportOnAbort !== false, }); const append = (target, chunk) => { if (settled) return; diff --git a/electron/bridges/boundedSshExec.test.cjs b/electron/bridges/boundedSshExec.test.cjs index 53b6d9aea..f4ad5d44e 100644 --- a/electron/bridges/boundedSshExec.test.cjs +++ b/electron/bridges/boundedSshExec.test.cjs @@ -192,6 +192,23 @@ test("bounded SSH exec aborts before callback and terminates a late stream", asy assert.equal(timers.active.size, 0); }); +test("bounded SSH exec can abort a shared-channel probe without invalidating the transport", async () => { + let invalidations = 0; + const controller = new AbortController(); + const sshClient = { + exec() {}, + destroy() { invalidations += 1; }, + }; + const result = executeBoundedSshCommand(sshClient, "probe", { + signal: controller.signal, + invalidateTransportOnAbort: false, + }); + + controller.abort(new Error("cancelled")); + await assert.rejects(result, /cancelled/); + assert.equal(invalidations, 0); +}); + test("bounded SSH exec clears its run deadline after normal completion", async () => { const stream = createStream(); const timers = trackedTimerApi(); diff --git a/electron/bridges/sshBridge/sessionOps.cjs b/electron/bridges/sshBridge/sessionOps.cjs index ce3e48643..afdee829e 100644 --- a/electron/bridges/sshBridge/sessionOps.cjs +++ b/electron/bridges/sshBridge/sessionOps.cjs @@ -666,7 +666,7 @@ function createSessionOpsApi(ctx) { // Resolve the directory the running `rz` writes to (its own cwd) and report // which of `names` already exist there. Returns { dir, existing } or null. - async function probeReceiveConflicts(session, names) { + async function probeReceiveConflicts(session, names, { signal } = {}) { if (!session || !session.conn || !Array.isArray(names) || names.length === 0) { return null; } @@ -707,6 +707,8 @@ function createSessionOpsApi(ctx) { openingTimeoutMs: 5000, runTimeoutMs: 5000, maxOutputBytes: 1024 * 1024, + signal, + invalidateTransportOnAbort: false, }); let dir = null; const existing = []; const modes = {}; for (const line of out.split("\n")) { @@ -724,19 +726,32 @@ function createSessionOpsApi(ctx) { } // rm -f the given absolute remote paths (quoted; injection-safe). - async function removeRemoteFiles(session, paths) { + async function removeRemoteFiles(session, paths, { signal } = {}) { if (!session || !session.conn || !Array.isArray(paths) || paths.length === 0) return; const argv = paths.map((p) => quoteShellArg(p)).join(" "); + const commitToken = "NETCATTY_ZMODEM_COMMIT"; + const command = `exec sh -c ${quoteShellArg( + `IFS= read -r token || exit 125; ` + + `[ "$token" = ${quoteShellArg(commitToken)} ] || exit 125; ` + + `rm -f -- "$@"`, + )} sh ${argv}`; await executeBoundedSshCommand( session.conn, - `exec sh -c 'rm -f -- "$@"' sh ${argv}`, - { openingTimeoutMs: 5000, runTimeoutMs: 5000, maxOutputBytes: 64 * 1024 }, + command, + { + openingTimeoutMs: 5000, + runTimeoutMs: 5000, + maxOutputBytes: 64 * 1024, + signal, + invalidateTransportOnAbort: false, + onStream: (stream) => stream.end(`${commitToken}\n`), + }, ).catch(() => {}); } // chmod the given { path, mode } entries back to their captured permissions // (parameterized; injection-safe). Modes are validated octal before use. - async function restoreRemoteModes(session, entries) { + async function restoreRemoteModes(session, entries, { signal } = {}) { if (!session || !session.conn || !Array.isArray(entries) || entries.length === 0) return; const args = []; for (const e of entries) { @@ -746,10 +761,23 @@ function createSessionOpsApi(ctx) { } if (args.length === 0) return; const script = 'while [ "$#" -ge 2 ]; do chmod "$1" "$2" 2>/dev/null; shift 2; done'; + const commitToken = "NETCATTY_ZMODEM_COMMIT"; + const command = `exec sh -c ${quoteShellArg( + `IFS= read -r token || exit 125; ` + + `[ "$token" = ${quoteShellArg(commitToken)} ] || exit 125; ` + + script, + )} sh ${args.join(" ")}`; await executeBoundedSshCommand( session.conn, - `exec sh -c ${quoteShellArg(script)} sh ${args.join(" ")}`, - { openingTimeoutMs: 5000, runTimeoutMs: 5000, maxOutputBytes: 64 * 1024 }, + command, + { + openingTimeoutMs: 5000, + runTimeoutMs: 5000, + maxOutputBytes: 64 * 1024, + signal, + invalidateTransportOnAbort: false, + onStream: (stream) => stream.end(`${commitToken}\n`), + }, ).catch(() => {}); } diff --git a/electron/bridges/sshBridge/startSession.cjs b/electron/bridges/sshBridge/startSession.cjs index 4879c743f..2cd07a6c8 100644 --- a/electron/bridges/sshBridge/startSession.cjs +++ b/electron/bridges/sshBridge/startSession.cjs @@ -442,26 +442,43 @@ function createStartSessionApi(ctx) { interruptRemote() { try { stream.signal?.("INT"); } catch { /* ignore */ } }, - probeReceiveConflicts(names) { - return probeReceiveConflicts(sessions.get(sessionId), names); + probeReceiveConflicts(names, { signal } = {}) { + return probeReceiveConflicts(sessions.get(sessionId), names, { signal }); }, - removeRemoteFiles(paths) { - return removeRemoteFiles(sessions.get(sessionId), paths); + removeRemoteFiles(paths, { signal } = {}) { + return removeRemoteFiles(sessions.get(sessionId), paths, { signal }); }, - restoreRemoteModes(entries) { - return restoreRemoteModes(sessions.get(sessionId), entries); + restoreRemoteModes(entries, { signal } = {}) { + return restoreRemoteModes(sessions.get(sessionId), entries, { signal }); }, - requestOverwriteDecision(filename) { + requestOverwriteDecision(filename, { signal } = {}) { return new Promise((resolve) => { const requestId = randomUUID(); - const timer = setTimeout(() => { + let settled = false; + const cleanup = () => { + clearTimeout(timer); + try { signal?.removeEventListener("abort", onAbort); } catch { /* ignore */ } zmodemOverwritePending.delete(requestId); - resolve({ action: "skip", applyToRest: false }); + }; + const finish = (decision) => { + if (settled) return; + settled = true; + cleanup(); + resolve(decision); + }; + const onAbort = () => finish({ action: "cancel", applyToRest: false }); + const timer = setTimeout(() => { + finish({ action: "skip", applyToRest: false }); }, 120000); - zmodemOverwritePending.set(requestId, (payload) => { - clearTimeout(timer); - resolve({ action: payload.action, applyToRest: !!payload.applyToRest }); - }); + zmodemOverwritePending.set(requestId, (payload) => finish({ + action: payload.action, + applyToRest: !!payload.applyToRest, + })); + if (signal?.aborted) { + onAbort(); + return; + } + signal?.addEventListener("abort", onAbort, { once: true }); safeSend(getCurrentSessionWebContents(), "netcatty:zmodem:overwrite-request", { sessionId, requestId, filename, }); diff --git a/electron/bridges/zmodemFastPath.cjs b/electron/bridges/zmodemFastPath.cjs new file mode 100644 index 000000000..4edfe018f --- /dev/null +++ b/electron/bridges/zmodemFastPath.cjs @@ -0,0 +1,1107 @@ +"use strict"; + +/** + * ZMODEM receive fast path for zmodem.js (0.1.10). + * + * The library processes every incoming byte as an element of a + * JavaScript number[] array: Sentry.consume() converts Buffers with + * `Array.prototype.slice.call(new Uint8Array(input))`, subpackets are + * ZDLE-decoded and CRC-checked with per-byte JS loops, and the decoded + * payloads are handed to on_input as number[] again. That costs tens of + * JS ops per byte and caps `sz` download throughput in the low tens of + * MB/s even though ssh2 can deliver ~100MB/s. The fast path deliberately + * keeps receive payloads as Uint8Array instances so the download bridge can + * pass them to the file stream without another full-payload copy. + * + * This module patches the library's prototypes so that while a receive + * session is streaming ZDATA subpackets, bytes stay as Buffers the whole + * way: the frame-end scan uses Buffer.indexOf (native memchr), the + * payload is ZDLE-decoded into a fresh Uint8Array in one pass, and + * CRC16/CRC32 run over typed arrays with precomputed tables. Headers are + * tiny and rare, so they keep the original array-based pipeline + * untouched — protocol state machines, error recovery, wire bytes, and + * event ordering behave exactly as before. The receive payload container is + * intentionally Uint8Array on this path for throughput. + * + * The same module also carries `applyZmodemSendSessionFixes()`, which + * patches zmodem.js Send-session correctness hazards that make `rz` + * uploads fail intermittently. Those are protocol fixes, not a + * throughput fast path, and are not gated by the kill-switch below. + * + * Kill-switch: set NETCATTY_ZMODEM_FAST_PATH=0 to disable both performance + * fast paths. The send-session correctness fixes remain enabled. + */ + +const ZDLE = 0x18; +const XON = 0x11; +const XOFF = 0x13; +const XON_HIGH = XON | 0x80; // 0x91 +const XOFF_HIGH = XOFF | 0x80; // 0x93 +const OVER_AND_OUT = [79, 79]; // "OO" + +// frame-end byte (104-107) → Subpacket.build() frameend key +const FRAME_END_KEYS = { + 104: "end_no_ack", // ZCRCE - frame ends, no ack + 105: "no_end_no_ack", // ZCRCG - frame continues + 106: "no_end_ack", // ZCRCQ - frame continues, ack expected + 107: "end_ack", // ZCRCW - frame ends, ack expected +}; + +const EMPTY_BUFFER = Buffer.alloc(0); + +function isFastPathDisabled() { + return process.env.NETCATTY_ZMODEM_FAST_PATH === "0"; +} + +//---------------------------------------------------------------------- +// CRC16 (CRC-CCITT/XModem) — replicates zcrc.js `_compute_crctab()`. +//---------------------------------------------------------------------- +const CRC16_TABLE = new Uint32Array(256); +for (let divident = 0; divident < 256; divident++) { + let currByte = (divident << 8) & 0xffff; + for (let bit = 0; bit < 8; bit++) { + currByte = + currByte & 0x8000 + ? ((currByte << 1) ^ 0x1021) & 0xffff + : (currByte << 1) & 0xffff; + } + CRC16_TABLE[divident] = currByte; +} + +//---------------------------------------------------------------------- +// CRC32 (standard reflected, poly 0xEDB88320) — matches the crc-32 +// package that zmodem.js uses via `CRC32_MOD.buf()`. +//---------------------------------------------------------------------- +const CRC32_TABLE = new Uint32Array(256); +for (let n = 0; n < 256; n++) { + let c = n; + for (let k = 0; k < 8; k++) { + c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1; + } + CRC32_TABLE[n] = c >>> 0; +} + +/** One CRC16 step; identical to zcrc.js `_updcrc()`. */ +function crc16Step(crc, byte) { + return CRC16_TABLE[(crc >> 8) & 255] ^ ((255 & crc) << 8) ^ byte; +} + +/** + * CRC16 over decoded payload + frame-end byte, including the two + * zero-byte passes that zmodem.js `CRC.crc16()` appends. + * + * TABLE[0] is 0, so seeding with 0 and updating with the first byte is + * equivalent to the library seeding with the first byte. + */ +function crc16Bytes(payload, frameEndNum) { + let crc = 0; + for (let i = 0; i < payload.length; i++) { + crc = crc16Step(crc, payload[i]); + } + crc = crc16Step(crc, frameEndNum); + crc = crc16Step(crc, 0); + crc = crc16Step(crc, 0); + return crc & 0xffff; +} + +/** CRC32 over decoded payload + frame-end byte (standard init/final xor). */ +function crc32Bytes(payload, frameEndNum) { + let crc = 0xffffffff; + for (let i = 0; i < payload.length; i++) { + crc = CRC32_TABLE[(crc ^ payload[i]) & 255] ^ (crc >>> 8); + } + crc = CRC32_TABLE[(crc ^ frameEndNum) & 255] ^ (crc >>> 8); + return (crc ^ 0xffffffff) >>> 0; +} + +/** + * ZDLE-decode `u8` into a fresh Uint8Array. + * + * Decoding is `second_byte - 64`, exactly like zmodem.js's + * `octets[o + 1] - 64`. (XOR 0x40 would only be equivalent for escaped + * values < 0x80; conforming encoders never emit escape pairs whose + * second byte is in 0x80-0xBF, but we replicate the library verbatim.) + * A trailing bare ZDLE is malformed input; it decodes to ZDLE-64 here + * and the CRC check downstream rejects it, same as the library. + */ +function zdleDecode(u8) { + const out = new Uint8Array(u8.length); // never longer than encoded + let w = 0; + for (let i = 0; i < u8.length; i++) { + let b = u8[i]; + if (b === ZDLE) { + i += 1; + b = u8[i] - 64; + } + out[w++] = b; + } + return out.subarray(0, w); +} + +/** + * Read `count` ZDLE-decoded bytes starting at `start` in `u8`. + * + * Returns null if the encoded bytes aren't fully available (including a + * trailing bare ZDLE), mirroring `Zmodem.ZDLE.splice()`. + */ +function readDecodedBytes(u8, start, count) { + const bytes = new Uint8Array(count); + let i = start; + let got = 0; + while (got < count) { + if (i >= u8.length) return null; + let b = u8[i]; + if (b === ZDLE) { + i += 1; + if (i >= u8.length) return null; // bare trailing ZDLE + b = u8[i] - 64; + } + bytes[got++] = b; + i += 1; + } + return { bytes, consumed: i - start }; +} + +/** + * Replicates Zmodem.ZMLIB.strip_ignored_bytes() (XON/XOFF and their + * high-bit variants) without the per-byte Array.splice(). + * Zero-copy when there's nothing to strip. + */ +function stripIgnoredBytesFast(u8) { + let dirty = false; + for (let i = 0; i < u8.length; i++) { + const b = u8[i]; + if (b === XON || b === XON_HIGH || b === XOFF || b === XOFF_HIGH) { + dirty = true; + break; + } + } + if (!dirty) return u8; + + const out = new Uint8Array(u8.length); + let w = 0; + for (let i = 0; i < u8.length; i++) { + const b = u8[i]; + if (b === XON || b === XON_HIGH || b === XOFF || b === XOFF_HIGH) { + continue; + } + out[w++] = b; + } + return out.subarray(0, w); +} + +/** + * Fast equivalent of zsubpacket.js `Subpacket._parse()` for Buffer input. + * + * - Returns null when the buffer doesn't (yet) hold a complete subpacket + * (i.e., the caller should keep accumulating bytes). + * - Throws `new Zmodem.Error("crc", got, expected)` on a CRC mismatch, + * exactly like the library's CRC.verify16()/verify32(). + * + * @param {Object} Zmodem - The zmodem.js module (used for + * Subpacket.build() and the Zmodem.Error class). + * @param {Buffer} buf - Encoded bytes, starting at a subpacket boundary. + * @param {number} crcLen - 2 for CRC16, 4 for CRC32. + * @returns {{ subpacket: Object, consumed: number } | null} + */ +function parseSubpacketFast(Zmodem, buf, crcLen) { + // Find the first ZDLE followed by a frame-end byte (104-107). ZDLE + // escaping guarantees no payload byte can decode to a frame-end value, + // so the first such pair is the marker — the same scan the library + // does in Subpacket._parse(). + let zdleAt = -1; + let frameEndKey = null; + let frameEndNum = 0; + while (true) { + zdleAt = buf.indexOf(ZDLE, zdleAt + 1); + if (zdleAt === -1) return null; // no marker yet → need more data + if (zdleAt + 1 >= buf.length) return null; // trailing ZDLE → need more data + frameEndNum = buf[zdleAt + 1]; + frameEndKey = FRAME_END_KEYS[frameEndNum]; + if (frameEndKey) break; + } + + // Payload is everything before the marker's ZDLE, ZDLE-decoded. + const payload = zdleDecode(buf.subarray(0, zdleAt)); + + // The CRC bytes follow the marker and are individually escaped. + const got = readDecodedBytes(buf, zdleAt + 2, crcLen); + if (!got) return null; // CRC straddles chunks → need more data + + // Verify the CRC over decoded payload + frame-end byte. + let expected; + if (crcLen === 2) { + const v = crc16Bytes(payload, frameEndNum); + expected = [(v >> 8) & 0xff, v & 0xff]; + } else { + const v = crc32Bytes(payload, frameEndNum); + expected = [v & 0xff, (v >> 8) & 0xff, (v >> 16) & 0xff, (v >> 24) & 0xff]; + } + for (let i = 0; i < crcLen; i++) { + if (got.bytes[i] !== expected[i]) { + throw new Zmodem.Error( + "crc", + Array.prototype.slice.call(got.bytes), + expected, + ); + } + } + + // Keep the decoded payload typed through the fast receive path. The + // application download handler turns this into a zero-copy Buffer view; + // converting to a number[] here would erase the throughput gain with a + // full payload allocation on every subpacket. + const subpacket = Zmodem.Subpacket.build(payload, frameEndKey); + return { subpacket, consumed: zdleAt + 2 + got.consumed }; +} + +/** Replicates `_trim_OO()` from zsession.js. */ +function trimOverAndOut(Zmodem, array) { + if (0 === Zmodem.ZMLIB.find_subarray(array, OVER_AND_OUT)) { + array.splice(0, OVER_AND_OUT.length); + } else if (array[0] === OVER_AND_OUT[OVER_AND_OUT.length - 1]) { + array.splice(0, 1); + } + return array; +} + +/** + * zsentry.js declares its Detection class privately; the patched + * Sentry.consume() needs to hand on_detect a Detection object of the + * same shape. Recreate it faithfully (confirm/deny/is_valid/ + * get_session_role — callers only ever use these methods). + */ +class Detection { + constructor(session_type, accepter, denier, checker) { + this._confirmer = accepter; + this._denier = denier; + this._is_valid = checker; + this._session_type = session_type; + } + + confirm() { + return this._confirmer.apply(this, arguments); + } + + deny() { + return this._denier.apply(this, arguments); + } + + is_valid() { + return this._is_valid.apply(this, arguments); + } + + get_session_role() { + return this._session_type; + } +} + +//---------------------------------------------------------------------- +// The patch itself. +//---------------------------------------------------------------------- + +/** + * Patch zmodem.js so receive sessions consume Buffers on a fast path. + * + * Safe to call more than once (idempotent), and a no-op when the + * NETCATTY_ZMODEM_FAST_PATH kill-switch is set. + * + * @param {Object} Zmodem - The zmodem.js module, as returned by require(). + * @returns {Object} The same module object, patched (or not). + */ +function applyZmodemFastPath(Zmodem) { + // Kill-switch for diagnosing protocol trouble in the field. + if (isFastPathDisabled()) return Zmodem; + if (Zmodem.__netcattyZmodemFastPathApplied) return Zmodem; + + const sentryProto = Zmodem?.Sentry?.prototype; + const sessionProto = Zmodem?.Session?.prototype; + if ( + !sentryProto || + !sessionProto || + typeof sentryProto.consume !== "function" || + typeof sessionProto.consume !== "function" || + !Zmodem.ZMLIB?.ABORT_SEQUENCE + ) { + return Zmodem; + } + + const ORIGINAL_SENTRY_CONSUME = sentryProto.consume; + const ORIGINAL_SESSION_CONSUME = sessionProto.consume; + const ABORT_SEQUENCE_BUF = Buffer.from(Zmodem.ZMLIB.ABORT_SEQUENCE); + + //-------------------------------------------------------------------- + // Session-level helpers. Bytes are kept in `_zmodem_fast_chunks`, a + // list of zero-copy Buffer views of the incoming chunks; consumed + // prefixes are dropped with subarray() instead of array.splice(). + //-------------------------------------------------------------------- + + function fastPush(u8) { + if (!u8.length) return; + if (!this._zmodem_fast_chunks) this._zmodem_fast_chunks = []; + this._zmodem_fast_chunks.push( + Buffer.isBuffer(u8) + ? u8 + : Buffer.from(u8.buffer, u8.byteOffset, u8.byteLength), + ); + this._zmodem_fast_length = (this._zmodem_fast_length || 0) + u8.length; + } + + /** A single Buffer with all pending fast bytes (concats on demand). */ + function fastContiguous() { + const chunks = this._zmodem_fast_chunks; + if (!chunks || !chunks.length) return EMPTY_BUFFER; + if (chunks.length > 1) { + this._zmodem_fast_chunks = [Buffer.concat(chunks)]; + // The chunk-index cursors refer to the old list. The frame marker, if + // any, is intentionally retained while its CRC bytes are still pending; + // the abort scan must restart on the merged buffer. + this._zmodem_fast_sequence_scan = null; + } + return this._zmodem_fast_chunks[0]; + } + + function fastLen() { + return this._zmodem_fast_length || 0; + } + + /** Find a small byte sequence incrementally without flattening chunks. */ + function fastFindSequence(needle) { + const chunks = this._zmodem_fast_chunks; + if (!chunks || !chunks.length || !needle.length) return -1; + let scan = this._zmodem_fast_sequence_scan; + if (!scan || scan.needle !== needle) { + scan = { needle, chunkIndex: 0, byteIndex: 0, offset: 0, matched: 0 }; + this._zmodem_fast_sequence_scan = scan; + } + for (let chunkIndex = scan.chunkIndex; chunkIndex < chunks.length; chunkIndex++) { + const chunk = chunks[chunkIndex]; + const start = chunkIndex === scan.chunkIndex ? scan.byteIndex : 0; + for (let i = start; i < chunk.length; i++) { + const byte = chunk[i]; + const offset = scan.offset++; + if (byte === needle[scan.matched]) { + scan.matched += 1; + if (scan.matched === needle.length) return offset - needle.length + 1; + } else { + scan.matched = byte === needle[0] ? 1 : 0; + } + } + scan.chunkIndex = chunkIndex + 1; + scan.byteIndex = 0; + } + scan.chunkIndex = chunks.length; + scan.byteIndex = 0; + return -1; + } + + /** Find a ZDLE frame-end marker incrementally without flattening a packet. */ + function fastFindFrameEnd() { + const chunks = this._zmodem_fast_chunks; + if (!chunks || !chunks.length) return -1; + const existing = this._zmodem_fast_frame_end_at; + if (existing !== undefined) return existing; + let scan = this._zmodem_fast_frame_scan; + if (!scan) { + scan = { chunkIndex: 0, byteIndex: 0, offset: 0, previousWasZdle: false }; + this._zmodem_fast_frame_scan = scan; + } + for (let chunkIndex = scan.chunkIndex; chunkIndex < chunks.length; chunkIndex++) { + const chunk = chunks[chunkIndex]; + const start = chunkIndex === scan.chunkIndex ? scan.byteIndex : 0; + for (let i = start; i < chunk.length; i++) { + const byte = chunk[i]; + const offset = scan.offset++; + if (scan.previousWasZdle && FRAME_END_KEYS[byte]) { + this._zmodem_fast_frame_end_at = offset - 1; + return this._zmodem_fast_frame_end_at; + } + scan.previousWasZdle = byte === ZDLE; + } + scan.chunkIndex = chunkIndex + 1; + scan.byteIndex = 0; + } + scan.chunkIndex = chunks.length; + scan.byteIndex = 0; + return -1; + } + + /** Drop the first `n` fast bytes (like Array.splice(0, n)). */ + function fastConsumeAt(n) { + const chunks = this._zmodem_fast_chunks; + const available = this._zmodem_fast_length || 0; + const toConsume = Math.min(Math.max(0, n), available); + while (n > 0 && chunks.length) { + const head = chunks[0]; + if (head.length > n) { + chunks[0] = head.subarray(n); + n = 0; + } else { + chunks.shift(); + n -= head.length; + } + } + this._zmodem_fast_length = available - toConsume; + this._zmodem_fast_frame_end_at = undefined; + this._zmodem_fast_frame_scan = null; + this._zmodem_fast_sequence_scan = null; + } + + /** Move all pending fast bytes into `_input_buffer` (header parsing). */ + function fastMoveAllToArray() { + const buf = this._zmodem_fast_contiguous(); + this._zmodem_fast_chunks = []; + this._zmodem_fast_length = 0; + this._zmodem_fast_frame_end_at = undefined; + this._zmodem_fast_frame_scan = null; + this._zmodem_fast_sequence_scan = null; + // push.apply() takes up to ~32k args per call; chunk defensively. + for (let i = 0; i < buf.length; i += 0x8000) { + Array.prototype.push.apply(this._input_buffer, buf.subarray(i, i + 0x8000)); + } + } + + /** + * Move array-side bytes back into the fast chunks, before them. + * Used when a ZDATA header and its first subpacket(s) arrive in one + * chunk: the header parser leaves the data bytes in `_input_buffer`, + * and they must be consumed in stream order. + */ + function fastMoveArrayToFast() { + if (!this._input_buffer.length) return; + const arr = this._input_buffer.splice(0); + if (!this._zmodem_fast_chunks) this._zmodem_fast_chunks = []; + this._zmodem_fast_chunks.unshift(Buffer.from(arr)); + this._zmodem_fast_length = arr.length + (this._zmodem_fast_length || 0); + this._zmodem_fast_frame_end_at = undefined; + this._zmodem_fast_frame_scan = null; + this._zmodem_fast_sequence_scan = null; + } + + /** Mirrors `_check_for_abort_sequence()`. */ + function fastCheckAbort() { + // Header parsing leaves incomplete frames in _input_buffer. Before a + // fast chunk is parsed, move it beside that buffered prefix so CAN x5 + // is detected even when the cancel sequence crosses the boundary. + // This also lets the upstream helper own the removal semantics. + if (this._input_buffer.length) { + this._zmodem_fast_move_all_to_array(); + return this._check_for_abort_sequence(); + } + + const at = this._zmodem_fast_find_sequence(ABORT_SEQUENCE_BUF); + if (at === -1) return false; + + this._zmodem_fast_consume_at(at + ABORT_SEQUENCE_BUF.length); + this._aborted = true; + this._on_session_end(); + throw new Zmodem.Error("peer_aborted"); + } + + /** Mirrors `_parse_and_consume_subpacket()`, parsing from fast bytes. */ + function fastParseSubpacket() { + if (this._zmodem_fast_find_frame_end() === -1) return null; + const buf = this._zmodem_fast_contiguous(); + + const crcLen = this._last_header_crc === 16 ? 2 : 4; + const parsed = parseSubpacketFast(Zmodem, buf, crcLen); + if (!parsed) return null; + + this._zmodem_fast_consume_at(parsed.consumed); + + if (Zmodem.DEBUG) { + console.debug(this.type, "RECEIVED SUBPACKET", parsed.subpacket); + } + + this._consume_data(parsed.subpacket); + + if (parsed.subpacket.frame_end()) { + this._next_subpacket_handler = null; + } + + return parsed.subpacket; + } + + /** Mirrors the `_consume_first()` do-while for the receive path. */ + function fastConsumeLoop() { + while (true) { + if (this._next_subpacket_handler) { + // A ZDATA header and its first subpacket may have arrived in one + // chunk and been routed through the array pipeline; pull those + // bytes back in order before fast-parsing. + this._zmodem_fast_move_array_to_fast(); + + if (!this._zmodem_fast_len()) break; + if (!this._zmodem_fast_parse_subpacket()) break; + } else { + // Headers are tiny and rare; hand bytes to the original + // array-based header pipeline. + this._zmodem_fast_move_all_to_array(); + + if (!this._input_buffer.length) break; + if (!this._parse_and_consume_header()) break; + } + } + } + + /** + * Fast equivalent of Session.consume() for a receive session fed + * with Uint8Array/Buffer input. + */ + function fastSessionConsume(octets) { + this._before_consume(octets); + + if (this._aborted) throw new Zmodem.Error("already_aborted"); + if (!octets.length) return; + + // Replicates _strip_and_enqueue_input(). The original path strips + // `octets` in place and keeps that same reference as + // _bytes_being_consumed; keep the stripped bytes so the post-"OO" + // trailing-bytes logic sees the same thing (per chunk, as before). + const stripped = stripIgnoredBytesFast(octets); + this._bytes_being_consumed = stripped; + + this._zmodem_fast_push(stripped); + + // The original Session.consume() checks for CAN x5 before its special + // post-ZFIN OO handling. Preserve that ordering on the fast path. + this._zmodem_fast_check_abort(); + + if (this._got_ZFIN) { + // Session is ending: only "OO" + trailing prompt bytes may follow. + // _trim_OO() and the header parser use Array.splice(), so route + // through the array pipeline (once, at session end). + this._zmodem_fast_move_all_to_array(); + + if (this._input_buffer.length < 2) return; + + if (Zmodem.ZMLIB.find_subarray(this._input_buffer, OVER_AND_OUT) === 0) { + // This doubles as an indication that the session has ended. + this._bytes_after_OO = trimOverAndOut( + Zmodem, + Array.prototype.slice.call(this._bytes_being_consumed), + ); + this._on_session_end(); + return; + } + + throw ( + "PROTOCOL: Only thing after ZFIN should be “OO” (79,79), not: " + + this._input_buffer.join() + ); + } + + this._zmodem_fast_consume_loop(); + } + + try { + sentryProto.consume = function consume(input) { + let consumedViaFastPath = false; + + if (!(input instanceof Array)) { + if (this._zsession && input instanceof Uint8Array) { + // Fast path: hand raw bytes straight to the session instead of + // converting every byte into a JS-array element (that + // conversion is what caps receive throughput). The session + // falls back to the original pipeline internally for send + // sessions and array input. + consumedViaFastPath = true; + + const session = this._zsession; + session.consume(input); + + if (!session.has_ended()) return; + + if (session.type === "receive") { + input = session.get_trailing_bytes(); + } else { + input = []; + } + } else { + input = Array.prototype.slice.call(new Uint8Array(input)); + } + } + + // Everything below is verbatim from the library's Sentry.consume() + // (zsentry.js), minus the fast-path branch above. + if (this._zsession && !consumedViaFastPath) { + var session_before_consume = this._zsession; + + session_before_consume.consume(input); + + if (session_before_consume.has_ended()) { + if (session_before_consume.type === "receive") { + input = session_before_consume.get_trailing_bytes(); + } else { + input = []; + } + } else return; + } + + var new_session = this._parse(input); + var to_terminal = input; + + if (new_session) { + let replacement_detect = !!this._parsed_session; + + if (replacement_detect) { + //no terminal output if the new session is of the + //same type as the old + if (this._parsed_session.type === new_session.type) { + to_terminal = []; + } + + this._on_retract(); + } + + this._parsed_session = new_session; + + var sentry = this; + + function checker() { + return sentry._parsed_session === new_session; + } + + //This runs with the Sentry object as the context. + function accepter() { + if (!this.is_valid()) { + throw "Stale ZMODEM session!"; + } + + new_session.on("garbage", sentry._to_terminal); + + new_session.on( + "session_end", + sentry._after_session_end.bind(sentry), + ); + + new_session.set_sender(sentry._sender); + + delete sentry._parsed_session; + + return (sentry._zsession = new_session); + } + + function denier() { + if (!this.is_valid()) return; + } + + this._on_detect( + new Detection( + new_session.type, + accepter, + this._send_abort.bind(this), + checker, + ), + ); + } else { + var expired_session = this._parsed_session; + + this._parsed_session = null; + + if (expired_session) { + //If we got a single “C” after parsing a session, + //that means our peer is trying to downgrade to YMODEM. + //That won’t work, so we just send the ABORT_SEQUENCE + //right away. + if (to_terminal.length === 1 && to_terminal[0] === 67) { + this._send_abort(); + } + + this._on_retract(); + } + } + + this._to_terminal(to_terminal); + }; + + sessionProto.consume = function consume(octets) { + if (this.type === "receive" && octets instanceof Uint8Array) { + return this._zmodem_fast_consume(octets); + } + + // Everything else (send sessions, array input) keeps the original + // pipeline. + return ORIGINAL_SESSION_CONSUME.call( + this, + octets instanceof Uint8Array + ? Array.prototype.slice.call(octets) + : octets, + ); + }; + + Object.assign(sessionProto, { + _zmodem_fast_consume: fastSessionConsume, + _zmodem_fast_push: fastPush, + _zmodem_fast_contiguous: fastContiguous, + _zmodem_fast_len: fastLen, + _zmodem_fast_find_sequence: fastFindSequence, + _zmodem_fast_find_frame_end: fastFindFrameEnd, + _zmodem_fast_consume_at: fastConsumeAt, + _zmodem_fast_move_all_to_array: fastMoveAllToArray, + _zmodem_fast_move_array_to_fast: fastMoveArrayToFast, + _zmodem_fast_check_abort: fastCheckAbort, + _zmodem_fast_parse_subpacket: fastParseSubpacket, + _zmodem_fast_consume_loop: fastConsumeLoop, + }); + + Zmodem.__netcattyZmodemFastPathApplied = true; + } catch (err) { + // Never break ZMODEM transfers over a patch problem. + sentryProto.consume = ORIGINAL_SENTRY_CONSUME; + sessionProto.consume = ORIGINAL_SESSION_CONSUME; + console.error( + "[ZMODEM] fast path patch failed; using original pipeline:", + err && err.message ? err.message : err, + ); + } + + return Zmodem; +} + +//---------------------------------------------------------------------- +// Send-session robustness fixes (rz upload path) +//---------------------------------------------------------------------- +// +// zmodem.js's Send session has a few hazards that make `rz` uploads fail +// intermittently: +// +// 1. Duplicate-ZSINIT race: send_offer() -> _ensure_receiver_escapes_ctrl_chars() +// sends a ZSINIT whenever the receiver's ZRINIT lacks ESCCTL and no ZACK +// has been seen yet, without checking whether the 5s keepalive just sent +// one. rz ACKs both ZSINITs; the second ZACK lands in a handler state that +// no longer expects it and consume() throws "Unhandled header: ZACK", +// killing the upload mid-transfer. Fix: track an in-flight ZSINIT +// (_zsinit_pending) and share its ZACK instead of sending a duplicate. +// +// 2. Shared-ACK plumbing: the keepalive's ZACK handler only sets +// _got_ZSINIT_ZACK, so a waiter parked on a pending ZSINIT (from #1) +// would never wake up. All send-session ZACKs now route through +// _on_zsinit_ack(), which sets the flag and flushes every waiter. +// +// 3. _stop_keepalive() nulls _keep_alive_promise (typo) instead of +// _keepalive_promise, so the keepalive can never restart after the first +// offer. Fix the field name. +// +// These are protocol-correctness fixes, not a throughput fast path, so they +// are not gated by NETCATTY_ZMODEM_FAST_PATH. + +function applyZmodemSendSessionFixes(Zmodem) { + if (Zmodem.__netcattyZmodemSendSessionFixesApplied) return Zmodem; + + const Send = Zmodem.Session && Zmodem.Session.Send; + if (!Send) return Zmodem; + const proto = Send.prototype; + if ( + [ + proto._send_ZSINIT, + proto._start_keepalive, + proto._stop_keepalive, + proto._ensure_receiver_escapes_ctrl_chars, + ].some((method) => typeof method !== "function") + ) { + return Zmodem; + } + + const ORIGINAL_SEND_ZSINIT = proto._send_ZSINIT; + const ORIGINAL_START_KEEPALIVE = proto._start_keepalive; + const ORIGINAL_STOP_KEEPALIVE = proto._stop_keepalive; + const ORIGINAL_ENSURE_ESCAPES = proto._ensure_receiver_escapes_ctrl_chars; + + // Deliberately leave _consume_ZRINIT untouched. A non-zero receiver + // buffer requires ZCRCW/ZACK pacing; TCP backpressure is not equivalent. + // The upstream fail-closed check prevents silent receiver-buffer overrun + // until a protocol-level window implementation is added. + + try { + // One shared ACK path for every ZSINIT (keepalive or offer-time). + proto._on_zsinit_ack = function _on_zsinit_ack() { + this._got_ZSINIT_ZACK = true; + this._zsinit_pending = false; + const waiters = this._zsinit_ack_waiters; + this._zsinit_ack_waiters = null; + if (waiters) { + for (const res of waiters) { + try { + res(); + } catch { + /* ignore */ + } + } + } + }; + + // _send_ZSINIT is the only ZSINIT emitter (keepalive + ensure). + proto._send_ZSINIT = function _send_ZSINIT() { + this._zsinit_pending = true; + return ORIGINAL_SEND_ZSINIT.apply(this, arguments); + }; + + proto._start_keepalive = function _start_keepalive() { + if (!this._keepalive_promise) { + const sess = this; + + this._keepalive_promise = new Promise(function (resolve) { + sess._keepalive_timeout = setTimeout(resolve, 5000); + }).then(function () { + sess._keepalive_promise = null; + // Never overlap ZSINIT frames. The receiver sends one ZACK per + // frame; replacing the handler while an earlier ACK is pending + // turns the later ACK into an unhandled-header failure. + if (!sess._zsinit_pending) { + sess._next_header_handler = { + ZACK: function () { + sess._on_zsinit_ack(); + }, + }; + sess._send_ZSINIT(); + } + sess._start_keepalive(); + }); + } + }; + + proto._stop_keepalive = function _stop_keepalive() { + if (this._keepalive_promise) { + clearTimeout(this._keepalive_timeout); + this._keepalive_promise = null; + } + }; + + proto._ensure_receiver_escapes_ctrl_chars = function _ensure_receiver_escapes_ctrl_chars() { + let promise; + + const needs_ZSINIT = + !this._last_ZRINIT.escape_ctrl_chars() && !this._got_ZSINIT_ZACK; + + if (needs_ZSINIT) { + const sess = this; + + if (sess._zsinit_pending) { + // A keepalive ZSINIT is awaiting its ZACK: sending another would + // produce a stray ZACK that aborts the session as + // "Unhandled header: ZACK". Share the pending ACK instead. + sess._next_header_handler = { + ZACK: function () { + sess._on_zsinit_ack(); + }, + }; + promise = new Promise(function (res) { + if (!sess._zsinit_ack_waiters) sess._zsinit_ack_waiters = []; + sess._zsinit_ack_waiters.push(res); + }); + } else { + promise = new Promise(function (res) { + if (!sess._zsinit_ack_waiters) sess._zsinit_ack_waiters = []; + sess._zsinit_ack_waiters.push(res); + sess._next_header_handler = { + ZACK: function () { + sess._on_zsinit_ack(); + }, + }; + sess._send_ZSINIT(); + }); + } + } else { + promise = Promise.resolve(); + } + + return promise; + }; + + Zmodem.__netcattyZmodemSendSessionFixesApplied = true; + } catch (err) { + // Never break ZMODEM uploads over a patch problem. + proto._send_ZSINIT = ORIGINAL_SEND_ZSINIT; + proto._start_keepalive = ORIGINAL_START_KEEPALIVE; + proto._stop_keepalive = ORIGINAL_STOP_KEEPALIVE; + proto._ensure_receiver_escapes_ctrl_chars = ORIGINAL_ENSURE_ESCAPES; + delete proto._on_zsinit_ack; + console.error( + "[ZMODEM] send-session fixes patch failed; using original Send session:", + err && err.message ? err.message : err, + ); + } + + return Zmodem; +} + +//---------------------------------------------------------------------- +// Send-side fast path (rz uploads). +//---------------------------------------------------------------------- +// +// `_send_file_part()` splits every chunk into MAX_CHUNK_LENGTH (8192) +// byte subpackets and routes each one through the array-based encode +// pipeline: the Uint8Array chunk is converted to a boxed number Array, +// then `Subpacket._encode()` makes several more full-payload copies +// (`slice(0)`, `zencoder.encode()`, `concat(frameend)` for the CRC, +// the final `.concat()`), and `CRC.crc16()` calls `_updcrc()` per byte. +// At ~60 MB/s that is ~7700 subpackets per second and GB/s of +// short-lived arrays: CPU-bound in the fast phase and GC-stormed into +// a few MB/s when the collector catches up. This is the same array +// pipeline disease the receive fast path already cured for downloads. +// +// The replacement builds each subpacket frame straight into one Buffer +// (one pass for the CRC, one pass to ZDLE-escape) and is byte-for-byte +// identical to the original under the encoder configuration the send +// sessions always use (FORCE_ESCAPE_CTRL_CHARS => escape_ctrl_chars on). + +const SEND_FRAME_END_NUM = { no_end_no_ack: 0x69, end_no_ack: 0x68 }; + +/** + * ZDLE-escape rule for the send sessions' fixed encoder configuration + * (escape_ctrl_chars on, turbo_escape off): bytes 0x00-0x1f and + * 0x80-0x9f escape as ZDLE + (byte ^ 0x40); everything else passes + * through. Mirrors the zdle.js zsendline_tab for that configuration. + */ +function sendSessionEscapes(byte) { + return (byte & 0x60) === 0; +} + +/** + * Build one ZMODEM data subpacket frame into a fresh Buffer: ZDLE- + * escaped payload (the `length` bytes of `source` starting at `offset`), + * the [ZDLE, frameEndNum] trailer, then the ZDLE-escaped CRC16. + * + * Wire-identical to `Subpacket.build(chunk, frameend).encode16(encoder)` + * with escape_ctrl_chars on, including the two zero-byte CRC passes and + * the empty-payload case. + */ +function buildSendSubpacketFast(source, offset, length, frameEndNum) { + let crc = 0; + let escapedLen = 0; + for (let i = offset; i < offset + length; i++) { + const b = source[i]; + crc = crc16Step(crc, b); + escapedLen += sendSessionEscapes(b) ? 2 : 1; + } + crc = crc16Step(crc, frameEndNum); + crc = crc16Step(crc, 0); + crc = crc16Step(crc, 0); + crc &= 0xffff; + + const crcHi = crc >> 8; + const crcLo = crc & 0xff; + + const out = Buffer.allocUnsafe( + escapedLen + + 2 + + (sendSessionEscapes(crcHi) ? 2 : 1) + + (sendSessionEscapes(crcLo) ? 2 : 1), + ); + let w = 0; + for (let i = offset; i < offset + length; i++) { + const b = source[i]; + if (sendSessionEscapes(b)) { + out[w++] = 0x18; // ZDLE + out[w++] = b ^ 0x40; + } else { + out[w++] = b; + } + } + out[w++] = 0x18; // ZDLE + out[w++] = frameEndNum; + for (const b of [crcHi, crcLo]) { + if (sendSessionEscapes(b)) { + out[w++] = 0x18; + out[w++] = b ^ 0x40; + } else { + out[w++] = b; + } + } + return out; +} + +/** + * Patch zmodem.js so the send session builds data subpackets directly + * as Buffers instead of churning through the array pipeline (see the + * block comment above). Falls back to the original for anything that + * is not the fixed escape_ctrl_chars configuration or an unknown + * frame-end, so wire behavior is unchanged elsewhere. + * + * @param {Object} Zmodem - The zmodem.js module, as returned by require(). + * @returns {Object} The same module object, patched (or not). + */ +function applyZmodemSendFastPath(Zmodem) { + // The field kill-switch must cover both throughput patches. Keep the + // independent send-session correctness fixes active for diagnostics. + if (isFastPathDisabled()) return Zmodem; + if (Zmodem.__netcattyZmodemSendFastPathApplied) return Zmodem; + + try { + const proto = Zmodem.Session.Send.prototype; + const ORIGINAL_SEND_FILE_PART = proto._send_file_part; + if (typeof ORIGINAL_SEND_FILE_PART !== "function") return Zmodem; + + proto._send_file_part = function _send_file_part_fast(bytes_obj, final_packetend) { + const frameEndNum = SEND_FRAME_END_NUM[final_packetend]; + if ( + !frameEndNum || + !this._zencoder || + !this._zencoder.escapes_ctrl_chars() + ) { + return ORIGINAL_SEND_FILE_PART.apply(this, arguments); + } + + if (!this._sent_ZDATA) { + this._send_header("ZDATA", this._file_offset); + this._sent_ZDATA = true; + } + + const bytes_count = bytes_obj.length; + let obj_offset = 0; + + // Same 8192-byte split as the library (MAX_CHUNK_LENGTH in + // zsession.js); intermediate subpackets use no_end_no_ack and the + // final one uses the caller's frame end. + while (true) { + const chunk_size = + Math.min(obj_offset + 8192, bytes_count) - obj_offset; + const at_end = (chunk_size + obj_offset) >= bytes_count; + + const frame = buildSendSubpacketFast( + bytes_obj, + obj_offset, + chunk_size, + at_end ? frameEndNum : SEND_FRAME_END_NUM.no_end_no_ack, + ); + this._sender(frame); + // zdle.js keeps _lastcode on the encoder; replicate it for + // fidelity (it does not affect escape_ctrl_chars output). + this._zencoder._lastcode = frame[frame.length - 1]; + + this._file_offset += chunk_size; + obj_offset += chunk_size; + + if (obj_offset >= bytes_count) break; + } + }; + + Zmodem.__netcattyZmodemSendFastPathApplied = true; + } catch (err) { + console.error( + "[ZMODEM] send fast-path patch failed; using original sender:", + err && err.message ? err.message : err, + ); + } + + return Zmodem; +} + +module.exports = { + applyZmodemFastPath, + applyZmodemSendSessionFixes, + applyZmodemSendFastPath, + + // Exposed for unit tests. + _internals: { + parseSubpacketFast, + zdleDecode, + readDecodedBytes, + stripIgnoredBytesFast, + crc16Bytes, + crc32Bytes, + buildSendSubpacketFast, + sendSessionEscapes, + }, +}; diff --git a/electron/bridges/zmodemFastPath.test.cjs b/electron/bridges/zmodemFastPath.test.cjs new file mode 100644 index 000000000..bc857f51b --- /dev/null +++ b/electron/bridges/zmodemFastPath.test.cjs @@ -0,0 +1,770 @@ +"use strict"; + +/** + * Tests for the zmodem.js Buffer fast path (zmodemFastPath.cjs). + * + * Strategy: the fast parser must be byte-for-byte equivalent to the + * original zmodem.js subpacket parser, so most tests build wire bytes + * with the library's own encoder and compare the fast parser against + * Subpacket.parse16()/parse32(). Complete receive sessions + * (ZFILE → ZDATA → ZEOF → ZFIN → OO) are then driven through both the + * patched Session/Sentry and — via a fresh unpatched copy of the + * library — the original array pipeline, to prove the state machine + * wiring behaves identically. + */ + +const { test } = require("node:test"); +const assert = require("node:assert"); + +const ZmodemLib = require("zmodem.js"); +const { + applyZmodemFastPath, + applyZmodemSendFastPath, + applyZmodemSendSessionFixes, + _internals: { + parseSubpacketFast, + zdleDecode, + readDecodedBytes, + stripIgnoredBytesFast, + crc16Bytes, + crc32Bytes, + }, +} = require("./zmodemFastPath.cjs"); + +const Zmodem = applyZmodemFastPath(ZmodemLib); + +//---------------------------------------------------------------------- +// helpers +//---------------------------------------------------------------------- + +/** Deterministic PRNG so failures are reproducible. */ +function makeRng(seed) { + let s = seed >>> 0; + return function next() { + s = (s * 1664525 + 1013904223) >>> 0; + return s; + }; +} + +/** + * Random payload; every 4th byte is one of the interesting values + * (ZDLE, XON/XOFF, CR, DEL, '@', 0xC0, NUL, LF). + */ +function randomPayload(rng, len) { + const specials = [0x18, 0x11, 0x13, 0x0d, 0x7f, 0x40, 0xc0, 0x00, 0x0a]; + const out = []; + for (let i = 0; i < len; i++) { + const r = rng() & 0xff; + out.push((r & 3) === 0 ? specials[r % specials.length] : r); + } + return out; +} + +const FRAME_ENDS = ["end_no_ack", "no_end_no_ack", "no_end_ack", "end_ack"]; + +/** Encode one subpacket with the library's own encoder. */ +function buildWire(payload, { crc32 = false, escCtl = false, frameEnd = "no_end_no_ack" } = {}) { + const encoder = new Zmodem.ZDLE({ escape_ctrl_chars: escCtl }); + const subpacket = Zmodem.Subpacket.build(payload, frameEnd); + return Buffer.from( + crc32 ? subpacket.encode32(encoder) : subpacket.encode16(encoder), + ); +} + +/** Parse a wire Buffer with the original library parser (array pipeline). */ +function parseWithOriginal(wireBuffer, crcLen) { + const arr = Array.prototype.slice.call(wireBuffer); // what Sentry.consume() used to do + const subpacket = crcLen === 2 + ? Zmodem.Subpacket.parse16(arr) + : Zmodem.Subpacket.parse32(arr); + return { + subpacket, + consumed: wireBuffer.length - arr.length, + payload: subpacket ? Array.from(subpacket.get_payload()) : null, + }; +} + +/** Parse a wire Buffer with the fast parser. */ +function parseWithFast(wireBuffer, crcLen) { + const parsed = parseSubpacketFast(Zmodem, wireBuffer, crcLen); + return { + subpacket: parsed && parsed.subpacket, + consumed: parsed ? parsed.consumed : null, + payload: parsed ? Array.from(parsed.subpacket.get_payload()) : null, + }; +} + +function assertParsersAgree(wireBuffer, crcLen, label) { + const original = parseWithOriginal(wireBuffer, crcLen); + const fast = parseWithFast(wireBuffer, crcLen); + + assert.ok(original.subpacket, `${label}: original parser should parse`); + assert.ok(fast.subpacket, `${label}: fast parser should parse`); + assert.strictEqual(fast.consumed, original.consumed, `${label}: consumed`); + assert.deepStrictEqual(fast.payload, original.payload, `${label}: payload`); + assert.strictEqual( + fast.subpacket.frame_end(), + original.subpacket.frame_end(), + `${label}: frame_end`, + ); + assert.strictEqual( + fast.subpacket.ack_expected(), + original.subpacket.ack_expected(), + `${label}: ack_expected`, + ); +} + +/** + * Build the fake `sz` sender's wire for a complete download: + * ZFILE + file-info subpacket, ZDATA + file subpackets, ZEOF, ZFIN. + * Headers for ZFILE/ZDATA are binary (like lrzsz); ZEOF/ZFIN are hex. + * The trailing "OO" + prompt is fed separately by the caller. + */ +function buildSenderWire(lib, filePayload, { crc32 = false } = {}) { + const encoder = new lib.ZDLE({ escape_ctrl_chars: true }); + const enc = crc32 ? "encode32" : "encode16"; + const bin = crc32 ? "to_binary32" : "to_binary16"; + const chunks = []; + + const zfileInfo = Array.from("fastpath-test.bin").map((c) => c.charCodeAt(0)); + zfileInfo.push(0); + const rest = `${filePayload.length} 0 0 0`.split("").map((c) => c.charCodeAt(0)); + zfileInfo.push(...rest); + + chunks.push( + Buffer.from( + lib.Header.build("ZFILE")[bin](encoder).concat( + lib.Subpacket.build(zfileInfo, "end_ack")[enc](encoder), + ), + ), + ); + + const zdata = [lib.Header.build("ZDATA", 0)[bin](encoder)]; + const SUB = 1024; + for (let off = 0; off < filePayload.length; off += SUB) { + const slice = filePayload.slice(off, off + SUB); + const atEnd = off + SUB >= filePayload.length; + zdata.push( + lib.Subpacket.build(slice, atEnd ? "end_no_ack" : "no_end_no_ack")[enc](encoder), + ); + } + chunks.push(Buffer.concat(zdata.map(Buffer.from))); + + chunks.push(Buffer.from(lib.Header.build("ZEOF", filePayload.length).to_hex())); + chunks.push(Buffer.from(lib.Header.build("ZFIN").to_hex())); + + return Buffer.concat(chunks); +} + +/** Decode a sent header byte block into its header name. */ +function sentHeaderName(lib, bytes) { + const parsed = lib.Header.parse(Array.prototype.slice.call(bytes)); + return parsed && parsed[0] && parsed[0].NAME; +} + +/** + * Get a second, unpatched copy of the zmodem.js classes by evicting the + * library's modules from the require cache. The patched instance held by + * this test file keeps working (its classes are separate objects). + */ +function requireFreshUnpatchedZmodem() { + for (const key of Object.keys(require.cache)) { + if (key.includes(`${require("node:path").sep}zmodem.js`)) { + delete require.cache[key]; + } + } + return require("zmodem.js"); +} + +//---------------------------------------------------------------------- +// patch / idempotency +//---------------------------------------------------------------------- + +test("applyZmodemFastPath patches the shared prototypes and is idempotent", () => { + assert.strictEqual(Zmodem.__netcattyZmodemFastPathApplied, true); + assert.strictEqual(applyZmodemFastPath(Zmodem), Zmodem); + assert.strictEqual( + typeof Zmodem.Session.prototype._zmodem_fast_consume, + "function", + ); +}); + +test("fast receive detects an abort sequence spanning header and Buffer state", () => { + const headerPrefix = Buffer.from([0x2a, 0x18, 0x41, 0x18, 0x18, 0x18]); + const abortTail = Buffer.from([0x18, 0x18]); + const session = new Zmodem.Session.Receive(); + + // An incomplete binary header is retained in _input_buffer. The remaining + // two CAN bytes arrive in a later Buffer, so detection must cover both. + session.consume(headerPrefix); + assert.throws( + () => session.consume(abortTail), + (err) => err && err.type === "peer_aborted", + ); + assert.equal(session.aborted(), true); +}); + +test("fast receive checks abort bytes before post-ZFIN OO validation", () => { + const session = new Zmodem.Session.Receive(); + session._got_ZFIN = true; + + assert.throws( + () => session.consume(Buffer.from([0x18, 0x18, 0x18, 0x18, 0x18])), + (err) => err && err.type === "peer_aborted", + ); + assert.equal(session.aborted(), true); + assert.equal(session.has_ended(), true); +}); + +test("fast parsed subpackets keep typed payloads for zero-copy downloads", () => { + const wire = buildWire([0x00, 0x18, 0xff, 0x41], { frameEnd: "end_no_ack" }); + const parsed = parseSubpacketFast(Zmodem, wire, 2); + assert.ok(parsed); + assert.equal(parsed.subpacket.get_payload() instanceof Uint8Array, true); +}); + +test("kill switch disables both performance fast paths while retaining send-session fixes", () => { + const previous = process.env.NETCATTY_ZMODEM_FAST_PATH; + process.env.NETCATTY_ZMODEM_FAST_PATH = "0"; + + try { + const fresh = requireFreshUnpatchedZmodem(); + const originalSentryConsume = fresh.Sentry.prototype.consume; + const originalSendFilePart = fresh.Session.Send.prototype._send_file_part; + + applyZmodemFastPath(fresh); + applyZmodemSendSessionFixes(fresh); + applyZmodemSendFastPath(fresh); + + assert.equal(fresh.__netcattyZmodemFastPathApplied, undefined); + assert.equal(fresh.__netcattyZmodemSendFastPathApplied, undefined); + assert.equal(fresh.__netcattyZmodemSendSessionFixesApplied, true); + assert.strictEqual(fresh.Sentry.prototype.consume, originalSentryConsume); + assert.strictEqual(fresh.Session.Send.prototype._send_file_part, originalSendFilePart); + } finally { + if (previous === undefined) delete process.env.NETCATTY_ZMODEM_FAST_PATH; + else process.env.NETCATTY_ZMODEM_FAST_PATH = previous; + } +}); + +//---------------------------------------------------------------------- +// CRC +//---------------------------------------------------------------------- + +test("crc16Bytes matches the library's CRC.crc16", () => { + const rng = makeRng(7); + for (let i = 0; i < 20; i++) { + const payload = randomPayload(rng, (rng() & 0x1ff) + 1); + const frameEnd = 105; + const lib = Zmodem.CRC.crc16(payload.concat([frameEnd])); + const mine = crc16Bytes(payload, frameEnd); + assert.strictEqual((mine >> 8) & 0xff, lib[0], `crc16 case ${i} hi`); + assert.strictEqual(mine & 0xff, lib[1], `crc16 case ${i} lo`); + } +}); + +test("crc32Bytes matches the library's CRC.crc32", () => { + const rng = makeRng(11); + for (let i = 0; i < 20; i++) { + const payload = randomPayload(rng, (rng() & 0x1ff) + 1); + const frameEnd = 105; + const lib = Zmodem.CRC.crc32(payload.concat([frameEnd])); + const mine = crc32Bytes(payload, frameEnd); + assert.deepStrictEqual( + [mine & 0xff, (mine >> 8) & 0xff, (mine >> 16) & 0xff, (mine >> 24) & 0xff], + lib, + `crc32 case ${i}`, + ); + } +}); + +//---------------------------------------------------------------------- +// subpacket parser vs original +//---------------------------------------------------------------------- + +test("parseSubpacketFast round-trips CRC16 subpackets for every frame-end type", () => { + const rng = makeRng(42); + for (const frameEnd of FRAME_ENDS) { + for (const escCtl of [false, true]) { + const payload = randomPayload(rng, (rng() & 0x3ff) + 1); + const wire = buildWire(payload, { frameEnd, escCtl }); + assertParsersAgree(wire, 2, `${frameEnd}/escCtl=${escCtl}`); + } + } +}); + +test("parseSubpacketFast round-trips CRC32 subpackets for every frame-end type", () => { + const rng = makeRng(43); + for (const frameEnd of FRAME_ENDS) { + const payload = randomPayload(rng, (rng() & 0x3ff) + 1); + const wire = buildWire(payload, { frameEnd, crc32: true, escCtl: true }); + assertParsersAgree(wire, 4, `${frameEnd}/crc32`); + } +}); + +test("parseSubpacketFast handles empty and single-byte payloads", () => { + for (const payload of [[], [0x18], [0x00], [0x41]]) { + assertParsersAgree(buildWire(payload), 2, `payload=[${payload}]`); + assertParsersAgree(buildWire(payload, { crc32: true }), 4, `payload=[${payload}] crc32`); + } +}); + +test("fuzz: byte-at-a-time feeding parses identically to the original", () => { + const rng = makeRng(99); + for (let round = 0; round < 30; round++) { + const payload = randomPayload(rng, (rng() & 0xff) + 1); + const crcLen = round % 2 === 0 ? 2 : 4; + const wire = buildWire(payload, { + crc32: crcLen === 4, + escCtl: round % 3 === 0, + frameEnd: FRAME_ENDS[round % FRAME_ENDS.length], + }); + const original = parseWithOriginal(wire, crcLen); + + // Feed the wire one byte at a time; the parser must return null + // until the subpacket is complete, then parse it exactly once and + // consume every byte. + let acc = Buffer.alloc(0); + let parsed = null; + for (let i = 0; i < wire.length; i++) { + acc = Buffer.concat([acc, wire.subarray(i, i + 1)]); + const got = parseSubpacketFast(Zmodem, acc, crcLen); + if (got) { + assert.strictEqual(parsed, null, `round ${round}: parsed twice`); + parsed = got; + assert.strictEqual(got.consumed, acc.length, `round ${round}: partial consume`); + assert.deepStrictEqual( + Array.from(got.subpacket.get_payload()), + original.payload, + `round ${round}: payload`, + ); + assert.strictEqual( + got.subpacket.frame_end(), + original.subpacket.frame_end(), + `round ${round}: frame_end`, + ); + } + } + assert.ok(parsed, `round ${round}: never parsed`); + } +}); + +test("CRC corruption throws the same 'crc' error as the library", () => { + const rng = makeRng(5); + for (const crcLen of [2, 4]) { + for (const corruptWhat of ["payload", "crc"]) { + const payload = randomPayload(rng, 200); + const wire = buildWire(payload, { crc32: crcLen === 4 }); + + let at = -1; + if (corruptWhat === "payload") { + // flip the first byte that isn't a ZDLE escape prefix + for (let i = 0; i < wire.length / 2; i++) { + if (wire[i] !== 0x18) { + at = i; + break; + } + } + } else { + at = wire.length - 1; + } + assert.ok(at !== -1); + wire[at] ^= 0x40; + + let libErr = null; + let fastErr = null; + try { + parseWithOriginal(wire, crcLen); + } catch (err) { + libErr = err; + } + try { + parseWithFast(wire, crcLen); + } catch (err) { + fastErr = err; + } + + assert.ok(libErr, `${crcLen}/${corruptWhat}: library should throw`); + assert.ok(fastErr, `${crcLen}/${corruptWhat}: fast parser should throw`); + assert.strictEqual(fastErr.type, "crc", `${crcLen}/${corruptWhat}: type`); + assert.ok( + fastErr.message.startsWith("CRC check failed!"), + `${crcLen}/${corruptWhat}: message`, + ); + } + } +}); + +//---------------------------------------------------------------------- +// low-level helpers vs the library +//---------------------------------------------------------------------- + +test("stripIgnoredBytesFast matches ZMLIB.strip_ignored_bytes", () => { + const rng = makeRng(123); + for (let i = 0; i < 20; i++) { + // sprinkle XON/XOFF (and high-bit variants) into random data + const ignored = [0x11, 0x13, 0x91, 0x93]; + const input = randomPayload(rng, (rng() & 0x7f) + 1); + for (let k = 0; k < 5; k++) { + input[rng() % input.length] = ignored[rng() % ignored.length]; + } + + const lib = input.slice(0); + Zmodem.ZMLIB.strip_ignored_bytes(lib); + + assert.deepStrictEqual( + Array.from(stripIgnoredBytesFast(Buffer.from(input))), + lib, + `case ${i}`, + ); + } + + // zero-copy on clean input + const clean = Buffer.from([1, 2, 3]); + assert.strictEqual(stripIgnoredBytesFast(clean), clean); +}); + +test("zdleDecode matches ZDLE.decode", () => { + const encoder = new Zmodem.ZDLE({ escape_ctrl_chars: false }); + const rng = makeRng(321); + for (let i = 0; i < 20; i++) { + const payload = randomPayload(rng, (rng() & 0x3ff) + 1); + const encoded = encoder.encode(payload.slice(0)); // mutates its input + const lib = Zmodem.ZDLE.decode(encoded.slice(0)); + assert.deepStrictEqual(Array.from(zdleDecode(Buffer.from(encoded))), lib, `case ${i}`); + } +}); + +test("readDecodedBytes matches ZDLE.splice semantics", () => { + const rng = makeRng(555); + for (let i = 0; i < 40; i++) { + // build an encoded byte array: mostly plain bytes, occasionally a + // ZDLE escape pair (only second bytes >= 0x40 occur on the wire — + // conforming encoders never emit 0x00-0x3F there); sometimes a bare + // trailing ZDLE + const raw = randomPayload(rng, (rng() & 0x1f) + 1); + const encoded = []; + for (const b of raw) { + if ((b & 3) === 0) { + encoded.push(0x18, 0x40 | (b & 0x3f)); + } else if ((b & 3) === 1) { + encoded.push(0x18, 0xc0 | (b & 0x3f)); + } else { + encoded.push(b); + } + } + if ((i & 7) === 0) encoded.push(0x18); // bare trailing ZDLE + + const count = 1 + (rng() & 7); + + const libArr = encoded.slice(0); + const lib = Zmodem.ZDLE.splice(libArr, 0, count); + + const mine = readDecodedBytes(Buffer.from(encoded), 0, count); + + if (lib === undefined) { + assert.strictEqual(mine, null, `case ${i}: both incomplete`); + } else { + assert.ok(mine, `case ${i}: both complete`); + assert.deepStrictEqual(Array.from(mine.bytes), lib, `case ${i}: bytes`); + assert.strictEqual(mine.consumed, encoded.length - libArr.length, `case ${i}: consumed`); + } + } +}); + +//---------------------------------------------------------------------- +// complete receive sessions +//---------------------------------------------------------------------- + +/** + * Drive a complete download through a receive session of `lib` (patched + * or original), feeding `chunkSize` bytes at a time as Buffers (fast + * path) or plain arrays (original pipeline, like the Sentry used to do). + * + * Returns everything observable, so two runs can be compared: + * { sentNames, receivedHex, trailing, ended, aborted, firstError }. + */ +function runSessionScenario({ lib, filePayload, chunkSize, asBuffers, trailingFeed, crc32 = false }) { + const fileBuffer = Buffer.from(filePayload); + + const sent = []; + const inputPayloads = []; + const offers = []; + let acceptPromise = null; + + const session = new lib.Session.Receive(); + session.set_sender((octets) => sent.push(Buffer.from(octets))); + session.on("offer", (xfer) => { + offers.push(xfer); + acceptPromise = xfer.accept({ + on_input(payload) { + inputPayloads.push(Buffer.from(payload)); + }, + }); + }); + + session.start(); + + const feed = (bytes) => { + if (asBuffers) { + session.consume(Buffer.from(bytes)); + } else { + session.consume(Array.prototype.slice.call(bytes)); + } + }; + + let firstError = null; + const feedAll = (chunks) => { + for (const chunk of chunks) { + try { + feed(chunk); + } catch (err) { + if (!firstError) firstError = String((err && err.message) || err); + break; // the session is broken from here on + } + } + }; + + const wire = buildSenderWire(lib, filePayload, { crc32 }); + const wireChunks = []; + for (let i = 0; i < wire.length; i += chunkSize) { + wireChunks.push(wire.subarray(i, Math.min(i + chunkSize, wire.length))); + } + feedAll(wireChunks); + feedAll(trailingFeed.map((piece) => Buffer.from(piece))); + + return { + offerCount: offers.length, + offerName: offers.length ? offers[0].get_details().name : null, + offerSize: offers.length ? offers[0].get_details().size : null, + sentNames: sent.map((b) => sentHeaderName(lib, b)), + receivedHex: Buffer.concat(inputPayloads).toString("hex"), + trailing: session.has_ended() && !session.aborted() + ? Buffer.from(session.get_trailing_bytes()).toString() + : null, + ended: session.has_ended(), + aborted: session.aborted(), + firstError, + acceptResolved: Boolean(acceptPromise), + }; +} + +const TRAILING_SPLIT = ["O", "Oprompt$ "]; // second O and the prompt together +const TRAILING_WHOLE = ["OOprompt$ "]; + +test("patched receive session: full CRC16 download, 1 byte at a time", async () => { + const r = runSessionScenario({ + lib: Zmodem, + filePayload: randomPayload(makeRng(777), 3000), + chunkSize: 1, + asBuffers: true, + trailingFeed: TRAILING_SPLIT, + }); + assert.strictEqual(r.firstError, null); + assert.strictEqual(r.offerCount, 1); + assert.strictEqual(r.offerName, "fastpath-test.bin"); + assert.strictEqual(r.offerSize, 3000); + assert.strictEqual(r.ended, true); + assert.strictEqual(r.aborted, false); + assert.strictEqual(r.trailing, "prompt$ "); + assert.deepStrictEqual(r.sentNames, ["ZRINIT", "ZRPOS", "ZRINIT", "ZFIN"]); + assert.strictEqual(r.receivedHex, Buffer.from(randomPayload(makeRng(777), 3000)).toString("hex")); +}); + +test("patched receive session: full CRC32 download, OO in one chunk", () => { + const r = runSessionScenario({ + lib: Zmodem, + filePayload: randomPayload(makeRng(778), 2000), + chunkSize: 7, + asBuffers: true, + trailingFeed: TRAILING_WHOLE, + crc32: true, + }); + assert.strictEqual(r.firstError, null); + assert.strictEqual(r.ended, true); + assert.strictEqual(r.trailing, "prompt$ "); + assert.deepStrictEqual(r.sentNames, ["ZRINIT", "ZRPOS", "ZRINIT", "ZFIN"]); +}); + +test("fast receive keeps fragmented packets segmented until a frame is complete", () => { + const wire = buildWire(Array.from(Buffer.alloc(64 * 1024, 0x5a)), { frameEnd: "end_no_ack" }); + const session = new Zmodem.Session.Receive(); + session._last_header_crc = 16; + session._next_subpacket_handler = () => {}; + + const originalConcat = Buffer.concat; + let concatCount = 0; + Buffer.concat = function countedConcat() { + concatCount += 1; + return originalConcat.apply(this, arguments); + }; + + try { + for (const byte of wire) session.consume(Buffer.from([byte])); + } finally { + Buffer.concat = originalConcat; + } + + assert.ok(concatCount <= 4, `fragmented packet was flattened ${concatCount} times`); +}); + +test("fast receive keeps abort detection after compacting an incomplete CRC", () => { + const wire = buildWire([0x41], { frameEnd: "end_no_ack" }); + const marker = wire.lastIndexOf(Buffer.from([0x18, 0x68])); + assert.ok(marker >= 0); + const session = new Zmodem.Session.Receive(); + session._last_header_crc = 16; + session._next_subpacket_handler = () => {}; + + // Leave one CRC byte pending after the frame marker. This makes the fast + // path compact its fragmented chunks before the remaining CRC and CAN + // abort sequence arrive. + for (let i = 0; i < marker + 3; i++) { + session.consume(wire.subarray(i, i + 1)); + } + assert.throws( + () => session.consume(Buffer.from([wire[marker + 3], 0x18, 0x18, 0x18, 0x18, 0x18])), + (err) => err && err.type === "peer_aborted", + ); + assert.equal(session.aborted(), true); +}); + +test("pending ZSINIT restores its ZACK handler before waiting", async () => { + const lib = requireFreshUnpatchedZmodem(); + applyZmodemSendSessionFixes(lib); + const session = new lib.Session.Send( + lib.Header.build("ZRINIT", ["CANFDX", "CANOVIO"], 0), + ); + session._zsinit_pending = true; + session._next_header_handler = { ZRINIT() {} }; + + const pending = session._ensure_receiver_escapes_ctrl_chars(); + assert.equal(typeof session._next_header_handler.ZACK, "function"); + session._on_zsinit_ack(); + await pending; +}); + +test("patched send file parts are wire-equivalent to the original", () => { + const originalLib = requireFreshUnpatchedZmodem(); + const patchedLib = requireFreshUnpatchedZmodem(); + applyZmodemSendFastPath(patchedLib); + + const payload = Buffer.alloc(8192 * 2 + 1); + for (let i = 0; i < payload.length; i++) { + payload[i] = (i * 73 + 19) & 0xff; + } + + function sendFilePart(lib) { + const zrinit = lib.Header.build("ZRINIT", ["CANFDX", "CANOVIO"], 0); + const session = new lib.Session.Send(zrinit); + const wire = []; + session.set_sender((bytes) => wire.push(Buffer.from(bytes))); + session._stop_keepalive(); + session._sending_file = true; + session._send_file_part(new Uint8Array(payload), "end_no_ack"); + return wire; + } + + assert.deepStrictEqual(sendFilePart(patchedLib), sendFilePart(originalLib)); +}); + +test("session-level equivalence: fast path matches the original library pipeline", () => { + const originalLib = requireFreshUnpatchedZmodem(); + assert.notStrictEqual(originalLib.Session.prototype.consume, Zmodem.Session.prototype.consume); + + const filePayload = randomPayload(makeRng(1234), 2500); + const scenarios = [ + { chunkSize: 1, trailingFeed: TRAILING_SPLIT }, + { chunkSize: 7, trailingFeed: TRAILING_WHOLE }, + { chunkSize: 5, trailingFeed: TRAILING_SPLIT }, + // feed the trailing block as one chunk; exercises the trim-2 branch + { chunkSize: 3, trailingFeed: TRAILING_WHOLE }, + ]; + + for (const s of scenarios) { + const original = runSessionScenario({ + lib: originalLib, + filePayload, + chunkSize: s.chunkSize, + asBuffers: false, // arrays, like the Sentry's old conversion + trailingFeed: s.trailingFeed, + }); + const fast = runSessionScenario({ + lib: Zmodem, + filePayload, + chunkSize: s.chunkSize, + asBuffers: true, + trailingFeed: s.trailingFeed, + }); + + assert.deepStrictEqual(fast, original, `scenario ${JSON.stringify(s)}`); + assert.strictEqual(fast.firstError, null, `scenario ${JSON.stringify(s)}: no errors`); + } +}); + +test("patched Sentry: detection, fast consume, trailing bytes to terminal", () => { + const filePayload = randomPayload(makeRng(888), 2000); + const fileBuffer = Buffer.from(filePayload); + + const terminalBytes = []; + const sent = []; + let detection = null; + let zsession = null; + const inputPayloads = []; + + const sentry = new Zmodem.Sentry({ + to_terminal(bytes) { + terminalBytes.push(Buffer.from(bytes)); + }, + on_detect(det) { + detection = det; + }, + on_retract() {}, + sender(bytes) { + sent.push(Buffer.from(bytes)); + }, + }); + + // the remote's `sz` starts with a ZRQINIT; the sentry detects it and + // echoes the init bytes to the terminal, like Netcatty does today + const zrqinitWire = Buffer.from(Zmodem.Header.build("ZRQINIT").to_hex()); + sentry.consume(zrqinitWire); + assert.ok(detection, "detection fired"); + assert.strictEqual( + Buffer.concat(terminalBytes).compare(zrqinitWire), + 0, + "init bytes echoed to terminal", + ); + + zsession = detection.confirm(); + assert.strictEqual(zsession.type, "receive"); + + // mimic handleDownload(): start the session, accept the offer + zsession.start(); + zsession.on("offer", (xfer) => { + xfer.accept({ + on_input(payload) { + inputPayloads.push(Buffer.from(payload)); + }, + }); + }); + + const wire = buildSenderWire(Zmodem, filePayload); + for (let i = 0; i < wire.length; i += 5) { + sentry.consume(wire.subarray(i, Math.min(i + 5, wire.length))); + } + sentry.consume(Buffer.from("OOprompt$ ")); + + assert.strictEqual(zsession.has_ended(), true); + assert.strictEqual( + Buffer.concat(inputPayloads).compare(fileBuffer), + 0, + "file reassembled", + ); + + // after the session ends, the sentry routes trailing bytes to the + // terminal again + const allTerminal = Buffer.concat(terminalBytes).toString(); + assert.ok(allTerminal.includes("prompt$ "), "prompt went to terminal"); + assert.strictEqual(sentry.get_confirmed_session(), null); +}); diff --git a/electron/bridges/zmodemHelper.cjs b/electron/bridges/zmodemHelper.cjs index 3880d42ac..b5ef86187 100644 --- a/electron/bridges/zmodemHelper.cjs +++ b/electron/bridges/zmodemHelper.cjs @@ -9,7 +9,17 @@ * The renderer is only notified for progress display via lightweight IPC events. */ -const Zmodem = require("zmodem.js"); +// Apply the Buffer fast paths to zmodem.js receive/send hot paths, plus the +// Send-session robustness fixes for `rz` uploads (see zmodemFastPath.cjs). +// NETCATTY_ZMODEM_FAST_PATH=0 disables only the performance paths. +const { + applyZmodemFastPath, + applyZmodemSendSessionFixes, + applyZmodemSendFastPath, +} = require("./zmodemFastPath.cjs"); +const Zmodem = applyZmodemSendFastPath( + applyZmodemSendSessionFixes(applyZmodemFastPath(require("zmodem.js"))), +); const fs = require("node:fs"); const path = require("node:path"); @@ -22,13 +32,14 @@ function getElectron() { /** * Resolve per-file overwrite choices into an upload plan. Pure (no I/O): - * `resolveDecision(name)` is awaited only for files in `existingList`, in input - * order; `{ applyToRest: true }` reuses that action for the remaining conflicts. + * `resolveDecision(name, { signal })` is awaited only for files in + * `existingList`, in input order; `{ applyToRest: true }` reuses that action + * for the remaining conflicts. * Returns indices into the original `names` array so callers preserve per-file * identity even when two files share a basename. * Actions: 'overwrite' (rm remote then send), 'skip' (don't send), 'cancel' (abort all). */ -async function buildUploadPlan(names, existingList, resolveDecision) { +async function buildUploadPlan(names, existingList, resolveDecision, signal) { const existing = new Set(existingList); const offerIndices = []; const removeIndices = []; @@ -38,7 +49,11 @@ async function buildUploadPlan(names, existingList, resolveDecision) { if (!existing.has(name)) { offerIndices.push(idx); continue; } let action = bulkAction; if (!action) { - const decision = (await resolveDecision(name)) || { action: "skip" }; + throwIfZmodemCancelled(signal); + const decision = (await racePromiseWithAbortSignal( + resolveDecision(name, { signal }), + signal, + )) || { action: "skip" }; action = decision.action; if (decision.applyToRest && action !== "cancel") bulkAction = action; } @@ -137,8 +152,8 @@ function createZmodemSentry(opts) { } function rememberOutgoingEcho(octets) { + if (!octets?.length || octets.length > ECHO_MAX_BYTES) return; const buf = Buffer.from(octets); - if (!buf.length || buf.length > ECHO_MAX_BYTES) return; prunePendingEchoes(); pendingEchoes.push({ buf, @@ -377,6 +392,31 @@ function createZmodemSentry(opts) { ); } + /** + * After an ignorable send-session consume error, the bytes that followed + * the offending header in the same chunk stay buffered in the session's + * _input_buffer — e.g. the post-file ZRINIT that rz sends right behind a + * final ZRPOS ping. Re-feed them so the pending handshake (xfer.end() / + * close()) resolves instead of stalling until its timeout. Best-effort: + * anything still unparseable is dropped or picked up by the next consume. + */ + function refeedRemainingSessionBytes() { + const zsession = currentZSession; + if ( + !zsession || + !Array.isArray(zsession._input_buffer) || + !zsession._input_buffer.length + ) { + return; + } + const rest = Buffer.from(zsession._input_buffer.splice(0)); + try { + sentry.consume(rest); + } catch { + /* ignore — next regular consume retries whatever is left */ + } + } + const sentry = new Zmodem.Sentry({ to_terminal(octets) { @@ -390,7 +430,13 @@ function createZmodemSentry(opts) { sender(octets) { // ZMODEM protocol bytes – send raw to remote. rememberOutgoingEcho(octets); - const ok = writeToRemote(Buffer.from(octets)); + // Zero-copy view for the send fast path's Buffers; number arrays + // (non-data frames) still go through the Buffer.from() conversion. + const wireBuf = + octets instanceof Uint8Array + ? Buffer.from(octets.buffer, octets.byteOffset, octets.byteLength) + : Buffer.from(octets); + const ok = writeToRemote(wireBuf); // Track backpressure: if stream.write() returned false, the // kernel TCP buffer is full. The upload loop should pause. if (ok === false) { @@ -440,6 +486,7 @@ function createZmodemSentry(opts) { const transferSignal = transferAbortController.signal; const transferOpts = { ...opts, + signal: transferSignal, getDragDropUpload: () => dragDropUpload, takeDragDropUpload, clearDragDropUpload, @@ -535,14 +582,18 @@ function createZmodemSentry(opts) { // but the repeated header is harmless, so ignore it and keep waiting. if (isIgnorableSendKeepaliveError(errMsg)) { console.log(`[ZMODEM][${label}] Ignoring repeated pre-offer ZRINIT`); + refeedRemainingSessionBytes(); return; } // Some receivers emit a final ZRPOS ping right before they send the // post-file ZRINIT. If that ping is processed a beat late, zmodem.js // complains even though the transfer can continue normally. + // Re-feed buffered bytes (e.g. that ZRINIT) so the pending + // xfer.end() handshake resolves instead of stalling to its timeout. if (isIgnorableSendResumePingError(errMsg)) { console.log(`[ZMODEM][${label}] Ignoring late post-file ZRPOS`); + refeedRemainingSessionBytes(); return; } @@ -678,12 +729,23 @@ const UPLOAD_SESSION_CLOSE_TIMEOUT_MS = 15000; /** Max wait for a single transport drain after write() returned false. */ const UPLOAD_DRAIN_TIMEOUT_MS = 60000; /** Upload read/send chunk size. */ +// Keep the batch bounded so SSH backpressure is observed before another +// large group of 8192-byte wire subpackets enters the channel queue. const UPLOAD_CHUNK_SIZE = 64 * 1024; +/** Default interval between non-final upload progress IPC events. */ +const DEFAULT_UPLOAD_PROGRESS_THROTTLE_MS = 100; function resolveTimeoutMs(value, fallback) { return Number.isFinite(value) && value >= 0 ? value : fallback; } +function resolveProgressThrottleMs(value) { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= 0 + ? parsed + : DEFAULT_UPLOAD_PROGRESS_THROTTLE_MS; +} + /** * Race a promise against a timeout. If the promise doesn't settle within * `ms`, reject instead of hanging forever. This prevents zmodem.js internal @@ -717,6 +779,10 @@ function createZmodemCancelledError() { return err; } +function throwIfZmodemCancelled(signal) { + if (signal?.aborted) throw createZmodemCancelledError(); +} + /** * Wait until a Node writable stream reports it can accept more data. * Used after stream.write() returns false so ZMODEM uploads do not flood @@ -849,6 +915,9 @@ function waitForWritableDrain(stream, opts = {}) { */ function createZmodemUploadDrainWaiter(opts) { return async function waitForDrain() { + if (opts.signal?.aborted) { + throw createZmodemCancelledError(); + } if (!opts.getNeedsDrain()) return; if (typeof opts.waitForTransportDrain === "function") { @@ -873,7 +942,10 @@ function createZmodemUploadDrainWaiter(opts) { } opts.clearNeedsDrain(); - await new Promise((resolve) => setImmediate(resolve)); + await racePromiseWithAbortSignal( + new Promise((resolve) => setImmediate(resolve)), + opts.signal, + ); }; } @@ -933,7 +1005,11 @@ function resolveUploadFileEndTimeoutMs(opts) { async function waitForUploadHandshake(promise, ms, message, opts) { try { - return await withTimeout(promise, ms, message); + return await withTimeout( + racePromiseWithAbortSignal(promise, opts?.signal), + ms, + message, + ); } catch (err) { if (isZmodemTimeoutError(err)) { try { opts.onUploadTimeout?.(); } catch { /* ignore */ } @@ -1001,6 +1077,7 @@ async function handleUpload(zsession, opts) { } try { + throwIfZmodemCancelled(opts.signal); const fileStats = filePaths.map((fp) => fs.statSync(fp)); // Conflict handling (SSH only — callbacks absent on local/telnet/serial). @@ -1016,28 +1093,47 @@ async function handleUpload(zsession, opts) { // transfer fails before the replacement is committed. if (!isDragDropUpload && opts.probeReceiveConflicts && opts.requestOverwriteDecision) { try { - const probe = await opts.probeReceiveConflicts(allNames); + const probe = await racePromiseWithAbortSignal( + opts.probeReceiveConflicts(allNames, { signal: opts.signal }), + opts.signal, + ); + throwIfZmodemCancelled(opts.signal); if (probe && probe.dir && Array.isArray(probe.existing) && probe.existing.length > 0) { probeDir = probe.dir; probeModes = probe.modes || {}; - plan = await buildUploadPlan(allNames, probe.existing, opts.requestOverwriteDecision); + plan = await buildUploadPlan( + allNames, + probe.existing, + opts.requestOverwriteDecision, + opts.signal, + ); + throwIfZmodemCancelled(opts.signal); if (plan.aborted) { try { zsession.abort(); } catch { /* ignore */ } abortRemoteProcess(opts.writeToRemote); throw new Error("Transfer cancelled"); } if (plan.removeIndices.length && opts.removeRemoteFiles) { + throwIfZmodemCancelled(opts.signal); const base = probe.dir.replace(/\/+$/, ""); const targets = [...new Set(plan.removeIndices.map((i) => `${base}/${allNames[i]}`))]; try { - await opts.removeRemoteFiles(targets); + await racePromiseWithAbortSignal( + opts.removeRemoteFiles(targets, { signal: opts.signal }), + opts.signal, + ); + throwIfZmodemCancelled(opts.signal); } catch (err) { + if (isZmodemCancelledError(err)) throw err; console.warn("[ZMODEM] removeRemoteFiles failed; rz will skip:", err?.message || err); } } } } catch (err) { - if (err instanceof Error && err.message === "Transfer cancelled") throw err; + if ( + (err instanceof Error && err.message === "Transfer cancelled") || + isZmodemCancelledError(err) + ) throw err; console.warn("[ZMODEM] conflict probe failed; proceeding:", err?.message || err); } } @@ -1051,6 +1147,7 @@ async function handleUpload(zsession, opts) { const skippedOfferIndices = []; for (let i = 0; i < offers.length; i++) { + throwIfZmodemCancelled(opts.signal); const { originalIndex, filePath, stat, name } = offers[i]; opts.resetUploadBackpressure?.(); @@ -1067,13 +1164,23 @@ async function handleUpload(zsession, opts) { let bytesRemaining = 0; for (let j = i; j < offers.length; j++) bytesRemaining += offers[j].stat.size; - const xfer = await zsession.send_offer({ - name, - size: stat.size, - mtime: new Date(stat.mtimeMs), - files_remaining: offers.length - i, - bytes_remaining: bytesRemaining, - }); + // The offer handshake is the only upload step without a built-in + // deadline: if rz dies before answering ZFILE (crash, YMODEM fallback), + // send_offer() would park this loop forever and block all future + // ZMODEM transfers. Bound it like xfer.end() / zsession.close(). + throwIfZmodemCancelled(opts.signal); + const xfer = await waitForUploadHandshake( + zsession.send_offer({ + name, + size: stat.size, + mtime: new Date(stat.mtimeMs), + files_remaining: offers.length - i, + bytes_remaining: bytesRemaining, + }), + resolveUploadFileEndTimeoutMs(opts), + `Remote did not respond to the offer for ${name}. The upload was stopped so the terminal can recover.`, + opts, + ); if (!xfer) { // Receiver protected/skipped this file (e.g. rz without -y). @@ -1085,6 +1192,10 @@ async function handleUpload(zsession, opts) { const fd = fs.openSync(filePath, "r"); const buf = Buffer.alloc(UPLOAD_CHUNK_SIZE); let sent = 0; + // Progress IPC is throttled by default for every transport. Pass 0 only + // when a caller explicitly needs one event per chunk. + const progressThrottleMs = resolveProgressThrottleMs(opts.progressThrottleMs); + let lastProgressEmitAt = -Infinity; try { while (true) { @@ -1098,15 +1209,19 @@ async function handleUpload(zsession, opts) { xfer.send(new Uint8Array(buf.buffer, buf.byteOffset, bytesRead)); sent += bytesRead; - safeSend(getWebContents(), "netcatty:zmodem:progress", { - sessionId, - filename: name, - transferred: sent, - total: stat.size, - fileIndex: i, - fileCount: offers.length, - transferType: "upload", - }); + const now = Date.now(); + if (progressThrottleMs === 0 || now - lastProgressEmitAt >= progressThrottleMs) { + safeSend(getWebContents(), "netcatty:zmodem:progress", { + sessionId, + filename: name, + transferred: sent, + total: stat.size, + fileIndex: i, + fileCount: offers.length, + transferType: "upload", + }); + lastProgressEmitAt = now; + } // Wait for transport to drain if its buffer is full, then yield // so inbound ZMODEM control frames can be processed. @@ -1152,8 +1267,13 @@ async function handleUpload(zsession, opts) { const restores = buildModeRestores(probeDir, allNames, restoreIndices, probeModes); if (!restores.length) return; try { - await opts.restoreRemoteModes(restores); + await racePromiseWithAbortSignal( + opts.restoreRemoteModes(restores, { signal: opts.signal }), + opts.signal, + ); + throwIfZmodemCancelled(opts.signal); } catch (err) { + if (isZmodemCancelledError(err)) throw err; console.warn("[ZMODEM] restoreRemoteModes failed:", err?.message || err); } } @@ -1260,7 +1380,12 @@ async function handleDownload(zsession, opts) { xfer.accept({ on_input(payload) { if (writeAborted) return; - const chunk = Buffer.from(payload); + // payload is a number[] on the original zmodem.js pipeline and a + // Uint8Array on the fast path; take a zero-copy view in the fast + // case instead of copying every payload byte again. + const chunk = Array.isArray(payload) + ? Buffer.from(payload) + : Buffer.from(payload.buffer, payload.byteOffset, payload.byteLength); ws.write(chunk); received += chunk.length; diff --git a/electron/bridges/zmodemHelper.test.cjs b/electron/bridges/zmodemHelper.test.cjs index 6ab93ef03..6f85d4faf 100644 --- a/electron/bridges/zmodemHelper.test.cjs +++ b/electron/bridges/zmodemHelper.test.cjs @@ -334,6 +334,50 @@ test("handleUpload does not read the next chunk until transport backpressure cle fs.rmSync(tempDir, { recursive: true, force: true }); }); +test("handleUpload throttles progress IPC by default for every transport", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-progress-")); + const filePath = path.join(tempDir, "large-upload.bin"); + const size = UPLOAD_CHUNK_SIZE + 1; + fs.writeFileSync(filePath, Buffer.alloc(size, 0x5a)); + const progress = []; + const originalNow = Date.now; + Date.now = () => 1000; + + try { + const zsession = { + async send_offer() { + return { + send() {}, + async end() {}, + }; + }, + async close() {}, + }; + + await handleUpload(zsession, { + sessionId: "session-1", + getWebContents: () => ({ + isDestroyed: () => false, + send(channel, data) { + if (channel === "netcatty:zmodem:progress") progress.push(data); + }, + }), + takeDragDropUpload: () => ({ + filePaths: [filePath], + remoteNames: ["large-upload.bin"], + }), + }); + + assert.deepEqual( + progress.map((event) => [event.transferred, Boolean(event.finalizing)]), + [[0, false], [UPLOAD_CHUNK_SIZE, false], [size, true]], + ); + } finally { + Date.now = originalNow; + fs.rmSync(tempDir, { recursive: true, force: true }); + } +}); + test("handleUpload progress follows a display rebind during transfer", async () => { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-rebind-")); const filePath = path.join(tempDir, "upload.txt"); @@ -547,6 +591,166 @@ test("handleUpload does not run timeout recovery when the remote rejects the fin fs.rmSync(tempDir, { recursive: true, force: true }); }); +test("handleUpload times out when the remote never answers the offer", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-")); + const filePath = path.join(tempDir, "upload.txt"); + fs.writeFileSync(filePath, "payload"); + const writes = []; + let timeoutNotified = false; + + // rz died before answering ZFILE: send_offer never resolves. The upload + // loop must not park here forever — bound it like xfer.end()/close(). + const zsession = { + send_offer: () => new Promise(() => {}), + async abort() {}, + async close() {}, + }; + + await assert.rejects( + handleUpload(zsession, { + sessionId: "session-1", + getWebContents: () => null, + writeToRemote: (buf) => { + writes.push(Buffer.from(buf)); + return true; + }, + takeDragDropUpload: () => ({ + filePaths: [filePath], + remoteNames: ["upload.txt"], + }), + uploadFileEndTimeoutMs: 50, + onUploadTimeout: () => { + timeoutNotified = true; + }, + }), + /Remote did not respond to the offer for upload\.txt/, + ); + + assert.equal(timeoutNotified, true); + assert.ok(writes.length > 0, "CAN abort bytes were sent to the remote"); + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + +test("handleUpload stops an unresolved offer promptly when cancelled", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-")); + const filePath = path.join(tempDir, "upload.txt"); + fs.writeFileSync(filePath, "payload"); + const controller = new AbortController(); + + const upload = handleUpload( + { + send_offer: () => new Promise(() => {}), + async close() {}, + }, + { + sessionId: "session-1", + signal: controller.signal, + getWebContents: () => null, + takeDragDropUpload: () => ({ + filePaths: [filePath], + remoteNames: ["upload.txt"], + }), + }, + ); + + await new Promise((resolve) => setImmediate(resolve)); + controller.abort(); + await assert.rejects( + () => upload, + (err) => err && err.code === "NETCATTY_ZMODEM_CANCELLED", + ); + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + +test("handleUpload does not delete an overwrite target after cancellation", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-cancel-overwrite-")); + const filePath = path.join(tempDir, "upload.txt"); + fs.writeFileSync(filePath, "payload"); + const controller = new AbortController(); + let removeCalls = 0; + let sendOfferCalls = 0; + let releaseDecision; + + const upload = handleUpload( + { + send_offer() { + sendOfferCalls += 1; + return Promise.reject(new Error("send_offer should not run after cancellation")); + }, + async close() {}, + }, + { + sessionId: "session-1", + signal: controller.signal, + getWebContents: () => null, + selectUploadFiles: async () => ({ canceled: false, filePaths: [filePath] }), + probeReceiveConflicts: async () => ({ + dir: "/home/u", + existing: ["upload.txt"], + modes: { "upload.txt": "644" }, + }), + requestOverwriteDecision: () => new Promise((resolve) => { + releaseDecision = resolve; + }), + removeRemoteFiles: async () => { + removeCalls += 1; + }, + }, + ); + + while (!releaseDecision) await new Promise((resolve) => setImmediate(resolve)); + controller.abort(); + releaseDecision({ action: "overwrite", applyToRest: false }); + await assert.rejects( + () => upload, + (err) => err && err.code === "NETCATTY_ZMODEM_CANCELLED", + ); + assert.equal(removeCalls, 0); + assert.equal(sendOfferCalls, 0); + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + +test("handleUpload aborts an in-flight remote overwrite cleanup", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-cancel-remove-")); + const filePath = path.join(tempDir, "upload.txt"); + fs.writeFileSync(filePath, "payload"); + const controller = new AbortController(); + let releaseRemove; + let removeSignal; + + const upload = handleUpload( + { + async send_offer() { + throw new Error("send_offer should not run after cancellation"); + }, + async close() {}, + }, + { + sessionId: "session-1", + signal: controller.signal, + getWebContents: () => null, + selectUploadFiles: async () => ({ canceled: false, filePaths: [filePath] }), + probeReceiveConflicts: async () => ({ dir: "/home/u", existing: ["upload.txt"] }), + requestOverwriteDecision: async () => ({ action: "overwrite", applyToRest: false }), + removeRemoteFiles: (_paths, { signal } = {}) => { + removeSignal = signal; + return new Promise((resolve) => { releaseRemove = resolve; }); + }, + }, + ); + + while (!releaseRemove) await new Promise((resolve) => setImmediate(resolve)); + controller.abort(); + await assert.rejects( + () => upload, + (err) => err && err.code === "NETCATTY_ZMODEM_CANCELLED", + ); + assert.equal(removeSignal, controller.signal); + assert.equal(removeSignal.aborted, true); + releaseRemove(); + fs.rmSync(tempDir, { recursive: true, force: true }); +}); + test("handleUpload allows a longer final wait after upload backpressure", async () => { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "netcatty-zmodem-")); const filePath = path.join(tempDir, "upload.txt");