(fix): stop the sender closing connection without the receiver finishing

Author: collapsinghierarchy

src/commands/send.js

@@ -20,6 +20,8 @@ import { formatProgressLine } from "../util/progress.js";
 import { closeTransport, waitForPeerClose } from "../util/rtc.js";
 
 const CHUNK = 64 * 1024;
+const DEFAULT_FIN_ACK_TIMEOUT_MS = 45000;
+const DEFAULT_PEER_CLOSE_TIMEOUT_MS = 5000;
 
 function deriveSendName(srcPath, opts) {
   if (opts?.name) return String(opts.name);
@@ -120,7 +122,7 @@ export async function run(paths, opts, ctx = {}) {
   } catch (err) {
     getLogger().debug("send: failed to announce telemetry ice-connected", err);
   }
-  getLogger().debug("send: Zipping the contents...");
+  getLogger().debug("send: Packing the contents...");
   if (tarPromise) {
     const { pack, totalSizeTar } = await tarPromise;
     sourceStream = pack;
@@ -167,48 +169,42 @@ export async function run(paths, opts, ctx = {}) {
 
 
     // Wait for FIN/OK from receiver or peer close
-    const waitForFinAck = () =>
-      new Promise((resolve) => {
-        let settled = false;
-        const offMsg = rtc.onMessage?.((m) => {
-          if (settled) return;
-          try {
-            let msg = m;
-            if (typeof m === "string") msg = JSON.parse(m);
-            if (m instanceof Uint8Array || ArrayBuffer.isView(m)) {
-              try {
-                msg = JSON.parse(new TextDecoder().decode(m));
-              } catch {}
-            }
-            const fin = msg ? parseStreamFin(msg) : null;
-            if (fin && fin.sessionId === sessionId && fin.ok === true) {
-              settled = true;
-              offMsg?.();
-              offClose?.();
-              resolve();
-            }
-          } catch {}
-        });
-        const offClose = rtc.onClose?.(() => {
-          if (settled) return;
-          settled = true;
-          offMsg?.();
-          resolve();
-        });
-        setTimeout(() => {
-          if (!settled) {
-            settled = true;
-            offMsg?.();
-            offClose?.();
-            resolve();
-          }
-        }, 3000);
-      });
+    const finAckTimeoutMs = normalizeTimeout(
+      opts?.finAckTimeoutMs,
+      DEFAULT_FIN_ACK_TIMEOUT_MS,
+    );
+    const peerCloseTimeoutMs = normalizeTimeout(
+      opts?.peerCloseTimeoutMs,
+      DEFAULT_PEER_CLOSE_TIMEOUT_MS,
+    );
 
-    await waitForFinAck();
-    // Drain bufferedAmount so wrtc doesn’t die on teardown
-    // Let the peer close first to avoid races
-    await waitForPeerClose(rtc, 1500);
+    const finAckResult = await waitForFinAck({
+      rtc,
+      sessionId,
+      timeoutMs: finAckTimeoutMs,
+    });
+    if (finAckResult === "timeout" && Number.isFinite(finAckTimeoutMs) && finAckTimeoutMs > 0) {
+      getLogger().info(
+        `send: timed out waiting for receiver FIN acknowledgement after ${finAckTimeoutMs} ms`,
+      );
+    } else if (finAckResult === "ack") {
+      getLogger().debug("send: received receiver FIN acknowledgement");
+    } else if (finAckResult === "peer-close") {
+      getLogger().debug("send: data channel closed before FIN acknowledgement was observed");
+    }
+
+    if (finAckResult !== "peer-close") {
+      const peerCloseResult = await waitForPeerClose(rtc, peerCloseTimeoutMs);
+      if (
+        peerCloseResult === "timeout" &&
+        Number.isFinite(peerCloseTimeoutMs) &&
+        peerCloseTimeoutMs > 0
+      ) {
+        getLogger().info(
+          `send: timed out waiting for peer connection to close after ${peerCloseTimeoutMs} ms`,
+        );
+      }
+    }
     process.stderr.write("\nDone • " + humanBytes(totalBytes) + "\n");
     // --- Silent workaround on success
   } finally {
@@ -225,6 +221,62 @@ export async function run(paths, opts, ctx = {}) {
 /* ------------------------------- utilities ------------------------------- */
 
 const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+function waitForFinAck({ rtc, sessionId, timeoutMs }) {
+  return new Promise((resolve) => {
+    let settled = false;
+    let timer = null;
+    let offMsg;
+    let offClose;
+    const finish = (reason) => {
+      if (settled) return;
+      settled = true;
+      try {
+        offMsg?.();
+      } catch {}
+      try {
+        offClose?.();
+      } catch {}
+      if (timer) {
+        clearTimeout(timer);
+      }
+      resolve(reason);
+    };
+    offMsg = rtc.onMessage?.((m) => {
+      if (settled) return;
+      try {
+        let msg = m;
+        if (typeof m === "string") msg = JSON.parse(m);
+        if (m instanceof Uint8Array || ArrayBuffer.isView(m)) {
+          try {
+            msg = JSON.parse(new TextDecoder().decode(m));
+          } catch {}
+        }
+        const fin = msg ? parseStreamFin(msg) : null;
+        if (fin && fin.sessionId === sessionId && fin.ok === true) {
+          finish("ack");
+        }
+      } catch {}
+    });
+    offClose = rtc.onClose?.(() => {
+      finish("peer-close");
+    });
+    timer = Number.isFinite(timeoutMs) && timeoutMs > 0
+      ? setTimeout(() => {
+          finish("timeout");
+        }, timeoutMs)
+      : null;
+  });
+}
+
+function normalizeTimeout(value, fallback) {
+  if (value == null) return fallback;
+  const ms = Number(value);
+  if (!Number.isFinite(ms)) return fallback;
+  if (ms < 0) return fallback;
+  return ms;
+}
+
 async function isRegularFile(p) {
   try {
     const st = await fsp.stat(p);

src/util/rtc.js

@@ -3,19 +3,27 @@ import { flush, forceCloseNoFlush, scrubTransport } from "@noisytransfer/transpo
 export function waitForPeerClose(rtc, ms = 1500) {
   return new Promise((resolve) => {
     let done = false;
-    const off = rtc.onClose?.(() => {
-      if (!done) {
-        done = true;
-        resolve();
-      }
-    });
-    setTimeout(() => {
-      if (!done) {
-        done = true;
+    let timer = null;
+    let off;
+    const finish = (reason) => {
+      if (done) return;
+      done = true;
+      try {
         off?.();
-        resolve();
+         } catch {}
+      if (timer) {
+        clearTimeout(timer);
       }
-    }, ms);
+       resolve(reason);
+    };
+    off = rtc.onClose?.(() => {
+      finish("closed");
+    });
+    timer = Number.isFinite(ms) && ms > 0
+      ? setTimeout(() => {
+          finish("timeout");
+        }, ms)
+      : null;
   });
 }