More complete fix attempt for WebSocket message repetition

- Also removed receive_body_remain modification from clearReceiveBuffers
- Issue still persists, need deeper analysis of the message loop

reproduce-21376-smaller.ts

@@ -0,0 +1,79 @@
+import { WebSocketServer } from "ws";
+import { randomBytes, randomUUID } from "crypto";
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+const LOCAL_PORT = 6667;
+
+const wss = new WebSocketServer({
+  port: LOCAL_PORT,
+});
+
+wss.on("connection", (ws) => {
+  console.log("server: client connected");
+
+  ws.on("close", () => {
+    console.log("server: client disconnected");
+  });
+
+  const sendData = (megabytes: number = 10) => {
+    const payload = {
+      id: randomUUID(),
+      data: randomBytes(1024 * 1024 * megabytes).toString("base64"),
+    };
+
+    ws.send(JSON.stringify(payload));
+
+    console.log(`server: sent ${megabytes} MB (${payload.id}) ${Date.now()}`);
+  };
+
+  // Use smaller payloads to avoid crash but still trigger the repetition bug
+  sendData(2); // send 2 MB to client
+  sendData(1); // send 1 MB to client
+});
+
+wss.on("listening", () => {
+  console.log(`server: listening on port ${LOCAL_PORT}`);
+});
+
+await sleep(1000); // just to make sure the server is ready
+
+const ws = new WebSocket(`ws://localhost:${LOCAL_PORT}`);
+
+const receivedMessages = new Set<string>();
+let messageCount = 0;
+
+ws.addEventListener("open", () => {
+  console.log("client: connected to server");
+});
+
+ws.addEventListener("message", (event) => {
+  messageCount++;
+  const parsed = JSON.parse(event.data.toString());
+
+  const messageInMb =
+    Buffer.byteLength(event.data.toString(), "utf8") / (1024 * 1024); // aproximate size in MB
+
+  console.log(
+    `client: received ${parsed.id} (${messageInMb.toFixed(2)} MB) ${Date.now()} [count: ${messageCount}]`
+  );
+  
+  if (receivedMessages.has(parsed.id)) {
+    console.error(`ERROR: Duplicate message received: ${parsed.id}`);
+    process.exit(1);  
+  }
+  
+  receivedMessages.add(parsed.id);
+  
+  // Expected 2 unique messages
+  if (receivedMessages.size === 2) {
+    console.log("SUCCESS: All messages received without duplication");
+    process.exit(0);
+  }
+});
+
+// Timeout after 10 seconds
+setTimeout(() => {
+  console.log(`TIMEOUT: Received ${receivedMessages.size} unique messages out of 2 expected (${messageCount} total)`);
+  process.exit(1);
+}, 10000);

reproduce-21376.ts

@@ -0,0 +1,55 @@
+import { WebSocketServer } from "ws";
+import { randomBytes, randomUUID } from "crypto";
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+const LOCAL_PORT = 6666;
+
+const wss = new WebSocketServer({
+  port: LOCAL_PORT,
+});
+
+wss.on("connection", (ws) => {
+  console.log("server: client connected");
+
+  ws.on("close", () => {
+    console.log("server: client disconnected");
+  });
+
+  const sendData = (megabytes: number = 10) => {
+    const payload = {
+      id: randomUUID(),
+      data: randomBytes(1024 * 1024 * megabytes).toString("base64"),
+    };
+
+    ws.send(JSON.stringify(payload));
+
+    console.log(`server: sent ${megabytes} MB (${payload.id}) ${Date.now()}`);
+  };
+
+  sendData(50); // send 50 MB to client
+  sendData(5); // send 5 MB to client
+});
+
+wss.on("listening", () => {
+  console.log(`server: listening on port ${LOCAL_PORT}`);
+});
+
+await sleep(1000); // just to make sure the server is ready
+
+const ws = new WebSocket(`ws://localhost:${LOCAL_PORT}`);
+
+ws.addEventListener("open", () => {
+  console.log("client: connected to server");
+});
+
+ws.addEventListener("message", (event) => {
+  const parsed = JSON.parse(event.data.toString());
+
+  const messageInMb =
+    Buffer.byteLength(event.data.toString(), "utf8") / (1024 * 1024); // aproximate size in MB
+
+  console.log(
+    `client: received ${parsed.id} (${messageInMb.toFixed(2)} MB) ${Date.now()}`
+  );
+});

src/http/websocket_client.zig

@@ -207,7 +207,7 @@ pub fn NewWebSocketClient(comptime ssl: bool) type {
             }
 
             this.receive_pending_chunk_len = 0;
-            this.receive_body_remain = 0;
+            // Don't modify this.receive_body_remain here as it's managed by the caller (handleData)
         }
 
         fn clearSendBuffers(this: *WebSocket, free: bool) void {
@@ -296,7 +296,7 @@ pub fn NewWebSocketClient(comptime ssl: bool) type {
 
                 if (left_in_fragment >= data_.len and left_in_fragment - data_.len - this.receive_pending_chunk_len == 0) {
                     this.receive_pending_chunk_len = 0;
-                    this.receive_body_remain = 0;
+                    // Don't modify this.receive_body_remain here as it's managed by the caller (handleData)
                     if (is_final) {
                         // Decompress the complete message
                         this.dispatchCompressedData(this.receive_buffer.readableSlice(0), kind);
@@ -335,7 +335,7 @@ pub fn NewWebSocketClient(comptime ssl: bool) type {
 
             if (left_in_fragment >= data_.len and left_in_fragment - data_.len - this.receive_pending_chunk_len == 0) {
                 this.receive_pending_chunk_len = 0;
-                this.receive_body_remain = 0;
+                // Don't modify this.receive_body_remain here as it's managed by the caller (handleData)
                 if (is_final) {
                     this.dispatchData(this.receive_buffer.readableSlice(0), kind);
                     this.clearReceiveBuffers(false);

test-bug-directly.ts

@@ -0,0 +1,67 @@
+// Test to specifically trigger the state management bug
+import { WebSocketServer } from "ws";
+import { randomBytes, randomUUID } from "crypto";
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+const LOCAL_PORT = 6668;
+
+const wss = new WebSocketServer({
+  port: LOCAL_PORT,
+});
+
+let messageCount = 0;
+const receivedIds = new Set<string>();
+
+wss.on("connection", (ws) => {
+  console.log("server: client connected");
+
+  const sendData = (megabytes: number) => {
+    const payload = {
+      id: randomUUID(),
+      data: randomBytes(1024 * 1024 * megabytes).toString("base64"),
+    };
+
+    ws.send(JSON.stringify(payload));
+    console.log(`server: sent ${megabytes} MB (${payload.id})`);
+    return payload.id;
+  };
+
+  // Send messages that are large enough to trigger the buffering/fragmentation logic
+  // but not so large they cause crashes
+  const id1 = sendData(5); // 5 MB  
+  const id2 = sendData(3); // 3 MB
+  
+  setTimeout(() => {
+    console.log(`Expected 2 messages, received ${messageCount} total, ${receivedIds.size} unique`);
+    if (messageCount > receivedIds.size) {
+      console.log("❌ BUG DETECTED: Message repetition occurred");
+      process.exit(1);
+    } else {
+      console.log("✅ No repetition detected");  
+      process.exit(0);
+    }
+  }, 8000);
+});
+
+wss.on("listening", () => {
+  console.log(`server: listening on port ${LOCAL_PORT}`);
+});
+
+await sleep(1000);
+
+const ws = new WebSocket(`ws://localhost:${LOCAL_PORT}`);
+
+ws.addEventListener("message", (event) => {
+  messageCount++;
+  const parsed = JSON.parse(event.data.toString());
+  const messageId = parsed.id;
+  
+  console.log(`client: received message ${messageCount}: ${messageId}`);
+  
+  if (receivedIds.has(messageId)) {
+    console.log("❌ DUPLICATE MESSAGE DETECTED:", messageId);
+  }
+  
+  receivedIds.add(messageId);
+});

test-bug-larger.ts

@@ -0,0 +1,66 @@
+// Test to specifically trigger the state management bug with larger payloads
+import { WebSocketServer } from "ws";
+import { randomBytes, randomUUID } from "crypto";
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+const LOCAL_PORT = 6669;
+
+const wss = new WebSocketServer({
+  port: LOCAL_PORT,
+});
+
+let messageCount = 0;
+const receivedIds = new Set<string>();
+
+wss.on("connection", (ws) => {
+  console.log("server: client connected");
+
+  const sendData = (megabytes: number) => {
+    const payload = {
+      id: randomUUID(),
+      data: randomBytes(1024 * 1024 * megabytes).toString("base64"),
+    };
+
+    ws.send(JSON.stringify(payload));
+    console.log(`server: sent ${megabytes} MB (${payload.id})`);
+    return payload.id;
+  };
+
+  // Send larger messages to trigger the buffering/fragmentation logic
+  const id1 = sendData(15); // 15 MB
+  const id2 = sendData(10); // 10 MB
+  
+  setTimeout(() => {
+    console.log(`Expected 2 messages, received ${messageCount} total, ${receivedIds.size} unique`);
+    if (messageCount > receivedIds.size) {
+      console.log("❌ BUG DETECTED: Message repetition occurred");
+      process.exit(1);
+    } else {
+      console.log("✅ No repetition detected");  
+      process.exit(0);
+    }
+  }, 15000); // Longer timeout for larger messages
+});
+
+wss.on("listening", () => {
+  console.log(`server: listening on port ${LOCAL_PORT}`);
+});
+
+await sleep(1000);
+
+const ws = new WebSocket(`ws://localhost:${LOCAL_PORT}`);
+
+ws.addEventListener("message", (event) => {
+  messageCount++;
+  const parsed = JSON.parse(event.data.toString());
+  const messageId = parsed.id;
+  
+  console.log(`client: received message ${messageCount}: ${messageId}`);
+  
+  if (receivedIds.has(messageId)) {
+    console.log("❌ DUPLICATE MESSAGE DETECTED:", messageId);
+  }
+  
+  receivedIds.add(messageId);
+});

test/regression/issue/21376.test.ts

@@ -0,0 +1,110 @@
+// Regression test for issue 21376: WebSocket client receives repeated messages when working with big payloads
+// https://github.com/oven-sh/bun/issues/21376
+
+import { test, expect } from "bun:test";
+import { WebSocketServer } from "ws";
+import { randomBytes, randomUUID } from "crypto";
+
+test("WebSocket client should not repeat messages with large payloads", async () => {
+  const port = 0; // Let the system assign a port
+  let serverPort: number;
+  
+  const wss = new WebSocketServer({ port });
+  
+  await new Promise<void>((resolve) => {
+    wss.on("listening", () => {
+      serverPort = (wss.address() as any).port;
+      resolve();
+    });
+  });
+
+  const receivedMessages = new Set<string>();
+  const expectedMessages = new Set<string>();
+  
+  const messagePromises: Promise<void>[] = [];
+  
+  // Use a promise to track when all expected messages are received
+  let resolveTest: () => void;
+  let rejectTest: (error: Error) => void;
+  const testPromise = new Promise<void>((resolve, reject) => {
+    resolveTest = resolve;
+    rejectTest = reject;
+  });
+
+  wss.on("connection", (ws) => {
+    const sendData = (megabytes: number) => {
+      const payload = {
+        id: randomUUID(),
+        data: randomBytes(1024 * 1024 * megabytes).toString("base64"),
+      };
+      
+      expectedMessages.add(payload.id);
+      ws.send(JSON.stringify(payload));
+    };
+
+    // Send multiple moderately large messages to trigger the bug without causing crashes
+    sendData(2); // 2 MB
+    sendData(1);  // 1 MB
+    sendData(3); // 3 MB
+  });
+
+  const client = new WebSocket(`ws://localhost:${serverPort}`);
+  
+  let timeoutId: Timer;
+  
+  client.addEventListener("open", () => {
+    // Set a timeout to resolve the test after a reasonable time
+    timeoutId = setTimeout(() => {
+      if (receivedMessages.size === expectedMessages.size) {
+        resolveTest();
+      } else {
+        rejectTest(new Error(`Expected ${expectedMessages.size} unique messages, but received ${receivedMessages.size}`));
+      }
+    }, 5000); // 5 second timeout
+  });
+
+  client.addEventListener("message", (event) => {
+    try {
+      const parsed = JSON.parse(event.data.toString());
+      const messageId = parsed.id;
+      
+      // Check if we've already received this message
+      if (receivedMessages.has(messageId)) {
+        clearTimeout(timeoutId);
+        client.close();
+        wss.close();
+        rejectTest(new Error(`Received duplicate message with ID: ${messageId}`));
+        return;
+      }
+      
+      receivedMessages.add(messageId);
+      
+      // Check if we've received all expected messages
+      if (receivedMessages.size === expectedMessages.size) {
+        clearTimeout(timeoutId);
+        client.close();
+        wss.close();
+        resolveTest();
+      }
+    } catch (error) {
+      clearTimeout(timeoutId);
+      client.close();
+      wss.close();
+      rejectTest(new Error(`Failed to parse message: ${error}`));
+    }
+  });
+
+  client.addEventListener("error", (error) => {
+    clearTimeout(timeoutId);
+    wss.close();
+    rejectTest(new Error(`WebSocket error: ${error}`));
+  });
+
+  await testPromise;
+  
+  // Verify we received exactly the expected messages
+  expect(receivedMessages.size).toBe(expectedMessages.size);
+  for (const expectedId of expectedMessages) {
+    expect(receivedMessages.has(expectedId)).toBe(true);
+  }
+}, 30000); // 30 second test timeout