chore(hotfix): apply DTLS SCTP backpressure to the datachannel

Author: collapsinghierarchy

src/transfer/default.js

@@ -15,6 +15,22 @@ import {
 import { getLogger } from "../util/logger.js";
 import { confirmSAS } from "../core/sas-confirm.js";
 
+// --- Back-pressure (congestion-cooperation) settings ---
+// Pause sending whenever the underlying RTC DataChannel queue grows too large.
+// Tune via env NT_SEND_HIGH_WATER (bytes). Default: 1 MiB.
+const SEND_HIGH_WATER = Number(process.env.NT_SEND_HIGH_WATER || (1 * 1024 * 1024));
+
+async function applyBackpressure(rtc) {
+  try {
+    if (typeof rtc?.bufferedAmount === "number" && typeof rtc?.flush === "function") {
+      if (rtc.bufferedAmount > SEND_HIGH_WATER) {
+        await rtc.flush();
+      }
+    }
+  } catch {}
+}
+
+
 /**
  * Try to read DTLS fingerprints for a short window; return null if not ready.
  * @returns {{local:{alg:string,bytes:Uint8Array},remote:{alg:string,bytes:Uint8Array}}|null}
@@ -157,6 +173,7 @@ export async function defaultSend(
     throw new Error("defaultSend: totalBytes must be a positive integer");
   const init = packStreamInit({ sessionId, totalBytes: total });
   rtc.send(init);
+  await applyBackpressure(rtc);
 
   // 1a) Optional filename: embed as first data frame (encrypted by DTLS)
   // [ 4 bytes magic = 'N' 'T' 'M' '1' ] [ 1 byte nameLen ] [ name UTF-8 bytes ]
@@ -165,6 +182,7 @@ export async function defaultSend(
     try {
       const header = buildMetaHeader(name);
       rtc.send(packStreamData({ sessionId, seq, chunk: header }));
+      await applyBackpressure(rtc);
       seq += 1;
     } catch {}
   }
@@ -175,6 +193,7 @@ export async function defaultSend(
     const u8 = chunk instanceof Uint8Array ? chunk : Buffer.from(chunk);
     if (!u8.byteLength) continue;
     rtc.send(packStreamData({ sessionId, seq, chunk: u8 }));
+    await applyBackpressure(rtc);
     sent += u8.byteLength;
     seq += 1;
     onProgress?.(Math.min(sent, total), total);
@@ -183,6 +202,7 @@ export async function defaultSend(
   // 3) FIN with ok=true if sizes match, else ok=false
   const ok = sent === total;
   rtc.send(packStreamFin({ sessionId, ok }));
+  await applyBackpressure(rtc);
 
   // 4) Flush if supported (mirrors tests' "flush" semantics)
   if (typeof rtc.flush === "function") {