Replace simple-peer with native WebRTC API

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:"
   }

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)));
+  });
+});

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
+const asChunkID = (n: number): ChunkID => n;
+
+export class Packet {
+  constructor(
+    public readonly messageID: MessageID,
+    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 === Packet.ACK_CHUNK_ID;
+  }
+
+  get isFirstChunk(): boolean {
+    return this.chunkID === asChunkID(0);
+  }
+
+  static ack(messageID: MessageID): Packet {
+    return new Packet(messageID, Packet.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(Packet.HEADER_SIZE);
+      buf.writeUInt16BE(this.messageID, 0);
+      buf.writeUInt32BE(Packet.ACK_CHUNK_ID, Packet.MESSAGE_ID_BYTES);
+      return buf;
+    }
+
+    if (this.isFirstChunk) {
+      if (this.totalChunks === undefined) {
+        throw new Error("first chunk packet requires totalChunks");
+      }
+      const buf = Buffer.alloc(Packet.FIRST_HEADER_SIZE + this.payload.length);
+      buf.writeUInt16BE(this.messageID, 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(Packet.HEADER_SIZE + this.payload.length);
+    buf.writeUInt16BE(this.messageID, 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 < Packet.HEADER_SIZE) {
+      throw new Error("received invalid message type");
+    }
+
+    const messageID: MessageID = data.readUInt16BE(0);
+    const chunkID: ChunkID = data.readUInt32BE(Packet.MESSAGE_ID_BYTES);
+
+    if (chunkID === Packet.ACK_CHUNK_ID) {
+      return Packet.ack(messageID);
+    }
+
+    if (chunkID === asChunkID(0)) {
+      if (data.length < Packet.FIRST_HEADER_SIZE) {
+        throw new Error("received invalid message type");
+      }
+      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 - Packet.HEADER_SIZE);
+    data.copy(payload, 0, Packet.HEADER_SIZE);
+    return Packet.chunk(messageID, chunkID, payload);
+  }
+}

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<Buffer>((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<Buffer>((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);
+  });
 });