feat(zmodem): accelerate upload and download transfers

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;

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();

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(() => {});
     }
 

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,
             });