(fix) fix race condition when a sender was receiving messages uninitialized from receiver

Author: collapsinghierarchy

src/commands/recv.js

@@ -44,6 +44,26 @@ export async function run(outDir, opts, ctx = {}) {
   });
   await signal.waitOpen?.(100000);
 
+  const waitForSenderReady = ({ fallbackMs = 2_000 } = {}) =>
+    new Promise((resolve) => {
+      let settled = false;
+      const finish = (ready) => {
+        if (settled) return;
+        settled = true;
+        clearTimeout(timer);
+        try {
+          off?.();
+        } catch {}
+        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);
+    });
+
   let totalBytes = 0;
   let written = 0;
   let startedAt = 0;
@@ -92,7 +112,13 @@ export async function run(outDir, opts, ctx = {}) {
   } else {
     mode = "dtls";
   }
-  await sleep(2_000); // give the sender some headstart after seeing room_full
+  const sawReadySignal = await waitForSenderReady();
+  if (sawReadySignal) {
+    getLogger().debug("recv: sender_ready signal received; skipping headstart delay");
+  } else {
+    getLogger().debug("recv: sender_ready signal missing; continuing after fallback delay");
+  }
+
   getLogger().debug("recv: selected mode =", mode);
 
   try {
@@ -172,10 +198,24 @@ function makeSniffingSink({ outToStdout, outPath, appID, overwrite, onStart, onP
     // baseName is already sanitized to a leaf; join is fine and clearer.
     let p = path.join(targetDir, baseName);
     if (!overwrite && fs.existsSync(p)) {
-      const stem = path.basename(baseName, path.extname(baseName));
       const ext = path.extname(baseName);
+      const stem = path.basename(baseName, ext);
       let i = 1;
-      while (fs.existsSync(p)) p = path.join(targetDir, `${stem}-${i++}${ext}`);
+      let dedup;
+      do {
+        dedup = path.join(targetDir, `${stem}-${i++}${ext}`);
+      } while (fs.existsSync(dedup));
+      try {
+        fs.renameSync(p, dedup);
+        getLogger().debug(
+          `resolveTargetPath: renamed existing ${baseName} -> ${path.basename(dedup)}`,
+        );
+      } catch (err) {
+        getLogger().info(
+          `resolveTargetPath: failed to rename existing ${baseName}: ${err?.message || err}`,
+        );
+        p = dedup;
+      }
     }
     return p;
   }

src/commands/send.js

@@ -114,15 +114,27 @@ export async function run(paths, opts, ctx = {}) {
   // Attach low-level logger BEFORE wiring auth/stream
   const detachTap = attachRawLogger(rtc, { label: "low" });
   getLogger().debug("send: RTC connected");
-
   try {
+    signal.send?.({ type: "telemetry", event: "ice-connected", sessionId });
+    getLogger().debug("send: announced telemetry ice-connected");
+  } catch (err) {
+    getLogger().debug("send: failed to announce telemetry ice-connected", err);
+  }
 
+  try {
     if (tarPromise) {
       const { pack, totalSizeTar } = await tarPromise;
       sourceStream = pack;
       totalBytes = totalSizeTar;
     }
 
+    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})`);
     }

test/integration/cli_auto_mode_e2e.test.mjs

@@ -32,19 +32,43 @@ const WAIT_FILE_MS = 12_000;
 async function waitForNewOrUpdatedFile(
   dir,
   prev = new Map(),
-  { timeoutMs = WAIT_FILE_MS, stableMs = 300 } = {},
+  { timeoutMs = WAIT_FILE_MS, stableMs = 300, preferredBasename } = {},
 ) {
   const deadline = Date.now() + timeoutMs;
   let lastName = null,
     lastSize = -1,
     lastT = 0;
+    const preferredName = preferredBasename ? String(preferredBasename) : null;
+    const preferredPath = preferredName ? path.join(dir, preferredName) : null;
+    let preferredSize = -1;
+    let preferredT = 0;
+    let preferredActive = false;
   while (Date.now() < deadline) {
     let names;
     try {
       names = await fsp.readdir(dir);
     } catch {
       names = [];
     }
+    if (preferredPath) {
+      let st;
+      try {
+        st = await fsp.stat(preferredPath);
+      } catch {}
+      if (st) {
+        const prevInfo = prev.get(preferredName);
+        const changed = !prevInfo || st.mtimeMs !== prevInfo.mtimeMs || st.size !== prevInfo.size;
+        if (changed) {
+          if (preferredActive && st.size === preferredSize) {
+            if (Date.now() - preferredT >= stableMs) return preferredPath;
+          } else {
+            preferredActive = true;
+            preferredSize = st.size;
+            preferredT = Date.now();
+          }
+        }
+      }
+    }
     for (const name of names) {
       const p = path.join(dir, name);
       let st;
@@ -126,6 +150,7 @@ async function runCase({
   assert.equal(rcode, 0, "receiver should exit cleanly (code 0)");
   const outPath = await waitForNewOrUpdatedFile(outDir, prev, {
     timeoutMs: 2000, // grace period after clean exit
+    preferredBasename: expectedBasename,
   });
 
   if (expectedBasename) {