🐛 Bug Report — Runtime APIs: pipeTo() from a native body into a socket writable never releases its locks with typescript_implemented_streams #8049

Description

With typescript_implemented_streams, when a native or buffer-backed ReadableStream (a fetch() or service-binding body, new Response("…").body, Blob.stream(), another socket's readable) is piped into a connect() socket's writable, pipeTo() never releases either lock. Once the pipe settles, source.locked and socket.writable.locked both stay true forever. This happens whether the pipe resolved, rejected because the source errored, or was aborted through signal.

As a result, socket.writable.close() and socket.writable.abort() reject with Cannot close a stream that is locked / Cannot abort a stream that is locked, getWriter() throws, and the source can never be read or cancelled again. Per the spec, pipeTo() releases its reader and writer when it finalizes. The C++ implementation does this, and so does the TypeScript one on its regular pipe path.

Only the direct native-to-native path is affected:

• With a write still pending on the socket when the pipe starts (the shape pinned by pipeBehindUnawaitedWrite in src/tests/streams/sockets/socket-streams.js), both endpoints are unlocked afterwards.
• With preventClose: true, both endpoints are unlocked afterwards.
• A JavaScript ReadableStream piped into the socket, or a native body piped into an IdentityTransformStream, releases both locks.

Reproduction

worker.js: a worker that sends a raw HTTP request to its own listening socket.

import { connect } from "cloudflare:sockets";

export default {
  async fetch(request) {
    const url = new URL(request.url);
    if (url.pathname === "/target") return new Response("ok");

    // Pipe a non-JS body into a TCP socket's writable, here a raw HTTP
    // request to this same server.
    const socket = connect(url.host);
    const body = new Response(`GET /target HTTP/1.1\r\nHost: ${url.host}\r\nConnection: close\r\n\r\n`).body;
    await body.pipeTo(socket.writable);
    const response = await new Response(socket.readable).text();

    return Response.json({
      response: response.split("\r\n")[0],
      sourceLocked: body.locked,
      writableLocked: socket.writable.locked,
    });
  },
};

config.capnp

using Workerd = import "/workerd/workerd.capnp";

const config :Workerd.Config = (
  services = [
    (name = "main", worker = .main),
    (name = "internet", network = (allow = ["local"])),
  ],
  sockets = [(name = "http", address = "127.0.0.1:18982", http = (), service = "main")],
  # typescript_implemented_streams only starts with this autogate.
  autogates = ["workerd-autogate-per-isolate-javascript-bootstrap"],
);

const main :Workerd.Worker = (
  modules = [(name = "worker", esModule = embed "worker.js")],
  compatibilityDate = "2026-10-27",
  compatibilityFlags = ["typescript_implemented_streams"],
  globalOutbound = "internet",
);

workerd serve --experimental config.capnp &
curl http://127.0.0.1:18982/

With the flag:

{"response":"HTTP/1.1 200 OK","sourceLocked":true,"writableLocked":true}

With compatibilityFlags = []:

{"response":"HTTP/1.1 200 OK","sourceLocked":false,"writableLocked":false}

Tested with workerd 1.20261027.1 on macOS arm64.

Expected behavior

Once pipeTo() settles, the source and the socket writable are unlocked, as they are without the flag and on the non-native pipe path.