additional fixes for synchronizing the receiver with sender

Author: collapsinghierarchy

src/commands/recv.js

@@ -44,24 +44,39 @@ export async function run(outDir, opts, ctx = {}) {
   });
   await signal.waitOpen?.(100000);
 
+  let sawSenderReadySignal = false;
+  let notifySenderReady = null;
+  const offSenderReady = signal.onMessage?.((msg) => {
+    if (!msg || msg.type !== "sender_ready") return;
+    if (msg.sessionId && sessionId && msg.sessionId !== sessionId) return;
+    sawSenderReadySignal = true;
+    const cb = notifySenderReady;
+    notifySenderReady = null;
+    try {
+      cb?.();
+    } catch {}
+  });
+
   const waitForSenderReady = ({ fallbackMs = 2_000 } = {}) =>
     new Promise((resolve) => {
+      if (sawSenderReadySignal) {
+        resolve(true);
+        return;
+      }
       let settled = false;
+      let timer = null;
       const finish = (ready) => {
         if (settled) return;
         settled = true;
         clearTimeout(timer);
-        try {
-          off?.();
-        } catch {}
+        if (notifySenderReady === onReady) {
+          notifySenderReady = null;
+        }
         resolve(ready);
       };
-      const off = signal.onMessage?.((msg) => {
-        if (!msg || msg.type !== "sender_ready") return;
-        if (msg.sessionId && sessionId && msg.sessionId !== sessionId) return;
-        finish(true);
-      });
-      const timer = setTimeout(() => finish(false), fallbackMs);
+      const onReady = () => finish(true);
+      notifySenderReady = onReady;
+      timer = setTimeout(() => finish(false), fallbackMs);
     });
 
   let totalBytes = 0;
@@ -161,6 +176,9 @@ export async function run(outDir, opts, ctx = {}) {
     process.stderr.write(`\nError: ${msg}\n`);
     throw e;
   } finally {
+    try {
+      offSenderReady?.();
+    } catch {}
     await closeTransport(rtc, {
       signal,
       waitForCloseMs: 0,

src/commands/send.js

@@ -121,24 +121,23 @@ export async function run(paths, opts, ctx = {}) {
     getLogger().debug("send: failed to announce telemetry ice-connected", err);
   }
 
+  if (tarPromise) {
+    const { pack, totalSizeTar } = await tarPromise;
+    sourceStream = pack;
+    totalBytes = totalSizeTar;
+  }
   try {
-    if (tarPromise) {
-      const { pack, totalSizeTar } = await tarPromise;
-      sourceStream = pack;
-      totalBytes = totalSizeTar;
-    }
+    signal.send?.({ type: "sender_ready", sessionId });
+    getLogger().debug("send: announced sender_ready");
+  } catch (err) {
+    getLogger().debug("send: failed to announce sender_ready", err);
+  }
 
-    try {
-      signal.send?.({ type: "sender_ready", sessionId });
-      getLogger().debug("send: announced sender_ready");
-    } catch (err) {
-      getLogger().debug("send: failed to announce sender_ready", err);
-    }
+  if (!Number.isInteger(totalBytes) || totalBytes <= 0) {
+    throw new Error(`internal: computed totalBytes invalid (${totalBytes})`);
+  }
 
-    if (!Number.isInteger(totalBytes) || totalBytes <= 0) {
-      throw new Error(`internal: computed totalBytes invalid (${totalBytes})`);
-    }
-    
+  try {    
     if (opts.pq) {
       //const rtcAuth = wrapAuthDC(rtc, { sessionId, label: "pq-auth-sender" });
       await pqSend(rtc, {

test/integration/cli_auto_mode_e2e.test.mjs

@@ -212,7 +212,15 @@ test(
   async () => {
     // Reuse the same directory across runs so we can observe conflicts/overwrite.
     const outDir = await fsp.mkdtemp(path.join(os.tmpdir(), "ntcli-bundle-"));
-    const paths = [path.join(ROOT, "README.md"), path.join(ROOT, "LICENSE")];
+    const srcTmp = await fsp.mkdtemp(path.join(os.tmpdir(), "ntcli-bundle-src-"));
+    const paths = [
+      path.join(srcTmp, "one.txt"),
+      path.join(srcTmp, "two.txt"),
+    ];
+    await Promise.all([
+      fsp.writeFile(paths[0], "hello bundle one\n".repeat(8)),
+      fsp.writeFile(paths[1], "hello bundle two\n".repeat(6)),
+    ]);
     const name = "bundle.tar";
     const tarPath = path.join(outDir, name);