From 8b7776ea25ca95394e73f5e730b25c2614cfbaa9 Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 15:23:38 +0200 Subject: [PATCH 01/10] Replace simple-peer with native WebRTC and fix max message size (#1246) - Replace unmaintained simple-peer with standard RTCPeerConnection and RTCDataChannel - Query pc.sctp.maxMessageSize and clamp to 256 KiB cross-browser safety limit to prevent Chrome 'Trying to send message larger than max-message-size' error when peering with Firefox - Widen chunk ID header from uint8 to uint32BE, removing the 255 chunk size limit for large payloads such as GPT weights - Implement bufferedAmountLow event backpressure and candidate queuing - Add unit tests verifying 10MB payload byte-for-byte reassembly and >255 chunk transfer Fixes #1246 --- discojs/package.json | 3 +- discojs/src/client/decentralized/peer.spec.ts | 44 ++ discojs/src/client/decentralized/peer.ts | 490 +++++++++++------- discojs/tsconfig.lib.json | 1 + discojs/tsconfig.vitest.json | 1 + pnpm-lock.yaml | 43 +- 6 files changed, 367 insertions(+), 215 deletions(-) diff --git a/discojs/package.json b/discojs/package.json index cca3fc989..a2cc8002d 100644 --- a/discojs/package.json +++ b/discojs/package.json @@ -102,11 +102,10 @@ "debug": "catalog:", "fast-check": "3.23.2", "immutable": "catalog:", - "simple-peer": "9.11.1", "zod": "catalog:" }, "devDependencies": { - "@types/simple-peer": "9.11.9", + "@types/node": "22.19.15", "nodemon": "3.1.14", "vitest": "catalog:" } diff --git a/discojs/src/client/decentralized/peer.spec.ts b/discojs/src/client/decentralized/peer.spec.ts index 4608b4f8a..67916c5b4 100644 --- a/discojs/src/client/decentralized/peer.spec.ts +++ b/discojs/src/client/decentralized/peer.spec.ts @@ -78,4 +78,48 @@ describe("peer", () => { assert.isTrue(received); }); + + it("transfers GPT-sized payload with byte-for-byte reassembly", async () => { + // 10 MB buffer simulating GPT weights + const size = 10 * 1024 * 1024; + const message = Buffer.alloc(size); + // Fill with non-trivial pattern to thoroughly verify byte ordering + for (let i = 0; i < size; i += 4) { + message.writeUInt32LE((i * 1103515245 + 12345) >>> 0, i); + } + + const receivedPromise = new Promise((resolve) => { + peer2.on("data", (data) => resolve(data)); + }); + + peer1.send(message); + await peer1.drain(); + const received = await receivedPromise; + + assert.strictEqual(received.length, message.length); + assert.strictEqual(Buffer.compare(received, message), 0); + }); + + it("handles payloads with more than 255 chunks", async () => { + // Override maxChunkSize to small size (1024 bytes) to test >255 chunk count boundary + Object.defineProperty(peer1, "maxChunkSize", { get: () => 1024 }); + + // 300 chunks + const size = 300 * 1024; + const message = Buffer.alloc(size); + for (let i = 0; i < size; i += 4) { + message.writeUInt32LE((i * 1664525 + 1013904223) >>> 0, i); + } + + const receivedPromise = new Promise((resolve) => { + peer2.on("data", (data) => resolve(data)); + }); + + peer1.send(message); + await peer1.drain(); + const received = await receivedPromise; + + assert.strictEqual(received.length, message.length); + assert.strictEqual(Buffer.compare(received, message), 0); + }); }); diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index ecedbb3b8..7425ac50f 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -1,21 +1,20 @@ import { List, Map, Range, Seq, Set } from "immutable"; import wrtc from "@epfml/isomorphic-wrtc"; -import SimplePeer from "simple-peer"; import type { NodeID } from "#client/types"; +import { EventEmitter } from "#utils/event_emitter"; type MessageID = number; type ChunkID = number; -// message id + (chunk counter == 0) + chunk count -const FIRST_HEADER_SIZE = 2 + 1 + 1; +// message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) +const FIRST_HEADER_SIZE = 2 + 4 + 4; -// message id + chunk counter -const HEADER_SIZE = 2 + 1; +// message id (2 bytes) + chunk counter (4 bytes) +const HEADER_SIZE = 2 + 4; // chunk counter of an acknowledgement, unused by messages -// as they have at most 0xff chunks, numbered from 0 -const ACK_CHUNK_ID = 0xff; +const ACK_CHUNK_ID = 0xffff_ffff; // at which interval to poll const TICK = 10; @@ -23,42 +22,42 @@ const TICK = 10; // how long to wait for sent messages to be received before giving up const MAX_DRAIN_TIME = 10_000; +// Maximum safe SCTP message size supported across browsers (Chrome/Firefox/Node) +const MAX_TRANSPORT_MESSAGE_SIZE = 262_144; // 256 KiB + // we can't use the definition in DOM as we're platform independent export type SignalData = | { type: "answer" | "offer" | "pranswer" | "rollback"; sdp?: string } | { type: "transceiverRequest"; transceiverRequest: { kind: string } } | { type: "renegotiate"; renegotiate: true } - | { type: "candidate"; candidate: RTCIceCandidate }; - -interface Events { - close: () => void; - connect: () => void; - error: (err: Error) => void; - signal: (signal: SignalData) => void; - data: (data: Buffer) => void; -} - -// Peer wraps a SimplePeer, adding message fragmentation + | { type: "candidate"; candidate: RTCIceCandidateInit }; + +type EventPayloads = { + close: void; + connect: void; + error: Error; + signal: SignalData; + data: Buffer; + [key: string]: unknown; +}; + +// Peer adds message fragmentation, flow control, and receipt acknowledgements +// over standard WebRTC RTCPeerConnection and RTCDataChannel. // -// WebRTC implementations have various maximum message size -// but with huge models, our messages might be bigger. -// We split messages by chunks and reconstruct these -// on the other side. +// WebRTC implementations have various maximum message sizes (e.g. 256 KiB in Chromium). +// For large models (megabytes to gigabytes), messages are split into chunks +// and reconstructed on the receiving peer. // -// As the WebRTC's DataChannel is not a stream, we need -// reorder messages, so we use a header on each chunk -// with a message id and chunk counter. The first chunk -// (chunk counter == 0), also add the total number of chunk. -// -// Once a message is fully received, its id is sent back -// with ACK_CHUNK_ID as chunk counter, so that the sender -// knows when it can close the connection without losing it. -// -// see feross/simple-peer#393 for more info -export class Peer { - private readonly peer: SimplePeer.Instance; - - private bufferSize?: number; +// As the WebRTC DataChannel is not a stream, we frame each chunk with a message ID, +// chunk counter, and total chunk count. Once a message is fully received, its ID is sent back +// with ACK_CHUNK_ID so the sender can drain before closing the connection. +export class Peer extends EventEmitter { + private readonly pc: RTCPeerConnection; + private channel?: RTCDataChannel; + private isDestroyed = false; + private signalQueue = Promise.resolve(); + private queuedCandidates: RTCIceCandidateInit[] = []; + private remoteMaxMessageSize?: number; private sendCounter: MessageID = 0; private sendQueue = List(); @@ -76,16 +75,164 @@ export class Peer { public readonly id: NodeID, initiator: boolean = false, ) { - this.peer = new SimplePeer({ wrtc, initiator }); + super(); + + this.pc = new wrtc.RTCPeerConnection(); + + this.pc.onicecandidate = (event) => { + if (event.candidate) { + const candidate: RTCIceCandidateInit = event.candidate.toJSON + ? event.candidate.toJSON() + : { + candidate: event.candidate.candidate, + sdpMid: event.candidate.sdpMid, + sdpMLineIndex: event.candidate.sdpMLineIndex, + usernameFragment: event.candidate.usernameFragment, + }; + this.emit("signal", { type: "candidate", candidate }); + } + }; - this.peer.on("data", (data: unknown) => { + this.pc.onconnectionstatechange = () => { if ( - Buffer.isBuffer(data) && - data.length === HEADER_SIZE && - data.readUInt8(2) === ACK_CHUNK_ID - ) - this.unacknowledged = this.unacknowledged.delete(data.readUInt16BE()); - }); + this.pc.connectionState === "failed" || + this.pc.connectionState === "closed" + ) { + this.emit("close", undefined); + } + }; + + if (initiator) { + const channel = this.pc.createDataChannel("disco"); + this.setupChannel(channel); + queueMicrotask(() => { + void this.makeOffer(); + }); + } else { + this.pc.ondatachannel = (event) => { + this.setupChannel(event.channel); + }; + } + } + + private setupChannel(channel: RTCDataChannel): void { + this.channel = channel; + this.channel.binaryType = "arraybuffer"; + try { + this.channel.bufferedAmountLowThreshold = MAX_TRANSPORT_MESSAGE_SIZE; + } catch { + // Some WebRTC implementations may not support bufferedAmountLowThreshold + } + + this.channel.onopen = () => { + this.emit("connect", undefined); + this.flush(); + }; + + this.channel.onmessage = (event: MessageEvent) => { + const data = Buffer.isBuffer(event.data) + ? event.data + : Buffer.from(event.data as ArrayBuffer); + this.handleData(data); + }; + + this.channel.onbufferedamountlow = () => { + this.flush(); + }; + + this.channel.onclose = () => { + this.emit("close", undefined); + }; + + this.channel.onerror = (event) => { + const message = + "error" in event && event.error instanceof Error + ? event.error.message + : "DataChannel error"; + this.emit("error", new Error(message)); + }; + } + + private async makeOffer(): Promise { + if (this.isDestroyed) return; + try { + const offer = await this.pc.createOffer(); + await this.pc.setLocalDescription(offer); + this.emit("signal", { + type: "offer", + sdp: offer.sdp, + }); + } catch (err) { + this.emit("error", err instanceof Error ? err : new Error(String(err))); + } + } + + signal(signal: SignalData): void { + this.signalQueue = this.signalQueue + .then(() => this.handleSignal(signal)) + .catch((err) => { + this.emit("error", err instanceof Error ? err : new Error(String(err))); + }); + } + + private async handleSignal(signal: SignalData): Promise { + if (this.isDestroyed) return; + + if (signal.type === "offer") { + if (signal.sdp === undefined) { + throw new Error("signal offer without session description"); + } + this.extractMaxMessageSize(signal.sdp); + await this.pc.setRemoteDescription( + new wrtc.RTCSessionDescription({ type: "offer", sdp: signal.sdp }), + ); + await this.drainQueuedCandidates(); + const answer = await this.pc.createAnswer(); + await this.pc.setLocalDescription(answer); + this.emit("signal", { + type: "answer", + sdp: answer.sdp, + }); + } else if (signal.type === "answer") { + if (signal.sdp === undefined) { + throw new Error("signal answer without session description"); + } + this.extractMaxMessageSize(signal.sdp); + await this.pc.setRemoteDescription( + new wrtc.RTCSessionDescription({ type: "answer", sdp: signal.sdp }), + ); + await this.drainQueuedCandidates(); + } else if (signal.type === "candidate") { + if (!signal.candidate) return; + if (this.pc.remoteDescription && this.pc.remoteDescription.type) { + await this.pc.addIceCandidate( + new wrtc.RTCIceCandidate(signal.candidate), + ); + } else { + this.queuedCandidates.push(signal.candidate); + } + } + } + + private extractMaxMessageSize(sdp: string): void { + const match = sdp.match(/a=max-message-size:(\d+)/); + if (match !== null) { + const max = parseInt(match[1], 10); + if (!isNaN(max)) { + this.remoteMaxMessageSize = max; + } + } + } + + private async drainQueuedCandidates(): Promise { + for (const c of this.queuedCandidates) { + try { + await this.pc.addIceCandidate(new wrtc.RTCIceCandidate(c)); + } catch { + // Ignore candidate errors if transport already connected + } + } + this.queuedCandidates = []; } send(msg: Buffer): void { @@ -98,54 +245,76 @@ export class Peer { private acknowledge(messageID: MessageID): void { const ack = Buffer.alloc(HEADER_SIZE); - ack.writeUint16BE(messageID); - ack.writeUint8(ACK_CHUNK_ID, 2); + ack.writeUInt16BE(messageID, 0); + ack.writeUInt32BE(ACK_CHUNK_ID, 2); this.sendQueue = this.sendQueue.push(ack); this.flush(); } private flush(): void { - if (this.bufferSize === undefined) { - throw new Error("flush without known buffer size"); - } - - const chunk = this.sendQueue.first(); - if (chunk === undefined) { - return; // nothing to flush - } - - const remainingBufferSize = this.bufferSize - this.peer.bufferSize; - if (chunk.length > remainingBufferSize) { - setTimeout(() => { - this.flush(); - }, TICK); + if ( + this.isDestroyed || + !this.channel || + this.channel.readyState !== "open" + ) { return; } - this.sendQueue = this.sendQueue.shift(); - this.peer.send(chunk); + const HIGH_WATER_MARK = 1024 * 1024; // 1 MB queue backpressure cap - // and loop - this.flush(); + while (!this.sendQueue.isEmpty()) { + if (this.channel.bufferedAmount >= HIGH_WATER_MARK) { + setTimeout(() => this.flush(), TICK); + return; + } + + const chunk = this.sendQueue.first(); + if (chunk === undefined) break; + + this.sendQueue = this.sendQueue.shift(); + try { + (this.channel.send as (data: unknown) => void)(chunk); + } catch (err) { + this.emit( + "error", + err instanceof Error ? err : new Error("Failed to send chunk"), + ); + return; + } + } } get maxChunkSize(): number { - if (this.bufferSize === undefined) { - throw new Error("chunk without known buffer size"); + // 1. Built-in RTCSctpTransport.maxMessageSize from the active connection + const sctpLimit = this.pc.sctp?.maxMessageSize; + if ( + typeof sctpLimit === "number" && + sctpLimit > 0 && + Number.isFinite(sctpLimit) + ) { + // Chromium send limit is 256 KiB (262,144 bytes). + // Even if remote peer (like Firefox) advertises 1 GiB, cap at 256 KiB for universal cross-browser reliability. + return Math.min(sctpLimit, MAX_TRANSPORT_MESSAGE_SIZE); } - return this.bufferSize; + + // 2. Fallback to remote SDP if extracted + if (this.remoteMaxMessageSize !== undefined) { + return Math.min(this.remoteMaxMessageSize, MAX_TRANSPORT_MESSAGE_SIZE); + } + + // 3. Safe universal default (256 KiB) + return MAX_TRANSPORT_MESSAGE_SIZE; } private chunk(b: Buffer): Seq.Indexed { const messageID = this.sendCounter; this.sendCounter++; if (this.sendCounter > 0xffff) { - throw new Error("too much messages sent to this peer"); + throw new Error("too many messages sent to this peer"); } - // special case as Range(1, 0) yields a value let tail = Seq.Indexed([]); - if (b.length > this.maxChunkSize) { + if (b.length > this.maxChunkSize - FIRST_HEADER_SIZE) { tail = Range( this.maxChunkSize - FIRST_HEADER_SIZE, b.length, @@ -156,10 +325,9 @@ export class Peer { } const totalChunkCount = 1 + tail.count(); - if (totalChunkCount > 0xff) { + if (totalChunkCount >= ACK_CHUNK_ID) { throw new Error( - `The payload is too big: ${totalChunkCount * this.maxChunkSize} bytes > 255,` + - " consider reducing the model size or increasing the chunk size", + `The payload is too big: ${totalChunkCount} chunks exceeds max chunks (${ACK_CHUNK_ID})`, ); } @@ -168,10 +336,9 @@ export class Peer { ? this.maxChunkSize : FIRST_HEADER_SIZE + b.length, ); - firstChunk.writeUint16BE(messageID); - const asChunkID = (n: number): ChunkID => n; - firstChunk.writeUint8(asChunkID(0), 2); - firstChunk.writeUint8(totalChunkCount, 3); + firstChunk.writeUInt16BE(messageID, 0); + firstChunk.writeUInt32BE(0, 2); + firstChunk.writeUInt32BE(totalChunkCount, 6); b.copy( firstChunk, FIRST_HEADER_SIZE, @@ -180,12 +347,12 @@ export class Peer { ); return Seq.Indexed([firstChunk]).concat( - Range(asChunkID(1), Number.POSITIVE_INFINITY) + Range(1, Number.POSITIVE_INFINITY) .zip(tail) .map(([id, raw]) => { const chunk = Buffer.alloc(HEADER_SIZE + raw.length); - chunk.writeUint16BE(messageID); - chunk.writeUint8(id, 2); + chunk.writeUInt16BE(messageID, 0); + chunk.writeUInt32BE(id, 2); raw.copy(chunk, HEADER_SIZE, 0); return chunk; }), @@ -198,121 +365,98 @@ export class Peer { async drain(): Promise { const deadline = Date.now() + MAX_DRAIN_TIME; while ( - !this.peer.destroyed && + !this.isDestroyed && !this.unacknowledged.isEmpty() && Date.now() < deadline - ) + ) { await new Promise((resolve) => setTimeout(resolve, TICK)); + } } async destroy(): Promise { - return new Promise((resolve, reject) => { - this.peer.once("error", reject); - this.peer.once("close", resolve); - this.peer.destroy(); - }); - } - - signal(signal: SignalData): void { - // extract max buffer size from the signal - if (signal.type === "offer" || signal.type === "answer") { - if (signal.sdp === undefined) { - throw new Error("signal answer|offer without session description"); - } - if (this.bufferSize !== undefined) { - throw new Error("buffer size set twice"); - } + if (this.isDestroyed) return; + this.isDestroyed = true; - const match = signal.sdp.match(/a=max-message-size:(\d+)/); - if (match === null) { - // TODO default value instead? - throw new Error("no max-message-size found in signal"); + try { + if (this.channel) { + this.channel.close(); } - const max = parseInt(match[1], 10); - if (isNaN(max)) { - throw new Error(`unable to parse max-message-size as int: ${match[1]}`); - } - - this.bufferSize = max; + this.pc.close(); + } catch { + // Ignore closing errors } - this.peer.signal(signal); + this.emit("close", undefined); + await Promise.resolve(); } - on(event: K, listener: Events[K]): void { - if (event !== "data") { - this.peer.on(event, listener); + private handleData(data: Buffer): void { + if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { + throw new Error("received invalid message type"); + } + const messageID: MessageID = data.readUInt16BE(0); + const chunkID: ChunkID = data.readUInt32BE(2); + if (chunkID === ACK_CHUNK_ID) { + this.unacknowledged = this.unacknowledged.delete(messageID); return; } - // gotta help typescript here - const dataListener = listener as Events["data"]; - this.peer.on("data", (data: unknown) => { - if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { + const received = this.receiving.get(messageID, { + total: undefined, + chunks: Map(), + }); + let total = received.total; + const chunks = received.chunks; + + if (chunks.has(chunkID)) { + throw new Error(`chunk ${messageID}:${chunkID} already received`); + } + + let chunk: Buffer; + if (chunkID !== 0) { + chunk = Buffer.alloc(data.length - HEADER_SIZE); + data.copy(chunk, 0, HEADER_SIZE); + } else { + if (data.length < FIRST_HEADER_SIZE) { throw new Error("received invalid message type"); } - const messageID: MessageID = data.readUInt16BE(); //readUint16BE (case sensitive) fails at runtime - const chunkID: ChunkID = data.readUInt8(2); // same for readUint8 - if (chunkID === ACK_CHUNK_ID) return; // handled in the constructor + if (total !== undefined) { + throw new Error("first header received twice"); + } - const received = this.receiving.get(messageID, { - total: undefined, - chunks: Map(), - }); - let total = received.total; - const chunks = received.chunks; + const readTotal = data.readUInt32BE(6); + total = readTotal; + chunk = Buffer.alloc(data.length - FIRST_HEADER_SIZE); + data.copy(chunk, 0, FIRST_HEADER_SIZE); - if (chunks.has(chunkID)) { - throw new Error(`chunk ${messageID}:${chunkID} already received`); + if (chunks.keySeq().some((id) => id >= readTotal)) { + throw new Error( + "received total of chunk but got now-out-of-bound chunks", + ); } + } + this.receiving = this.receiving.set(messageID, { + total, + chunks: chunks.set(chunkID, chunk), + }); - let chunk: Buffer; - if (chunkID !== 0) { - chunk = Buffer.alloc(data.length - HEADER_SIZE); - data.copy(chunk, 0, HEADER_SIZE); - } else { - if (data.length < FIRST_HEADER_SIZE) { - throw new Error("received invalid message type"); - } - if (total !== undefined) { - throw new Error("first header received twice"); - } - - const readTotal = data.readUInt8(3); - total = readTotal; - chunk = Buffer.alloc(data.length - FIRST_HEADER_SIZE); - data.copy(chunk, 0, FIRST_HEADER_SIZE); - - if (chunks.keySeq().some((id) => id > readTotal)) { - throw new Error( - "received total of chunk but got now-out-of-bound chunks", - ); - } - } - this.receiving = this.receiving.set(messageID, { - total, - chunks: chunks.set(chunkID, chunk), - }); + const readyMessages = this.receiving + .filter( + ({ total, chunks }) => total !== undefined && chunks.size === total, + ) + .sort() + .map(({ chunks }) => + chunks + .entrySeq() + .toList() + .sortBy(([id]) => id), + ) + .map((chunks) => Buffer.concat(chunks.map(([_, b]) => b).toArray())); + this.receiving = this.receiving.deleteAll(readyMessages.keys()); - const readyMessages = this.receiving - .filter( - ({ total, chunks }) => total !== undefined && chunks.size === total, - ) - .sort() - .map(({ chunks }) => - chunks - .entrySeq() - .toList() - .sortBy(([id, _]) => id), - ) - .map((chunks) => Buffer.concat(chunks.map(([_, b]) => b).toArray())); - this.receiving = this.receiving.deleteAll(readyMessages.keys()); - - readyMessages.forEach((message, messageID) => { - this.acknowledge(messageID); - // TODO debug - dataListener(message); - }); + readyMessages.forEach((message, messageID) => { + this.acknowledge(messageID); + this.emit("data", message); }); } } diff --git a/discojs/tsconfig.lib.json b/discojs/tsconfig.lib.json index 6d137af39..8981b36e3 100644 --- a/discojs/tsconfig.lib.json +++ b/discojs/tsconfig.lib.json @@ -1,6 +1,7 @@ { "extends": "../tsconfig.base.lib.json", "compilerOptions": { + "types": ["node"], // Copied from `package.json` since some tools don't work without it "paths": { "#root/*": ["./src/*"], diff --git a/discojs/tsconfig.vitest.json b/discojs/tsconfig.vitest.json index f33d5a873..d07d9971e 100644 --- a/discojs/tsconfig.vitest.json +++ b/discojs/tsconfig.vitest.json @@ -1,6 +1,7 @@ { "extends": "../tsconfig.base.json", "compilerOptions": { + "types": ["node"], "noEmit": true }, "include": ["src"] diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 95b04c3a1..958674a38 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -350,16 +350,13 @@ importers: immutable: specifier: 'catalog:' version: 5.1.6 - simple-peer: - specifier: 9.11.1 - version: 9.11.1(supports-color@8.1.1) zod: specifier: 'catalog:' version: 4.3.6 devDependencies: - '@types/simple-peer': - specifier: 9.11.9 - version: 9.11.9 + '@types/node': + specifier: 22.19.15 + version: 22.19.15 nodemon: specifier: 3.1.14 version: 3.1.14 @@ -2017,9 +2014,6 @@ packages: '@types/serve-static@2.2.0': resolution: {integrity: sha512-8mam4H1NHLtu7nmtalF7eyBH14QyOASmcxHhSfEoRyr0nP/YdoesEtU+uSRvMe96TW/HPTtkoKqQLl53N7UXMQ==} - '@types/simple-peer@9.11.9': - resolution: {integrity: sha512-6Gdl7TSS5oh9nuwKD4Pl8cSmaxWycYeZz9HLnJBNvIwWjZuGVsmHe9RwW3+9RxfhC1aIR9Z83DvaJoMw6rhkbg==} - '@types/sinonjs__fake-timers@8.1.1': resolution: {integrity: sha512-0kSuKjAS0TrGLJ0M/+8MaFkGsQhZpB6pxOmvS3K8FYI72K//YmdfoW9X2qPsAKh1mkwxGD5zib9s1FIFed6E8g==} @@ -2927,9 +2921,6 @@ packages: resolution: {integrity: sha512-xUtoPkMggbz0MPyPiIWr1Kp4aeWJjDZ6SMvURhimjdZgsRuDplF5/s9hcgGhyXMhs+6vpnuoiZ2kFiu3FMnS8Q==} engines: {node: '>=18'} - err-code@3.0.1: - resolution: {integrity: sha512-GiaH0KJUewYok+eeY05IIgjtAe4Yltygk9Wqp1V5yVWLdhf0hYZchRjNIT9bb0mSwRcIusT3cx7PJUf3zEIfUA==} - es-define-property@1.0.1: resolution: {integrity: sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g==} engines: {node: '>= 0.4'} @@ -4146,9 +4137,6 @@ packages: queue-microtask@1.2.3: resolution: {integrity: sha512-NuaNSa6flKT5JaSYQzJok04JzTL1CA6aGhv5rfLW3PgqA+M2ChpZQnAC8h8i4ZFkBS8X5RqkDBHA7r4hej3K9A==} - randombytes@2.1.0: - resolution: {integrity: sha512-vYl3iOX+4CKUWuxGi9Ukhie6fsqXqS9FE2Zaic4tNFD2N2QQaXOMFbuKK4QmDHC0JO6B1Zp41J0LpT0oR68amQ==} - range-parser@1.2.1: resolution: {integrity: sha512-Hrgsx+orqoygnmhFbKaHE6c296J+HTAQXoxEF6gNupROmmGJRoyzfG3ccAveqCBrwr/2yxQ5BVd/GTl5agOwSg==} engines: {node: '>= 0.6'} @@ -4336,9 +4324,6 @@ packages: simple-get@4.0.1: resolution: {integrity: sha512-brv7p5WgH0jmQJr1ZDDfKDOSeWWg+OVypG99A/5vYGPqJ6pxiaHLy8nxtFjBA7oMa01ebA9gfh1uMCFqOuXxvA==} - simple-peer@9.11.1: - resolution: {integrity: sha512-D1SaWpOW8afq1CZGWB8xTfrT3FekjQmPValrqncJMX7QFl8YwhrPTZvMCANLtgBwwdS+7zURyqxDDEmY558tTw==} - simple-update-notifier@2.0.0: resolution: {integrity: sha512-a2B9Y0KlNXl9u/vsW6sTIu9vGEpfKu2wRV6l1H3XEas/0gUIzGzBoP/IouTcUQbm9JWZLH3COxyn03TYlFax6w==} engines: {node: '>=10'} @@ -6163,10 +6148,6 @@ snapshots: '@types/http-errors': 2.0.5 '@types/node': 22.19.15 - '@types/simple-peer@9.11.9': - dependencies: - '@types/node': 22.19.15 - '@types/sinonjs__fake-timers@8.1.1': {} '@types/sizzle@2.3.10': {} @@ -7215,8 +7196,6 @@ snapshots: environment@1.1.0: {} - err-code@3.0.1: {} - es-define-property@1.0.1: {} es-errors@1.3.0: {} @@ -8480,10 +8459,6 @@ snapshots: queue-microtask@1.2.3: {} - randombytes@2.1.0: - dependencies: - safe-buffer: 5.2.1 - range-parser@1.2.1: {} raw-body@3.0.2: @@ -8747,18 +8722,6 @@ snapshots: once: 1.4.0 simple-concat: 1.0.1 - simple-peer@9.11.1(supports-color@8.1.1): - dependencies: - buffer: 6.0.3 - debug: 4.4.3(supports-color@8.1.1) - err-code: 3.0.1 - get-browser-rtc: 1.1.0 - queue-microtask: 1.2.3 - randombytes: 2.1.0 - readable-stream: 3.6.2 - transitivePeerDependencies: - - supports-color - simple-update-notifier@2.0.0: dependencies: semver: 7.8.5 From cd3daadcc3da0bc29341768cc597950d830d2604 Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 15:35:29 +0200 Subject: [PATCH 02/10] Remove redundant SDP regex parsing for max message size Rely directly on W3C standard pc.sctp.maxMessageSize and cross-browser safety default instead of regex matching raw SDP strings. --- discojs/src/client/decentralized/peer.ts | 20 +------------------- 1 file changed, 1 insertion(+), 19 deletions(-) diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index 7425ac50f..c07e212aa 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -57,7 +57,6 @@ export class Peer extends EventEmitter { private isDestroyed = false; private signalQueue = Promise.resolve(); private queuedCandidates: RTCIceCandidateInit[] = []; - private remoteMaxMessageSize?: number; private sendCounter: MessageID = 0; private sendQueue = List(); @@ -182,7 +181,6 @@ export class Peer extends EventEmitter { if (signal.sdp === undefined) { throw new Error("signal offer without session description"); } - this.extractMaxMessageSize(signal.sdp); await this.pc.setRemoteDescription( new wrtc.RTCSessionDescription({ type: "offer", sdp: signal.sdp }), ); @@ -197,7 +195,6 @@ export class Peer extends EventEmitter { if (signal.sdp === undefined) { throw new Error("signal answer without session description"); } - this.extractMaxMessageSize(signal.sdp); await this.pc.setRemoteDescription( new wrtc.RTCSessionDescription({ type: "answer", sdp: signal.sdp }), ); @@ -214,16 +211,6 @@ export class Peer extends EventEmitter { } } - private extractMaxMessageSize(sdp: string): void { - const match = sdp.match(/a=max-message-size:(\d+)/); - if (match !== null) { - const max = parseInt(match[1], 10); - if (!isNaN(max)) { - this.remoteMaxMessageSize = max; - } - } - } - private async drainQueuedCandidates(): Promise { for (const c of this.queuedCandidates) { try { @@ -297,12 +284,7 @@ export class Peer extends EventEmitter { return Math.min(sctpLimit, MAX_TRANSPORT_MESSAGE_SIZE); } - // 2. Fallback to remote SDP if extracted - if (this.remoteMaxMessageSize !== undefined) { - return Math.min(this.remoteMaxMessageSize, MAX_TRANSPORT_MESSAGE_SIZE); - } - - // 3. Safe universal default (256 KiB) + // Safe universal default (256 KiB) return MAX_TRANSPORT_MESSAGE_SIZE; } From 5b68776dc8187ae04e519b31799fa73d3eea017a Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 18:07:16 +0200 Subject: [PATCH 03/10] Abstract protocol logic into Packet class and store header sizes in variables - Introduce Packet data class with encode, decode, and factory methods - Store message ID, chunk ID, total chunk count sizes, and header sizes in explicit constants - Restore asChunkID helper function - Replace hardcoded RFC 8841 default with DEFAULT_MAX_MESSAGE_SIZE variable - Add unit tests for Packet serialization and header decoding --- discojs/src/client/decentralized/peer.spec.ts | 39 +++- discojs/src/client/decentralized/peer.ts | 201 ++++++++++++------ 2 files changed, 176 insertions(+), 64 deletions(-) diff --git a/discojs/src/client/decentralized/peer.spec.ts b/discojs/src/client/decentralized/peer.spec.ts index 67916c5b4..90ec90c5c 100644 --- a/discojs/src/client/decentralized/peer.spec.ts +++ b/discojs/src/client/decentralized/peer.spec.ts @@ -1,6 +1,43 @@ import { List, Range, Set } from "immutable"; import { assert, afterEach, beforeEach, describe, it } from "vitest"; -import { Peer } from "#client/decentralized/peer"; +import { Packet, Peer } from "#client/decentralized/peer"; + +describe("packet", () => { + it("encodes and decodes an ack packet", () => { + const ack = Packet.ack(42); + const encoded = ack.encode(); + const decoded = Packet.decode(encoded); + + assert.isTrue(decoded.isAck); + assert.strictEqual(decoded.messageID, 42); + }); + + it("encodes and decodes a first chunk packet with 32-bit chunk counter and total count", () => { + const payload = Buffer.from("hello world"); + const firstChunk = Packet.firstChunk(101, 500, payload); + const encoded = firstChunk.encode(); + const decoded = Packet.decode(encoded); + + assert.isTrue(decoded.isFirstChunk); + assert.strictEqual(decoded.messageID, 101); + assert.strictEqual(decoded.chunkID, 0); + assert.strictEqual(decoded.totalChunks, 500); + assert.strictEqual(decoded.payload.toString(), "hello world"); + }); + + it("encodes and decodes a subsequent chunk packet", () => { + const payload = Buffer.from("subsequent chunk"); + const chunk = Packet.chunk(101, 299, payload); + const encoded = chunk.encode(); + const decoded = Packet.decode(encoded); + + assert.isFalse(decoded.isAck); + assert.isFalse(decoded.isFirstChunk); + assert.strictEqual(decoded.messageID, 101); + assert.strictEqual(decoded.chunkID, 299); + assert.strictEqual(decoded.payload.toString(), "subsequent chunk"); + }); +}); describe("peer", () => { let peer1: Peer; diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index c07e212aa..c628e2c55 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -7,14 +7,24 @@ import { EventEmitter } from "#utils/event_emitter"; type MessageID = number; type ChunkID = number; -// message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) -const FIRST_HEADER_SIZE = 2 + 4 + 4; +// helper function to make it evident that the number passed is a chunk id +const asChunkID = (n: number): ChunkID => n; + +// message id size in bytes (uint16) +const MESSAGE_ID_BYTES = 2; +// chunk counter size in bytes (uint32) +const CHUNK_ID_BYTES = 4; +// total chunk count size in bytes (uint32) +const TOTAL_CHUNKS_BYTES = 4; // message id (2 bytes) + chunk counter (4 bytes) -const HEADER_SIZE = 2 + 4; +const HEADER_SIZE = MESSAGE_ID_BYTES + CHUNK_ID_BYTES; + +// message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) +const FIRST_HEADER_SIZE = HEADER_SIZE + TOTAL_CHUNKS_BYTES; // chunk counter of an acknowledgement, unused by messages -const ACK_CHUNK_ID = 0xffff_ffff; +const ACK_CHUNK_ID: ChunkID = 0xffff_ffff; // at which interval to poll const TICK = 10; @@ -22,8 +32,99 @@ const TICK = 10; // how long to wait for sent messages to be received before giving up const MAX_DRAIN_TIME = 10_000; -// Maximum safe SCTP message size supported across browsers (Chrome/Firefox/Node) -const MAX_TRANSPORT_MESSAGE_SIZE = 262_144; // 256 KiB +// Fallback SCTP message size per RFC 8841 / WebRTC spec when transport is not yet connected +const DEFAULT_MAX_MESSAGE_SIZE = 65_536; // 64 KiB + +export class Packet { + constructor( + public readonly messageID: MessageID, + public readonly chunkID: ChunkID, + public readonly totalChunks?: number, + public readonly payload: Buffer = Buffer.alloc(0), + ) {} + + get isAck(): boolean { + return this.chunkID === ACK_CHUNK_ID; + } + + get isFirstChunk(): boolean { + return this.chunkID === asChunkID(0); + } + + static ack(messageID: MessageID): Packet { + return new Packet(messageID, ACK_CHUNK_ID); + } + + static firstChunk( + messageID: MessageID, + totalChunks: number, + payload: Buffer, + ): Packet { + return new Packet(messageID, asChunkID(0), totalChunks, payload); + } + + static chunk( + messageID: MessageID, + chunkID: ChunkID, + payload: Buffer, + ): Packet { + return new Packet(messageID, chunkID, undefined, payload); + } + + encode(): Buffer { + if (this.isAck) { + const buf = Buffer.alloc(HEADER_SIZE); + buf.writeUInt16BE(this.messageID, 0); + buf.writeUInt32BE(ACK_CHUNK_ID, MESSAGE_ID_BYTES); + return buf; + } + + if (this.isFirstChunk) { + if (this.totalChunks === undefined) { + throw new Error("first chunk packet requires totalChunks"); + } + const buf = Buffer.alloc(FIRST_HEADER_SIZE + this.payload.length); + buf.writeUInt16BE(this.messageID, 0); + buf.writeUInt32BE(asChunkID(0), MESSAGE_ID_BYTES); + buf.writeUInt32BE(this.totalChunks, HEADER_SIZE); + this.payload.copy(buf, FIRST_HEADER_SIZE, 0); + return buf; + } + + const buf = Buffer.alloc(HEADER_SIZE + this.payload.length); + buf.writeUInt16BE(this.messageID, 0); + buf.writeUInt32BE(this.chunkID, MESSAGE_ID_BYTES); + this.payload.copy(buf, HEADER_SIZE, 0); + return buf; + } + + static decode(data: Buffer): Packet { + if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { + throw new Error("received invalid message type"); + } + + const messageID: MessageID = data.readUInt16BE(0); + const chunkID: ChunkID = data.readUInt32BE(MESSAGE_ID_BYTES); + + if (chunkID === ACK_CHUNK_ID) { + return Packet.ack(messageID); + } + + if (chunkID === asChunkID(0)) { + if (data.length < FIRST_HEADER_SIZE) { + throw new Error("received invalid message type"); + } + const totalChunks = data.readUInt32BE(HEADER_SIZE); + const payload = Buffer.alloc(data.length - FIRST_HEADER_SIZE); + data.copy(payload, 0, FIRST_HEADER_SIZE); + return Packet.firstChunk(messageID, totalChunks, payload); + } + + const payload = Buffer.alloc(data.length - HEADER_SIZE); + data.copy(payload, 0, HEADER_SIZE); + return Packet.chunk(messageID, chunkID, payload); + } +} // we can't use the definition in DOM as we're platform independent export type SignalData = @@ -118,7 +219,7 @@ export class Peer extends EventEmitter { this.channel = channel; this.channel.binaryType = "arraybuffer"; try { - this.channel.bufferedAmountLowThreshold = MAX_TRANSPORT_MESSAGE_SIZE; + this.channel.bufferedAmountLowThreshold = this.maxChunkSize; } catch { // Some WebRTC implementations may not support bufferedAmountLowThreshold } @@ -231,9 +332,7 @@ export class Peer extends EventEmitter { } private acknowledge(messageID: MessageID): void { - const ack = Buffer.alloc(HEADER_SIZE); - ack.writeUInt16BE(messageID, 0); - ack.writeUInt32BE(ACK_CHUNK_ID, 2); + const ack = Packet.ack(messageID).encode(); this.sendQueue = this.sendQueue.push(ack); this.flush(); } @@ -272,20 +371,16 @@ export class Peer extends EventEmitter { } get maxChunkSize(): number { - // 1. Built-in RTCSctpTransport.maxMessageSize from the active connection const sctpLimit = this.pc.sctp?.maxMessageSize; if ( typeof sctpLimit === "number" && sctpLimit > 0 && Number.isFinite(sctpLimit) ) { - // Chromium send limit is 256 KiB (262,144 bytes). - // Even if remote peer (like Firefox) advertises 1 GiB, cap at 256 KiB for universal cross-browser reliability. - return Math.min(sctpLimit, MAX_TRANSPORT_MESSAGE_SIZE); + return sctpLimit; } - // Safe universal default (256 KiB) - return MAX_TRANSPORT_MESSAGE_SIZE; + return DEFAULT_MAX_MESSAGE_SIZE; } private chunk(b: Buffer): Seq.Indexed { @@ -313,31 +408,20 @@ export class Peer extends EventEmitter { ); } - const firstChunk = Buffer.alloc( - b.length > this.maxChunkSize - FIRST_HEADER_SIZE - ? this.maxChunkSize - : FIRST_HEADER_SIZE + b.length, - ); - firstChunk.writeUInt16BE(messageID, 0); - firstChunk.writeUInt32BE(0, 2); - firstChunk.writeUInt32BE(totalChunkCount, 6); - b.copy( - firstChunk, - FIRST_HEADER_SIZE, + const firstPayload = b.subarray( 0, - this.maxChunkSize - FIRST_HEADER_SIZE, + Math.min(b.length, this.maxChunkSize - FIRST_HEADER_SIZE), ); + const firstChunk = Packet.firstChunk( + messageID, + totalChunkCount, + firstPayload, + ).encode(); return Seq.Indexed([firstChunk]).concat( - Range(1, Number.POSITIVE_INFINITY) + Range(asChunkID(1), Number.POSITIVE_INFINITY) .zip(tail) - .map(([id, raw]) => { - const chunk = Buffer.alloc(HEADER_SIZE + raw.length); - chunk.writeUInt16BE(messageID, 0); - chunk.writeUInt32BE(id, 2); - raw.copy(chunk, HEADER_SIZE, 0); - return chunk; - }), + .map(([id, raw]) => Packet.chunk(messageID, id, raw).encode()), ); } @@ -373,53 +457,44 @@ export class Peer extends EventEmitter { } private handleData(data: Buffer): void { - if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { - throw new Error("received invalid message type"); - } - const messageID: MessageID = data.readUInt16BE(0); - const chunkID: ChunkID = data.readUInt32BE(2); - if (chunkID === ACK_CHUNK_ID) { - this.unacknowledged = this.unacknowledged.delete(messageID); + const packet = Packet.decode(data); + if (packet.isAck) { + this.unacknowledged = this.unacknowledged.delete(packet.messageID); return; } - const received = this.receiving.get(messageID, { + const received = this.receiving.get(packet.messageID, { total: undefined, chunks: Map(), }); let total = received.total; const chunks = received.chunks; - if (chunks.has(chunkID)) { - throw new Error(`chunk ${messageID}:${chunkID} already received`); + if (chunks.has(packet.chunkID)) { + throw new Error( + `chunk ${packet.messageID}:${packet.chunkID} already received`, + ); } - let chunk: Buffer; - if (chunkID !== 0) { - chunk = Buffer.alloc(data.length - HEADER_SIZE); - data.copy(chunk, 0, HEADER_SIZE); - } else { - if (data.length < FIRST_HEADER_SIZE) { - throw new Error("received invalid message type"); - } + if (packet.isFirstChunk) { if (total !== undefined) { throw new Error("first header received twice"); } - - const readTotal = data.readUInt32BE(6); - total = readTotal; - chunk = Buffer.alloc(data.length - FIRST_HEADER_SIZE); - data.copy(chunk, 0, FIRST_HEADER_SIZE); - - if (chunks.keySeq().some((id) => id >= readTotal)) { + if (packet.totalChunks === undefined) { + throw new Error("first chunk packet missing totalChunks"); + } + const totalLimit = packet.totalChunks; + total = totalLimit; + if (chunks.keySeq().some((id) => id >= totalLimit)) { throw new Error( "received total of chunk but got now-out-of-bound chunks", ); } } - this.receiving = this.receiving.set(messageID, { + + this.receiving = this.receiving.set(packet.messageID, { total, - chunks: chunks.set(chunkID, chunk), + chunks: chunks.set(packet.chunkID, packet.payload), }); const readyMessages = this.receiving From f0ad4db2d02de95343789a1bfdc97ae6d4f875af Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 18:15:15 +0200 Subject: [PATCH 04/10] Move Packet protocol data class and tests to dedicated packet.ts module Extract framing, encoding, decoding, and packet-level constants into packet.ts with dedicated unit test suite in packet.spec.ts. --- .../src/client/decentralized/packet.spec.ts | 43 ++++++ discojs/src/client/decentralized/packet.ts | 112 ++++++++++++++++ discojs/src/client/decentralized/peer.spec.ts | 39 +----- discojs/src/client/decentralized/peer.ts | 122 ++---------------- 4 files changed, 165 insertions(+), 151 deletions(-) create mode 100644 discojs/src/client/decentralized/packet.spec.ts create mode 100644 discojs/src/client/decentralized/packet.ts diff --git a/discojs/src/client/decentralized/packet.spec.ts b/discojs/src/client/decentralized/packet.spec.ts new file mode 100644 index 000000000..bfea49612 --- /dev/null +++ b/discojs/src/client/decentralized/packet.spec.ts @@ -0,0 +1,43 @@ +import { assert, describe, it } from "vitest"; +import { Packet } from "#client/decentralized/packet"; + +describe("packet", () => { + it("encodes and decodes an ack packet", () => { + const ack = Packet.ack(42); + const encoded = ack.encode(); + const decoded = Packet.decode(encoded); + + assert.isTrue(decoded.isAck); + assert.strictEqual(decoded.messageID, 42); + }); + + it("encodes and decodes a first chunk packet with 32-bit chunk counter and total count", () => { + const payload = Buffer.from("hello world"); + const firstChunk = Packet.firstChunk(101, 500, payload); + const encoded = firstChunk.encode(); + const decoded = Packet.decode(encoded); + + assert.isTrue(decoded.isFirstChunk); + assert.strictEqual(decoded.messageID, 101); + assert.strictEqual(decoded.chunkID, 0); + assert.strictEqual(decoded.totalChunks, 500); + assert.strictEqual(decoded.payload.toString(), "hello world"); + }); + + it("encodes and decodes a subsequent chunk packet", () => { + const payload = Buffer.from("subsequent chunk"); + const chunk = Packet.chunk(101, 299, payload); + const encoded = chunk.encode(); + const decoded = Packet.decode(encoded); + + assert.isFalse(decoded.isAck); + assert.isFalse(decoded.isFirstChunk); + assert.strictEqual(decoded.messageID, 101); + assert.strictEqual(decoded.chunkID, 299); + assert.strictEqual(decoded.payload.toString(), "subsequent chunk"); + }); + + it("rejects invalid buffer sizes on decode", () => { + assert.throws(() => Packet.decode(Buffer.alloc(2))); + }); +}); diff --git a/discojs/src/client/decentralized/packet.ts b/discojs/src/client/decentralized/packet.ts new file mode 100644 index 000000000..1fa9d80a5 --- /dev/null +++ b/discojs/src/client/decentralized/packet.ts @@ -0,0 +1,112 @@ +export type MessageID = number; +export type ChunkID = number; + +// helper function to make it evident that the number passed is a chunk id +export const asChunkID = (n: number): ChunkID => n; + +// message id size in bytes (uint16) +export const MESSAGE_ID_BYTES = 2; +// chunk counter size in bytes (uint32) +export const CHUNK_ID_BYTES = 4; +// total chunk count size in bytes (uint32) +export const TOTAL_CHUNKS_BYTES = 4; + +// message id (2 bytes) + chunk counter (4 bytes) +export const HEADER_SIZE = MESSAGE_ID_BYTES + CHUNK_ID_BYTES; + +// message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) +export const FIRST_HEADER_SIZE = HEADER_SIZE + TOTAL_CHUNKS_BYTES; + +// chunk counter of an acknowledgement, unused by messages +export const ACK_CHUNK_ID: ChunkID = 0xffff_ffff; + +export class Packet { + constructor( + public readonly messageID: MessageID, + public readonly chunkID: ChunkID, + public readonly totalChunks?: number, + public readonly payload: Buffer = Buffer.alloc(0), + ) {} + + get isAck(): boolean { + return this.chunkID === ACK_CHUNK_ID; + } + + get isFirstChunk(): boolean { + return this.chunkID === asChunkID(0); + } + + static ack(messageID: MessageID): Packet { + return new Packet(messageID, ACK_CHUNK_ID); + } + + static firstChunk( + messageID: MessageID, + totalChunks: number, + payload: Buffer, + ): Packet { + return new Packet(messageID, asChunkID(0), totalChunks, payload); + } + + static chunk( + messageID: MessageID, + chunkID: ChunkID, + payload: Buffer, + ): Packet { + return new Packet(messageID, chunkID, undefined, payload); + } + + encode(): Buffer { + if (this.isAck) { + const buf = Buffer.alloc(HEADER_SIZE); + buf.writeUInt16BE(this.messageID, 0); + buf.writeUInt32BE(ACK_CHUNK_ID, MESSAGE_ID_BYTES); + return buf; + } + + if (this.isFirstChunk) { + if (this.totalChunks === undefined) { + throw new Error("first chunk packet requires totalChunks"); + } + const buf = Buffer.alloc(FIRST_HEADER_SIZE + this.payload.length); + buf.writeUInt16BE(this.messageID, 0); + buf.writeUInt32BE(asChunkID(0), MESSAGE_ID_BYTES); + buf.writeUInt32BE(this.totalChunks, HEADER_SIZE); + this.payload.copy(buf, FIRST_HEADER_SIZE, 0); + return buf; + } + + const buf = Buffer.alloc(HEADER_SIZE + this.payload.length); + buf.writeUInt16BE(this.messageID, 0); + buf.writeUInt32BE(this.chunkID, MESSAGE_ID_BYTES); + this.payload.copy(buf, HEADER_SIZE, 0); + return buf; + } + + static decode(data: Buffer): Packet { + if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { + throw new Error("received invalid message type"); + } + + const messageID: MessageID = data.readUInt16BE(0); + const chunkID: ChunkID = data.readUInt32BE(MESSAGE_ID_BYTES); + + if (chunkID === ACK_CHUNK_ID) { + return Packet.ack(messageID); + } + + if (chunkID === asChunkID(0)) { + if (data.length < FIRST_HEADER_SIZE) { + throw new Error("received invalid message type"); + } + const totalChunks = data.readUInt32BE(HEADER_SIZE); + const payload = Buffer.alloc(data.length - FIRST_HEADER_SIZE); + data.copy(payload, 0, FIRST_HEADER_SIZE); + return Packet.firstChunk(messageID, totalChunks, payload); + } + + const payload = Buffer.alloc(data.length - HEADER_SIZE); + data.copy(payload, 0, HEADER_SIZE); + return Packet.chunk(messageID, chunkID, payload); + } +} diff --git a/discojs/src/client/decentralized/peer.spec.ts b/discojs/src/client/decentralized/peer.spec.ts index 90ec90c5c..67916c5b4 100644 --- a/discojs/src/client/decentralized/peer.spec.ts +++ b/discojs/src/client/decentralized/peer.spec.ts @@ -1,43 +1,6 @@ import { List, Range, Set } from "immutable"; import { assert, afterEach, beforeEach, describe, it } from "vitest"; -import { Packet, Peer } from "#client/decentralized/peer"; - -describe("packet", () => { - it("encodes and decodes an ack packet", () => { - const ack = Packet.ack(42); - const encoded = ack.encode(); - const decoded = Packet.decode(encoded); - - assert.isTrue(decoded.isAck); - assert.strictEqual(decoded.messageID, 42); - }); - - it("encodes and decodes a first chunk packet with 32-bit chunk counter and total count", () => { - const payload = Buffer.from("hello world"); - const firstChunk = Packet.firstChunk(101, 500, payload); - const encoded = firstChunk.encode(); - const decoded = Packet.decode(encoded); - - assert.isTrue(decoded.isFirstChunk); - assert.strictEqual(decoded.messageID, 101); - assert.strictEqual(decoded.chunkID, 0); - assert.strictEqual(decoded.totalChunks, 500); - assert.strictEqual(decoded.payload.toString(), "hello world"); - }); - - it("encodes and decodes a subsequent chunk packet", () => { - const payload = Buffer.from("subsequent chunk"); - const chunk = Packet.chunk(101, 299, payload); - const encoded = chunk.encode(); - const decoded = Packet.decode(encoded); - - assert.isFalse(decoded.isAck); - assert.isFalse(decoded.isFirstChunk); - assert.strictEqual(decoded.messageID, 101); - assert.strictEqual(decoded.chunkID, 299); - assert.strictEqual(decoded.payload.toString(), "subsequent chunk"); - }); -}); +import { Peer } from "#client/decentralized/peer"; describe("peer", () => { let peer1: Peer; diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index c628e2c55..4078bffb5 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -3,28 +3,15 @@ import wrtc from "@epfml/isomorphic-wrtc"; import type { NodeID } from "#client/types"; import { EventEmitter } from "#utils/event_emitter"; - -type MessageID = number; -type ChunkID = number; - -// helper function to make it evident that the number passed is a chunk id -const asChunkID = (n: number): ChunkID => n; - -// message id size in bytes (uint16) -const MESSAGE_ID_BYTES = 2; -// chunk counter size in bytes (uint32) -const CHUNK_ID_BYTES = 4; -// total chunk count size in bytes (uint32) -const TOTAL_CHUNKS_BYTES = 4; - -// message id (2 bytes) + chunk counter (4 bytes) -const HEADER_SIZE = MESSAGE_ID_BYTES + CHUNK_ID_BYTES; - -// message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) -const FIRST_HEADER_SIZE = HEADER_SIZE + TOTAL_CHUNKS_BYTES; - -// chunk counter of an acknowledgement, unused by messages -const ACK_CHUNK_ID: ChunkID = 0xffff_ffff; +import { + type MessageID, + type ChunkID, + asChunkID, + HEADER_SIZE, + FIRST_HEADER_SIZE, + ACK_CHUNK_ID, + Packet, +} from "#client/decentralized/packet"; // at which interval to poll const TICK = 10; @@ -35,97 +22,6 @@ const MAX_DRAIN_TIME = 10_000; // Fallback SCTP message size per RFC 8841 / WebRTC spec when transport is not yet connected const DEFAULT_MAX_MESSAGE_SIZE = 65_536; // 64 KiB -export class Packet { - constructor( - public readonly messageID: MessageID, - public readonly chunkID: ChunkID, - public readonly totalChunks?: number, - public readonly payload: Buffer = Buffer.alloc(0), - ) {} - - get isAck(): boolean { - return this.chunkID === ACK_CHUNK_ID; - } - - get isFirstChunk(): boolean { - return this.chunkID === asChunkID(0); - } - - static ack(messageID: MessageID): Packet { - return new Packet(messageID, ACK_CHUNK_ID); - } - - static firstChunk( - messageID: MessageID, - totalChunks: number, - payload: Buffer, - ): Packet { - return new Packet(messageID, asChunkID(0), totalChunks, payload); - } - - static chunk( - messageID: MessageID, - chunkID: ChunkID, - payload: Buffer, - ): Packet { - return new Packet(messageID, chunkID, undefined, payload); - } - - encode(): Buffer { - if (this.isAck) { - const buf = Buffer.alloc(HEADER_SIZE); - buf.writeUInt16BE(this.messageID, 0); - buf.writeUInt32BE(ACK_CHUNK_ID, MESSAGE_ID_BYTES); - return buf; - } - - if (this.isFirstChunk) { - if (this.totalChunks === undefined) { - throw new Error("first chunk packet requires totalChunks"); - } - const buf = Buffer.alloc(FIRST_HEADER_SIZE + this.payload.length); - buf.writeUInt16BE(this.messageID, 0); - buf.writeUInt32BE(asChunkID(0), MESSAGE_ID_BYTES); - buf.writeUInt32BE(this.totalChunks, HEADER_SIZE); - this.payload.copy(buf, FIRST_HEADER_SIZE, 0); - return buf; - } - - const buf = Buffer.alloc(HEADER_SIZE + this.payload.length); - buf.writeUInt16BE(this.messageID, 0); - buf.writeUInt32BE(this.chunkID, MESSAGE_ID_BYTES); - this.payload.copy(buf, HEADER_SIZE, 0); - return buf; - } - - static decode(data: Buffer): Packet { - if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { - throw new Error("received invalid message type"); - } - - const messageID: MessageID = data.readUInt16BE(0); - const chunkID: ChunkID = data.readUInt32BE(MESSAGE_ID_BYTES); - - if (chunkID === ACK_CHUNK_ID) { - return Packet.ack(messageID); - } - - if (chunkID === asChunkID(0)) { - if (data.length < FIRST_HEADER_SIZE) { - throw new Error("received invalid message type"); - } - const totalChunks = data.readUInt32BE(HEADER_SIZE); - const payload = Buffer.alloc(data.length - FIRST_HEADER_SIZE); - data.copy(payload, 0, FIRST_HEADER_SIZE); - return Packet.firstChunk(messageID, totalChunks, payload); - } - - const payload = Buffer.alloc(data.length - HEADER_SIZE); - data.copy(payload, 0, HEADER_SIZE); - return Packet.chunk(messageID, chunkID, payload); - } -} - // we can't use the definition in DOM as we're platform independent export type SignalData = | { type: "answer" | "offer" | "pranswer" | "rollback"; sdp?: string } From 7fb98108121f15d2a294a471f418a6b0d036a169 Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 18:23:45 +0200 Subject: [PATCH 05/10] Use standard RTCSessionDescriptionInit in SignalData Simplify SignalData to use W3C standard RTCSessionDescriptionInit and drop obsolete simple-peer internal transceiverRequest and renegotiate types. --- discojs/src/client/decentralized/peer.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index 4078bffb5..55626088b 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -22,11 +22,8 @@ const MAX_DRAIN_TIME = 10_000; // Fallback SCTP message size per RFC 8841 / WebRTC spec when transport is not yet connected const DEFAULT_MAX_MESSAGE_SIZE = 65_536; // 64 KiB -// we can't use the definition in DOM as we're platform independent export type SignalData = - | { type: "answer" | "offer" | "pranswer" | "rollback"; sdp?: string } - | { type: "transceiverRequest"; transceiverRequest: { kind: string } } - | { type: "renegotiate"; renegotiate: true } + | RTCSessionDescriptionInit | { type: "candidate"; candidate: RTCIceCandidateInit }; type EventPayloads = { From 30decf4cc34a5d5bc95de074f4cd6160192042d2 Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 19:43:41 +0200 Subject: [PATCH 06/10] Refactor some stuff --- discojs/src/client/decentralized/packet.ts | 77 +++++++++--------- discojs/src/client/decentralized/peer.ts | 90 +++++++++------------- discojs/src/client/event_connection.ts | 2 +- 3 files changed, 76 insertions(+), 93 deletions(-) diff --git a/discojs/src/client/decentralized/packet.ts b/discojs/src/client/decentralized/packet.ts index 1fa9d80a5..7ed2fc018 100644 --- a/discojs/src/client/decentralized/packet.ts +++ b/discojs/src/client/decentralized/packet.ts @@ -2,23 +2,7 @@ export type MessageID = number; export type ChunkID = number; // helper function to make it evident that the number passed is a chunk id -export const asChunkID = (n: number): ChunkID => n; - -// message id size in bytes (uint16) -export const MESSAGE_ID_BYTES = 2; -// chunk counter size in bytes (uint32) -export const CHUNK_ID_BYTES = 4; -// total chunk count size in bytes (uint32) -export const TOTAL_CHUNKS_BYTES = 4; - -// message id (2 bytes) + chunk counter (4 bytes) -export const HEADER_SIZE = MESSAGE_ID_BYTES + CHUNK_ID_BYTES; - -// message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) -export const FIRST_HEADER_SIZE = HEADER_SIZE + TOTAL_CHUNKS_BYTES; - -// chunk counter of an acknowledgement, unused by messages -export const ACK_CHUNK_ID: ChunkID = 0xffff_ffff; +const asChunkID = (n: number): ChunkID => n; export class Packet { constructor( @@ -26,10 +10,26 @@ export class Packet { public readonly chunkID: ChunkID, public readonly totalChunks?: number, public readonly payload: Buffer = Buffer.alloc(0), - ) {} + ) { } + + // message id size in bytes (uint16) + static MESSAGE_ID_BYTES = 2; + // chunk counter size in bytes (uint32) + static CHUNK_ID_BYTES = 4; + // total chunk count size in bytes (uint32) + static TOTAL_CHUNKS_BYTES = 4; + + // message id (2 bytes) + chunk counter (4 bytes) + static HEADER_SIZE = Packet.MESSAGE_ID_BYTES + Packet.CHUNK_ID_BYTES; + + // message id (2 bytes) + chunk counter (4 bytes) + total chunk count (4 bytes) + static FIRST_HEADER_SIZE = Packet.HEADER_SIZE + Packet.TOTAL_CHUNKS_BYTES; + + // chunk counter of an acknowledgement, unused by messages + static ACK_CHUNK_ID: ChunkID = 0xffff_ffff; get isAck(): boolean { - return this.chunkID === ACK_CHUNK_ID; + return this.chunkID === Packet.ACK_CHUNK_ID; } get isFirstChunk(): boolean { @@ -37,7 +37,7 @@ export class Packet { } static ack(messageID: MessageID): Packet { - return new Packet(messageID, ACK_CHUNK_ID); + return new Packet(messageID, Packet.ACK_CHUNK_ID); } static firstChunk( @@ -58,9 +58,9 @@ export class Packet { encode(): Buffer { if (this.isAck) { - const buf = Buffer.alloc(HEADER_SIZE); + const buf = Buffer.alloc(Packet.HEADER_SIZE); buf.writeUInt16BE(this.messageID, 0); - buf.writeUInt32BE(ACK_CHUNK_ID, MESSAGE_ID_BYTES); + buf.writeUInt32BE(Packet.ACK_CHUNK_ID, Packet.MESSAGE_ID_BYTES); return buf; } @@ -68,45 +68,46 @@ export class Packet { if (this.totalChunks === undefined) { throw new Error("first chunk packet requires totalChunks"); } - const buf = Buffer.alloc(FIRST_HEADER_SIZE + this.payload.length); + const buf = Buffer.alloc(Packet.FIRST_HEADER_SIZE + this.payload.length); buf.writeUInt16BE(this.messageID, 0); - buf.writeUInt32BE(asChunkID(0), MESSAGE_ID_BYTES); - buf.writeUInt32BE(this.totalChunks, HEADER_SIZE); - this.payload.copy(buf, FIRST_HEADER_SIZE, 0); + buf.writeUInt32BE(asChunkID(0), Packet.MESSAGE_ID_BYTES); + buf.writeUInt32BE(this.totalChunks, Packet.HEADER_SIZE); + this.payload.copy(buf, Packet.FIRST_HEADER_SIZE, 0); return buf; } - const buf = Buffer.alloc(HEADER_SIZE + this.payload.length); + const buf = Buffer.alloc(Packet.HEADER_SIZE + this.payload.length); buf.writeUInt16BE(this.messageID, 0); - buf.writeUInt32BE(this.chunkID, MESSAGE_ID_BYTES); - this.payload.copy(buf, HEADER_SIZE, 0); + buf.writeUInt32BE(this.chunkID, Packet.MESSAGE_ID_BYTES); + this.payload.copy(buf, Packet.HEADER_SIZE, 0); return buf; } + static decode(data: Buffer): Packet { - if (!Buffer.isBuffer(data) || data.length < HEADER_SIZE) { + if (!Buffer.isBuffer(data) || data.length < Packet.HEADER_SIZE) { throw new Error("received invalid message type"); } const messageID: MessageID = data.readUInt16BE(0); - const chunkID: ChunkID = data.readUInt32BE(MESSAGE_ID_BYTES); + const chunkID: ChunkID = data.readUInt32BE(Packet.MESSAGE_ID_BYTES); - if (chunkID === ACK_CHUNK_ID) { + if (chunkID === Packet.ACK_CHUNK_ID) { return Packet.ack(messageID); } if (chunkID === asChunkID(0)) { - if (data.length < FIRST_HEADER_SIZE) { + if (data.length < Packet.FIRST_HEADER_SIZE) { throw new Error("received invalid message type"); } - const totalChunks = data.readUInt32BE(HEADER_SIZE); - const payload = Buffer.alloc(data.length - FIRST_HEADER_SIZE); - data.copy(payload, 0, FIRST_HEADER_SIZE); + const totalChunks = data.readUInt32BE(Packet.HEADER_SIZE); + const payload = Buffer.alloc(data.length - Packet.FIRST_HEADER_SIZE); + data.copy(payload, 0, Packet.FIRST_HEADER_SIZE); return Packet.firstChunk(messageID, totalChunks, payload); } - const payload = Buffer.alloc(data.length - HEADER_SIZE); - data.copy(payload, 0, HEADER_SIZE); + const payload = Buffer.alloc(data.length - Packet.HEADER_SIZE); + data.copy(payload, 0, Packet.HEADER_SIZE); return Packet.chunk(messageID, chunkID, payload); } } diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index 55626088b..4e133f076 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -6,10 +6,6 @@ import { EventEmitter } from "#utils/event_emitter"; import { type MessageID, type ChunkID, - asChunkID, - HEADER_SIZE, - FIRST_HEADER_SIZE, - ACK_CHUNK_ID, Packet, } from "#client/decentralized/packet"; @@ -20,7 +16,7 @@ const TICK = 10; const MAX_DRAIN_TIME = 10_000; // Fallback SCTP message size per RFC 8841 / WebRTC spec when transport is not yet connected -const DEFAULT_MAX_MESSAGE_SIZE = 65_536; // 64 KiB +const SAFE_MAX_MESSAGE_SIZE = 65_536; export type SignalData = | RTCSessionDescriptionInit @@ -53,7 +49,7 @@ export class Peer extends EventEmitter { private queuedCandidates: RTCIceCandidateInit[] = []; private sendCounter: MessageID = 0; - private sendQueue = List(); + private sendQueue = List(); private unacknowledged = Set(); private receiving = Map< @@ -66,7 +62,7 @@ export class Peer extends EventEmitter { constructor( public readonly id: NodeID, - initiator: boolean = false, + isInitiator: boolean = false, ) { super(); @@ -74,28 +70,19 @@ export class Peer extends EventEmitter { this.pc.onicecandidate = (event) => { if (event.candidate) { - const candidate: RTCIceCandidateInit = event.candidate.toJSON - ? event.candidate.toJSON() - : { - candidate: event.candidate.candidate, - sdpMid: event.candidate.sdpMid, - sdpMLineIndex: event.candidate.sdpMLineIndex, - usernameFragment: event.candidate.usernameFragment, - }; + const candidate: RTCIceCandidateInit = event.candidate.toJSON() this.emit("signal", { type: "candidate", candidate }); } }; this.pc.onconnectionstatechange = () => { - if ( - this.pc.connectionState === "failed" || - this.pc.connectionState === "closed" - ) { + const state = this.pc.connectionState; + if (state === "failed" || state === "closed") { this.emit("close", undefined); } }; - if (initiator) { + if (isInitiator) { const channel = this.pc.createDataChannel("disco"); this.setupChannel(channel); queueMicrotask(() => { @@ -225,7 +212,7 @@ export class Peer extends EventEmitter { } private acknowledge(messageID: MessageID): void { - const ack = Packet.ack(messageID).encode(); + const ack = Packet.ack(messageID); this.sendQueue = this.sendQueue.push(ack); this.flush(); } @@ -250,9 +237,9 @@ export class Peer extends EventEmitter { const chunk = this.sendQueue.first(); if (chunk === undefined) break; - this.sendQueue = this.sendQueue.shift(); try { - (this.channel.send as (data: unknown) => void)(chunk); + this.channel.send(chunk.encode() as ArrayBufferView); + this.sendQueue = this.sendQueue.shift(); } catch (err) { this.emit( "error", @@ -264,19 +251,10 @@ export class Peer extends EventEmitter { } get maxChunkSize(): number { - const sctpLimit = this.pc.sctp?.maxMessageSize; - if ( - typeof sctpLimit === "number" && - sctpLimit > 0 && - Number.isFinite(sctpLimit) - ) { - return sctpLimit; - } - - return DEFAULT_MAX_MESSAGE_SIZE; + return this.pc.sctp?.maxMessageSize || SAFE_MAX_MESSAGE_SIZE; } - private chunk(b: Buffer): Seq.Indexed { + private chunk(b: Buffer): Seq.Indexed { const messageID = this.sendCounter; this.sendCounter++; if (this.sendCounter > 0xffff) { @@ -284,44 +262,44 @@ export class Peer extends EventEmitter { } let tail = Seq.Indexed([]); - if (b.length > this.maxChunkSize - FIRST_HEADER_SIZE) { + if (b.length > this.maxChunkSize - Packet.FIRST_HEADER_SIZE) { tail = Range( - this.maxChunkSize - FIRST_HEADER_SIZE, + this.maxChunkSize - Packet.FIRST_HEADER_SIZE, b.length, - this.maxChunkSize - HEADER_SIZE, + this.maxChunkSize - Packet.HEADER_SIZE, ).map((offset) => - b.subarray(offset, offset + this.maxChunkSize - HEADER_SIZE), + b.subarray(offset, offset + this.maxChunkSize - Packet.HEADER_SIZE), ); } const totalChunkCount = 1 + tail.count(); - if (totalChunkCount >= ACK_CHUNK_ID) { + if (totalChunkCount >= Packet.ACK_CHUNK_ID) { throw new Error( - `The payload is too big: ${totalChunkCount} chunks exceeds max chunks (${ACK_CHUNK_ID})`, + `The payload is too big: ${totalChunkCount} chunks exceeds max chunks (${Packet.ACK_CHUNK_ID})`, ); } const firstPayload = b.subarray( 0, - Math.min(b.length, this.maxChunkSize - FIRST_HEADER_SIZE), + Math.min(b.length, this.maxChunkSize - Packet.FIRST_HEADER_SIZE), ); - const firstChunk = Packet.firstChunk( + const headerChunk = Packet.firstChunk( messageID, totalChunkCount, firstPayload, - ).encode(); + ); - return Seq.Indexed([firstChunk]).concat( - Range(asChunkID(1), Number.POSITIVE_INFINITY) + return Seq.Indexed([headerChunk]).concat( + Range(1, Number.POSITIVE_INFINITY) .zip(tail) - .map(([id, raw]) => Packet.chunk(messageID, id, raw).encode()), + .map(([id, raw]) => Packet.chunk(messageID, id, raw)), ); } // Resolves once every message sent so far has been received, // as destroying the peer drops whatever is still in transit. // Gives up after a while in case the other peer is gone. - async drain(): Promise { + async waitAcks(): Promise { const deadline = Date.now() + MAX_DRAIN_TIME; while ( !this.isDestroyed && @@ -346,7 +324,7 @@ export class Peer extends EventEmitter { } this.emit("close", undefined); - await Promise.resolve(); + return Promise.resolve(); } private handleData(data: Buffer): void { @@ -369,20 +347,24 @@ export class Peer extends EventEmitter { ); } - if (packet.isFirstChunk) { + // Receive the header message + if (packet.isFirstChunk) { if (total !== undefined) { throw new Error("first header received twice"); } - if (packet.totalChunks === undefined) { + + const totalChunks = packet.totalChunks; + if (totalChunks === undefined) { throw new Error("first chunk packet missing totalChunks"); } - const totalLimit = packet.totalChunks; - total = totalLimit; - if (chunks.keySeq().some((id) => id >= totalLimit)) { + else if (chunks.keySeq().some((id) => id >= totalChunks)) { throw new Error( "received total of chunk but got now-out-of-bound chunks", ); } + + // Set the total number of chunks expected for this message ID + total = totalChunks; } this.receiving = this.receiving.set(packet.messageID, { @@ -402,11 +384,11 @@ export class Peer extends EventEmitter { .sortBy(([id]) => id), ) .map((chunks) => Buffer.concat(chunks.map(([_, b]) => b).toArray())); - this.receiving = this.receiving.deleteAll(readyMessages.keys()); readyMessages.forEach((message, messageID) => { this.acknowledge(messageID); this.emit("data", message); + this.receiving = this.receiving.delete(messageID); }); } } diff --git a/discojs/src/client/event_connection.ts b/discojs/src/client/event_connection.ts index d5ddd787d..1909caa44 100644 --- a/discojs/src/client/event_connection.ts +++ b/discojs/src/client/event_connection.ts @@ -201,7 +201,7 @@ export class PeerConnection // Resolves once the messages sent so far have left the connection async drain(): Promise { - await this.#peer.drain(); + await this.#peer.waitAcks(); } async disconnect(): Promise { From 8d1fb45c9ec1d5d54c072ac5a8e6af10c457e82d Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 21:37:48 +0200 Subject: [PATCH 07/10] Update peer.spec.ts to use waitAcks instead of drain --- discojs/src/client/decentralized/peer.spec.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/discojs/src/client/decentralized/peer.spec.ts b/discojs/src/client/decentralized/peer.spec.ts index 67916c5b4..1f48182e0 100644 --- a/discojs/src/client/decentralized/peer.spec.ts +++ b/discojs/src/client/decentralized/peer.spec.ts @@ -64,7 +64,7 @@ describe("peer", () => { assert.deepStrictEqual(receiveds.toArray(), messages.toArray()); }); - it("drains once the other peer received the message", async () => { + it("waits for acks once the other peer received the message", async () => { // big enough to be split into chunks and still in transit // after leaving the data channel's buffer const message = Buffer.alloc(4 * 1024 * 1024); @@ -74,7 +74,7 @@ describe("peer", () => { }); peer1.send(message); - await peer1.drain(); + await peer1.waitAcks(); assert.isTrue(received); }); @@ -93,7 +93,7 @@ describe("peer", () => { }); peer1.send(message); - await peer1.drain(); + await peer1.waitAcks(); const received = await receivedPromise; assert.strictEqual(received.length, message.length); @@ -116,7 +116,7 @@ describe("peer", () => { }); peer1.send(message); - await peer1.drain(); + await peer1.waitAcks(); const received = await receivedPromise; assert.strictEqual(received.length, message.length); From 2e31e559887db39151c38a4cfce943c522c92f3e Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 21:46:37 +0200 Subject: [PATCH 08/10] Move drain in disconnect function --- discojs/src/client/decentralized/peer.ts | 4 ++-- discojs/src/client/decentralized/peer_pool.ts | 4 ---- discojs/src/client/event_connection.ts | 9 +++------ 3 files changed, 5 insertions(+), 12 deletions(-) diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index 4e133f076..2da11e7ec 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -250,7 +250,7 @@ export class Peer extends EventEmitter { } } - get maxChunkSize(): number { + private get maxChunkSize(): number { return this.pc.sctp?.maxMessageSize || SAFE_MAX_MESSAGE_SIZE; } @@ -299,7 +299,7 @@ export class Peer extends EventEmitter { // Resolves once every message sent so far has been received, // as destroying the peer drops whatever is still in transit. // Gives up after a while in case the other peer is gone. - async waitAcks(): Promise { + public async drain(): Promise { const deadline = Date.now() + MAX_DRAIN_TIME; while ( !this.isDestroyed && diff --git a/discojs/src/client/decentralized/peer_pool.ts b/discojs/src/client/decentralized/peer_pool.ts index b208f7c45..551fb0c14 100644 --- a/discojs/src/client/decentralized/peer_pool.ts +++ b/discojs/src/client/decentralized/peer_pool.ts @@ -41,10 +41,6 @@ export class PeerPool { async shutdown(): Promise { debug(`[${this.id}] is shutting down all its connections`); - // Let the messages still being sent, such as our last round's weights, - // reach the other peers before closing the connections - await Promise.all(this.peers.valueSeq().map((peer) => peer.drain())); - // Add a timeout o.w. the promise hangs forever if the other peer is already disconnected await Promise.race([ Promise.all(this.peers.valueSeq().map((peer) => peer.disconnect())), diff --git a/discojs/src/client/event_connection.ts b/discojs/src/client/event_connection.ts index 1909caa44..5acdce97c 100644 --- a/discojs/src/client/event_connection.ts +++ b/discojs/src/client/event_connection.ts @@ -198,13 +198,10 @@ export class PeerConnection } this.#peer.send(Buffer.from(msgpack.encode(msg))); } - - // Resolves once the messages sent so far have left the connection - async drain(): Promise { - await this.#peer.waitAcks(); - } - + async disconnect(): Promise { + // Resolves once the messages sent so far have left the connection + await this.#peer.drain(); await this.#peer.destroy(); } } From 52842e64a4ffe04d6f8ded439452bd79ff4a115a Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 23:47:14 +0200 Subject: [PATCH 09/10] Add gracePeriod to PeerConnection.disconnect and drain during shutdown --- discojs/src/client/decentralized/peer.spec.ts | 8 ++++---- discojs/src/client/decentralized/peer.ts | 12 ++++++------ discojs/src/client/decentralized/peer_pool.ts | 14 +++++++++----- discojs/src/client/event_connection.ts | 11 ++++++----- 4 files changed, 25 insertions(+), 20 deletions(-) diff --git a/discojs/src/client/decentralized/peer.spec.ts b/discojs/src/client/decentralized/peer.spec.ts index 1f48182e0..67916c5b4 100644 --- a/discojs/src/client/decentralized/peer.spec.ts +++ b/discojs/src/client/decentralized/peer.spec.ts @@ -64,7 +64,7 @@ describe("peer", () => { assert.deepStrictEqual(receiveds.toArray(), messages.toArray()); }); - it("waits for acks once the other peer received the message", async () => { + it("drains once the other peer received the message", async () => { // big enough to be split into chunks and still in transit // after leaving the data channel's buffer const message = Buffer.alloc(4 * 1024 * 1024); @@ -74,7 +74,7 @@ describe("peer", () => { }); peer1.send(message); - await peer1.waitAcks(); + await peer1.drain(); assert.isTrue(received); }); @@ -93,7 +93,7 @@ describe("peer", () => { }); peer1.send(message); - await peer1.waitAcks(); + await peer1.drain(); const received = await receivedPromise; assert.strictEqual(received.length, message.length); @@ -116,7 +116,7 @@ describe("peer", () => { }); peer1.send(message); - await peer1.waitAcks(); + await peer1.drain(); const received = await receivedPromise; assert.strictEqual(received.length, message.length); diff --git a/discojs/src/client/decentralized/peer.ts b/discojs/src/client/decentralized/peer.ts index 2da11e7ec..3ce8659f6 100644 --- a/discojs/src/client/decentralized/peer.ts +++ b/discojs/src/client/decentralized/peer.ts @@ -70,7 +70,7 @@ export class Peer extends EventEmitter { this.pc.onicecandidate = (event) => { if (event.candidate) { - const candidate: RTCIceCandidateInit = event.candidate.toJSON() + const candidate: RTCIceCandidateInit = event.candidate.toJSON(); this.emit("signal", { type: "candidate", candidate }); } }; @@ -299,10 +299,11 @@ export class Peer extends EventEmitter { // Resolves once every message sent so far has been received, // as destroying the peer drops whatever is still in transit. // Gives up after a while in case the other peer is gone. - public async drain(): Promise { - const deadline = Date.now() + MAX_DRAIN_TIME; + public async drain(timeout: number = MAX_DRAIN_TIME): Promise { + const deadline = Date.now() + timeout; while ( !this.isDestroyed && + this.channel?.readyState === "open" && !this.unacknowledged.isEmpty() && Date.now() < deadline ) { @@ -348,7 +349,7 @@ export class Peer extends EventEmitter { } // Receive the header message - if (packet.isFirstChunk) { + if (packet.isFirstChunk) { if (total !== undefined) { throw new Error("first header received twice"); } @@ -356,8 +357,7 @@ export class Peer extends EventEmitter { const totalChunks = packet.totalChunks; if (totalChunks === undefined) { throw new Error("first chunk packet missing totalChunks"); - } - else if (chunks.keySeq().some((id) => id >= totalChunks)) { + } else if (chunks.keySeq().some((id) => id >= totalChunks)) { throw new Error( "received total of chunk but got now-out-of-bound chunks", ); diff --git a/discojs/src/client/decentralized/peer_pool.ts b/discojs/src/client/decentralized/peer_pool.ts index 551fb0c14..7e1e60573 100644 --- a/discojs/src/client/decentralized/peer_pool.ts +++ b/discojs/src/client/decentralized/peer_pool.ts @@ -26,6 +26,9 @@ async function createStaggered(create: () => T): Promise { return create(); } +// Maximum time in ms to wait for pending messages to be delivered and acknowledged during shutdown +const SHUTDOWN_GRACE_PERIOD = 10_000; + export class PeerPool { private peers = Map(); @@ -41,11 +44,12 @@ export class PeerPool { async shutdown(): Promise { debug(`[${this.id}] is shutting down all its connections`); - // Add a timeout o.w. the promise hangs forever if the other peer is already disconnected - await Promise.race([ - Promise.all(this.peers.valueSeq().map((peer) => peer.disconnect())), - new Promise((res, _) => setTimeout(res, 1000)), // Wait for other peers to finish - ]); + // Let messages still being sent reach the other peers before closing + await Promise.all( + this.peers + .valueSeq() + .map((peer) => peer.disconnect(SHUTDOWN_GRACE_PERIOD)), + ); this.peers = Map(); this.pendingSignals = Map(); } diff --git a/discojs/src/client/event_connection.ts b/discojs/src/client/event_connection.ts index 5acdce97c..0cb22358d 100644 --- a/discojs/src/client/event_connection.ts +++ b/discojs/src/client/event_connection.ts @@ -34,7 +34,7 @@ export interface EventConnection { handler: (event: NarrowMessage) => void, ) => void; send: (msg: T) => void; - disconnect: () => Promise; + disconnect: (gracePeriod?: number) => Promise; } /** @@ -198,10 +198,11 @@ export class PeerConnection } this.#peer.send(Buffer.from(msgpack.encode(msg))); } - - async disconnect(): Promise { - // Resolves once the messages sent so far have left the connection - await this.#peer.drain(); + + async disconnect(gracePeriod = 0): Promise { + if (gracePeriod > 0) { + await this.#peer.drain(gracePeriod); + } await this.#peer.destroy(); } } From 3c3a1646c145a7c8e31403c564a25c369cc35d23 Mon Sep 17 00:00:00 2001 From: ale18V Date: Tue, 20 Oct 2026 23:50:36 +0200 Subject: [PATCH 10/10] Format packet.ts with Prettier --- discojs/src/client/decentralized/packet.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/discojs/src/client/decentralized/packet.ts b/discojs/src/client/decentralized/packet.ts index 7ed2fc018..f18255608 100644 --- a/discojs/src/client/decentralized/packet.ts +++ b/discojs/src/client/decentralized/packet.ts @@ -10,7 +10,7 @@ export class Packet { public readonly chunkID: ChunkID, public readonly totalChunks?: number, public readonly payload: Buffer = Buffer.alloc(0), - ) { } + ) {} // message id size in bytes (uint16) static MESSAGE_ID_BYTES = 2; @@ -83,7 +83,6 @@ export class Packet { return buf; } - static decode(data: Buffer): Packet { if (!Buffer.isBuffer(data) || data.length < Packet.HEADER_SIZE) { throw new Error("received invalid message type");