fix(ext/node): simultaneous reads can leak into each other (#20223)

Reported in #20188

This was caused by re-use of a global buffer `BUF` during simultaneous
async reads.

cli/tests/unit_node/net_test.ts

@@ -130,3 +130,60 @@ Deno.test("[node/net] connection event has socket value", async () => {
 
   await Promise.all([p, p2]);
 });
+
+// https://github.com/denoland/deno/issues/20188
+Deno.test("[node/net] multiple Sockets should get correct server data", async () => {
+  const p = deferred();
+  const p2 = deferred();
+
+  const dataReceived1 = deferred();
+  const dataReceived2 = deferred();
+
+  const events1: string[] = [];
+  const events2: string[] = [];
+
+  const server = net.createServer();
+  server.on("connection", (socket) => {
+    assert(socket !== undefined);
+    socket.on("data", (data) => {
+      socket.write(new TextDecoder().decode(data));
+    });
+  });
+
+  server.listen(async () => {
+    // deno-lint-ignore no-explicit-any
+    const { port } = server.address() as any;
+
+    const socket1 = net.createConnection(port);
+    const socket2 = net.createConnection(port);
+
+    socket1.on("data", (data) => {
+      events1.push(new TextDecoder().decode(data));
+      dataReceived1.resolve();
+    });
+
+    socket2.on("data", (data) => {
+      events2.push(new TextDecoder().decode(data));
+      dataReceived2.resolve();
+    });
+
+    socket1.write("111");
+    socket2.write("222");
+
+    await Promise.all([dataReceived1, dataReceived2]);
+
+    socket1.end();
+    socket2.end();
+
+    server.close(() => {
+      p.resolve();
+    });
+
+    p2.resolve();
+  });
+
+  await Promise.all([p, p2]);
+
+  assertEquals(events1, ["111"]);
+  assertEquals(events2, ["222"]);
+});

ext/node/polyfills/internal_binding/stream_wrap.ts

@@ -311,56 +311,61 @@ export class LibuvStreamWrap extends HandleWrap {
 
   /** Internal method for reading from the attached stream. */
   async #read() {
-    let buf = BUF;
+    const isOwnedBuf = bufLocked;
-
+    let buf = bufLocked ? new Uint8Array(SUGGESTED_SIZE) : BUF;
-    let nread: number | null;
+    bufLocked = true;
-    const ridBefore = this[kStreamBaseField]!.rid;
     try {
-      nread = await this[kStreamBaseField]!.read(buf);
+      let nread: number | null;
-    } catch (e) {
+      const ridBefore = this[kStreamBaseField]!.rid;
-      // Try to read again if the underlying stream resource
+      try {
-      // changed. This can happen during TLS upgrades (eg. STARTTLS)
+        nread = await this[kStreamBaseField]!.read(buf);
-      if (ridBefore != this[kStreamBaseField]!.rid) {
+      } catch (e) {
-        return this.#read();
+        // Try to read again if the underlying stream resource
+        // changed. This can happen during TLS upgrades (eg. STARTTLS)
+        if (ridBefore != this[kStreamBaseField]!.rid) {
+          return this.#read();
+        }
+
+        if (
+          e instanceof Deno.errors.Interrupted ||
+          e instanceof Deno.errors.BadResource
+        ) {
+          nread = codeMap.get("EOF")!;
+        } else if (
+          e instanceof Deno.errors.ConnectionReset ||
+          e instanceof Deno.errors.ConnectionAborted
+        ) {
+          nread = codeMap.get("ECONNRESET")!;
+        } else {
+          nread = codeMap.get("UNKNOWN")!;
+        }
+
+        buf = new Uint8Array(0);
       }
 
-      if (
+      nread ??= codeMap.get("EOF")!;
-        e instanceof Deno.errors.Interrupted ||
+
-        e instanceof Deno.errors.BadResource
+      streamBaseState[kReadBytesOrError] = nread;
-      ) {
+
-        nread = codeMap.get("EOF")!;
+      if (nread > 0) {
-      } else if (
+        this.bytesRead += nread;
-        e instanceof Deno.errors.ConnectionReset ||
-        e instanceof Deno.errors.ConnectionAborted
-      ) {
-        nread = codeMap.get("ECONNRESET")!;
-      } else {
-        nread = codeMap.get("UNKNOWN")!;
       }
 
-      buf = new Uint8Array(0);
+      buf = isOwnedBuf ? buf.subarray(0, nread) : buf.slice(0, nread);
-    }
 
-    nread ??= codeMap.get("EOF")!;
+      streamBaseState[kArrayBufferOffset] = 0;
 
-    streamBaseState[kReadBytesOrError] = nread;
+      try {
+        this.onread!(buf, nread);
+      } catch {
+        // swallow callback errors.
+      }
 
-    if (nread > 0) {
+      if (nread >= 0 && this.#reading) {
-      this.bytesRead += nread;
+        this.#read();
-    }
+      }
-
+    } finally {
-    buf = buf.slice(0, nread);
+      bufLocked = false;
-
-    streamBaseState[kArrayBufferOffset] = 0;
-
-    try {
-      this.onread!(buf, nread);
-    } catch {
-      // swallow callback errors.
-    }
-
-    if (nread >= 0 && this.#reading) {
-      this.#read();
     }
   }
 
@@ -423,4 +428,7 @@ export class LibuvStreamWrap extends HandleWrap {
   }
 }
 
+// Used in #read above
 const BUF = new Uint8Array(SUGGESTED_SIZE);
+// We need to ensure that only one inflight read request uses the cached buffer above
+let bufLocked = false;