diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index b25166957614..e1e80eb953c0 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -46,10 +46,6 @@ const {
   DOMException,
 } = internalBinding('messaging');
 
-const {
-  markPromiseAsHandled,
-} = internalBinding('util');
-
 const {
   isArrayBufferView,
   isDataView,
@@ -139,7 +135,7 @@ const {
   writableStreamCloseQueuedOrInFlight,
   writableStreamDefaultWriterCloseWithErrorPropagation,
   writableStreamDefaultWriterRelease,
-  writableStreamDefaultWriterWrite,
+  writableStreamDefaultWriterWriteWithRequest,
   writerClosedPromise,
   writerReadyPromise,
 } = require('internal/webstreams/writablestream');
@@ -1530,8 +1526,38 @@ function readableStreamPipeTo(
 
   const promise = PromiseWithResolvers();
 
-  const state = {
-    currentWrite: PromiseResolve(),
+  // One shared write request tracks every chunk written to the
+  // destination, instead of a { promise, resolve, reject } record per
+  // write. `stall` is armed by waitForPendingWrites() during shutdown;
+  // `failed`/`failure` latch a write that could not proceed.
+  const writeTracker = {
+    // Non-undefined: queue entries are discriminated from kNilRequest
+    // by `promise === undefined`.
+    promise: null,
+    pending: 0,
+    failed: false,
+    failure: undefined,
+    stall: null,
+    resolve() {
+      if (--this.pending === 0 && this.stall !== null) {
+        const stall = this.stall;
+        this.stall = null;
+        if (this.failed)
+          stall.reject(this.failure);
+        else
+          stall.resolve();
+      }
+    },
+    reject(error) {
+      this.pending--;
+      this.failed = true;
+      this.failure = error;
+      if (this.stall !== null) {
+        const stall = this.stall;
+        this.stall = null;
+        stall.reject(error);
+      }
+    },
   };
 
   // The error here can be undefined. The rejected arg
@@ -1548,11 +1574,14 @@ function readableStreamPipeTo(
       promise.resolve();
   }
 
-  async function waitForCurrentWrite() {
-    const write = state.currentWrite;
-    await write;
-    if (write !== state.currentWrite)
-      await waitForCurrentWrite();
+  function waitForPendingWrites() {
+    if (writeTracker.pending === 0) {
+      return writeTracker.failed ?
+        PromiseReject(writeTracker.failure) :
+        PromiseResolve();
+    }
+    writeTracker.stall = PromiseWithResolvers();
+    return writeTracker.stall.promise;
   }
 
   function shutdownWithAnAction(action, rejected, originalError) {
@@ -1561,7 +1590,7 @@ function readableStreamPipeTo(
     if (dest[kState].state === 'writable' &&
         !writableStreamCloseQueuedOrInFlight(dest)) {
       PromisePrototypeThen(
-        waitForCurrentWrite(),
+        waitForPendingWrites(),
         complete,
         (error) => finalize(true, error));
       return;
@@ -1582,7 +1611,7 @@ function readableStreamPipeTo(
     if (dest[kState].state === 'writable' &&
         !writableStreamCloseQueuedOrInFlight(dest)) {
       PromisePrototypeThen(
-        waitForCurrentWrite(),
+        waitForPendingWrites(),
         () => finalize(rejected, error),
         (error) => finalize(true, error));
       return;
@@ -1639,25 +1668,46 @@ function readableStreamPipeTo(
       PromisePrototypeThen(promise, action, () => {});
   }
 
-  async function step() {
-    if (shuttingDown) return true;
+  // The pump loop is callback-driven to avoid per-iteration promise
+  // allocations. At most one read is in flight at a time, so one read
+  // request and one forwarding function are reused for every chunk;
+  // the chunk travels through `pendingChunk`.
+  let pendingChunk;
+  let readRequest;
+
+  // Ready promise rejection is handled by the destination-errored
+  // watcher.
+  function ignoreReadyRejection() {}
+
+  function forwardChunk() {
+    const chunk = pendingChunk;
+    pendingChunk = undefined;
+    writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
+    pump();
+  }
+
+  function pump() {
+    if (shuttingDown) return;
 
     if (dest[kState].backpressure) {
-      await writerReadyPromise(writer).promise;
-      if (shuttingDown) return true;
+      PromisePrototypeThen(
+        writerReadyPromise(writer).promise,
+        pump,
+        ignoreReadyRejection);
+      return;
     }
 
     const controller = source[kState].controller;
 
     // Fast path: batch reads when data is buffered in a default controller.
-    // This avoids creating PipeToReadableStreamReadRequest objects and
-    // reduces promise allocation overhead.
+    // This avoids parking read requests and reduces promise allocation
+    // overhead.
     if (source[kState].state === 'readable' &&
         isReadableStreamDefaultController(controller) &&
         controller[kState].queue.length > 0) {
 
       while (controller[kState].queue.length > 0) {
-        if (shuttingDown) return true;
+        if (shuttingDown) return;
 
         const chunk = dequeueValue(controller);
 
@@ -1668,8 +1718,7 @@ function readableStreamPipeTo(
 
         // Write the chunk - we're already in a separate microtask from enqueue
         // because we awaited the writer ready promise above.
-        state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
-        markPromiseAsHandled(state.currentWrite);
+        writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
 
         // Check backpressure after each write
         if (dest[kState].backpressure) {
@@ -1686,24 +1735,29 @@ function readableStreamPipeTo(
 
       // Check if stream closed during batch
       if (source[kState].state === 'closed') {
-        return true;
+        return;
       }
 
-      // Yield to microtask queue between batches to allow events/signals to fire
-      return false;
+      // Yield to microtask queue between batches to allow events/signals
+      // to fire
+      queueMicrotask(pump);
+      return;
     }
 
-    // Slow path: use read request for async reads
-    const promise = PromiseWithResolvers();
-    // eslint-disable-next-line no-use-before-define
-    readableStreamDefaultReaderRead(reader, new PipeToReadableStreamReadRequest(writer, state, promise));
-
-    return promise.promise;
-  }
-
-  async function run() {
-    // Run until step resolves as true
-    while (!await step());
+    // Slow path: park a lazily materialized read request. Close and
+    // error are handled by the source watchers.
+    readRequest ??= {
+      [kChunk](chunk) {
+        // Per spec, pipeTo must queue a microtask for the write to avoid
+        // synchronous write during enqueue(). See WHATWG Streams spec
+        // "ReadableStreamPipeTo" step 15's "chunk steps".
+        pendingChunk = chunk;
+        queueMicrotask(forwardChunk);
+      },
+      [kClose]() {},
+      [kError]() {},
+    };
+    readableStreamDefaultReaderRead(reader, readRequest);
   }
 
   if (signal !== undefined) {
@@ -1715,7 +1769,7 @@ function readableStreamPipeTo(
     disposable = addAbortListener(signal, abortAlgorithm);
   }
 
-  setPromiseHandled(run());
+  pump();
 
   watchErrored(source, readerClosedPromise(reader).promise, (error) => {
     if (!preventAbort) {
@@ -1760,33 +1814,6 @@ function readableStreamPipeTo(
   return promise.promise;
 }
 
-class PipeToReadableStreamReadRequest {
-  constructor(writer, state, promise) {
-    this.writer = writer;
-    this.state = state;
-    this.promise = promise;
-  }
-
-  [kChunk](chunk) {
-    // Per spec, pipeTo must queue a microtask for the write to avoid
-    // synchronous write during enqueue(). See WHATWG Streams spec
-    // "ReadableStreamPipeTo" step 15's "chunk steps".
-    queueMicrotask(() => {
-      this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk);
-      markPromiseAsHandled(this.state.currentWrite);
-      this.promise.resolve(false);
-    });
-  }
-
-  [kClose]() {
-    this.promise.resolve(true);
-  }
-
-  [kError](error) {
-    this.promise.reject(error);
-  }
-}
-
 function readableStreamTee(stream, cloneForBranch2) {
   if (isReadableByteStreamController(stream[kState].controller)) {
     return readableByteStreamTee(stream);
diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js
index f0eb687be129..87e3bcaa2850 100644
--- a/lib/internal/webstreams/writablestream.js
+++ b/lib/internal/webstreams/writablestream.js
@@ -1001,6 +1001,61 @@ function writableStreamDefaultWriterWrite(writer, chunk) {
   return promise;
 }
 
+// Variant of writableStreamDefaultWriterWrite for pipeTo: the caller
+// provides a shared request object instead of a per-write promise record.
+// `pending` is incremented before the controller write, which can settle
+// requests synchronously when the stream starts erroring; precondition
+// failures are latched on `failed`/`failure`.
+function writableStreamDefaultWriterWriteWithRequest(writer, chunk, request) {
+  const writerState = writer[kState];
+  const stream = writerState.stream;
+  assert(stream !== undefined);
+  const streamState = stream[kState];
+  const {
+    controller,
+  } = streamState;
+  const chunkSize = writableStreamDefaultControllerGetChunkSize(
+    controller,
+    chunk);
+  if (stream !== writerState.stream) {
+    request.failed = true;
+    request.failure =
+      new ERR_INVALID_STATE.TypeError('Mismatched WritableStreams');
+    return;
+  }
+  const {
+    state,
+  } = streamState;
+
+  if (state === 'errored') {
+    request.failed = true;
+    request.failure = streamState.storedError;
+    return;
+  }
+
+  if (streamState.closeQueuedOrInFlight || state === 'closed') {
+    request.failed = true;
+    request.failure =
+      new ERR_INVALID_STATE.TypeError('WritableStream is closed');
+    return;
+  }
+
+  if (state === 'erroring') {
+    request.failed = true;
+    request.failure = streamState.storedError;
+    return;
+  }
+
+  assert(state === 'writable');
+
+  let writeRequests = streamState.writeRequests;
+  if (writeRequests === kEmptyQueue)
+    writeRequests = streamState.writeRequests = new Queue();
+  writeRequests.push(request);
+  request.pending++;
+  writableStreamDefaultControllerWrite(controller, chunk, chunkSize);
+}
+
 function writableStreamDefaultWriterRelease(writer) {
   const {
     stream,
@@ -1376,6 +1431,7 @@ module.exports = {
   writableStreamCloseQueuedOrInFlight,
   writableStreamAddWriteRequest,
   writableStreamDefaultWriterWrite,
+  writableStreamDefaultWriterWriteWithRequest,
   writableStreamDefaultWriterRelease,
   writableStreamDefaultWriterGetDesiredSize,
   writableStreamDefaultWriterEnsureReadyPromiseRejected,
diff --git a/test/parallel/test-webstreams-pipeto-write-request.js b/test/parallel/test-webstreams-pipeto-write-request.js
new file mode 100644
index 000000000000..23e973a400da
--- /dev/null
+++ b/test/parallel/test-webstreams-pipeto-write-request.js
@@ -0,0 +1,162 @@
+// Flags: --expose-internals --no-warnings
+'use strict';
+
+// Exercises writableStreamDefaultWriterWriteWithRequest: settle paths
+// and the latched precondition failures.
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  WritableStream,
+  WritableStreamDefaultWriter,
+  ReadableStream,
+} = require('stream/web');
+
+const {
+  writableStreamDefaultWriterWriteWithRequest,
+} = require('internal/webstreams/writablestream');
+
+function makeRequest(overrides = {}) {
+  return {
+    promise: null,
+    pending: 0,
+    failed: false,
+    failure: undefined,
+    resolve: common.mustNotCall('resolve'),
+    reject: common.mustNotCall('reject'),
+    ...overrides,
+  };
+}
+
+{
+  // A write to a writable destination bumps `pending` and settles through
+  // request.resolve().
+  const request = makeRequest({
+    resolve: common.mustCall(function() {
+      assert.strictEqual(this, request);
+      assert.strictEqual(this.pending, 1);
+      assert.strictEqual(this.failed, false);
+    }),
+  });
+  const ws = new WritableStream({
+    write: common.mustCall((chunk) => {
+      assert.strictEqual(chunk, 'chunk');
+    }),
+  });
+  const writer = new WritableStreamDefaultWriter(ws);
+  writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request);
+  assert.strictEqual(request.pending, 1);
+}
+
+{
+  // A write whose sink rejects settles through request.reject() with the
+  // sink's error.
+  const error = new Error('sink failure');
+  const request = makeRequest({
+    reject: common.mustCall(function(reason) {
+      assert.strictEqual(this, request);
+      assert.strictEqual(reason, error);
+    }),
+  });
+  const ws = new WritableStream({
+    write: common.mustCall(() => Promise.reject(error)),
+  });
+  const writer = new WritableStreamDefaultWriter(ws);
+  writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request);
+  assert.strictEqual(request.pending, 1);
+}
+
+{
+  // Writing to an errored destination latches the stored error without
+  // queueing the write.
+  const error = new Error('start failure');
+  const ws = new WritableStream({
+    start(controller) { controller.error(error); },
+  });
+  queueMicrotask(common.mustCall(() => {
+    const writer = new WritableStreamDefaultWriter(ws);
+    const request = makeRequest();
+    writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request);
+    assert.strictEqual(request.pending, 0);
+    assert.strictEqual(request.failed, true);
+    assert.strictEqual(request.failure, error);
+  }));
+}
+
+{
+  // Writing to a destination with a queued close latches an
+  // ERR_INVALID_STATE TypeError.
+  const ws = new WritableStream({});
+  const writer = new WritableStreamDefaultWriter(ws);
+  writer.close().then(common.mustCall());
+  const request = makeRequest();
+  writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request);
+  assert.strictEqual(request.pending, 0);
+  assert.strictEqual(request.failed, true);
+  assert.match(request.failure.message, /WritableStream is closed/);
+}
+
+{
+  // Writing to an erroring destination latches the abort reason. The
+  // stream stays in the 'erroring' state until its start algorithm
+  // settles, so aborting right after construction reaches it
+  // deterministically.
+  const reason = new Error('abort reason');
+  const ws = new WritableStream({});
+  const writer = new WritableStreamDefaultWriter(ws);
+  writer.abort(reason).then(common.mustCall());
+  const request = makeRequest();
+  writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request);
+  assert.strictEqual(request.pending, 0);
+  assert.strictEqual(request.failed, true);
+  assert.strictEqual(request.failure, reason);
+}
+
+{
+  // A size algorithm that detaches the writer makes the write latch a
+  // mismatched-streams error.
+  let writer;
+  const ws = new WritableStream({}, {
+    size: common.mustCall(() => {
+      writer.releaseLock();
+      return 1;
+    }),
+    highWaterMark: 1,
+  });
+  writer = new WritableStreamDefaultWriter(ws);
+  const request = makeRequest();
+  writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request);
+  assert.strictEqual(request.pending, 0);
+  assert.strictEqual(request.failed, true);
+  assert.match(request.failure.message, /Mismatched WritableStreams/);
+}
+
+{
+  // End to end: aborting a pipe with a write in flight still waits for
+  // the in-flight write before aborting the destination, and the pipe
+  // rejects with an AbortError.
+  const ac = new AbortController();
+  const order = [];
+  const { promise: gate, resolve: openGate } = Promise.withResolvers();
+  let i = 0;
+  const rs = new ReadableStream({
+    pull(controller) { controller.enqueue(i++); },
+  });
+  const pipe = rs.pipeTo(new WritableStream({
+    write: common.mustCall((chunk) => {
+      order.push(`write:${chunk}`);
+      ac.abort();
+      queueMicrotask(() => {
+        order.push('settle:0');
+        openGate();
+      });
+      return gate;
+    }),
+    abort: common.mustCall(() => {
+      order.push('abort');
+      assert.deepStrictEqual(order, ['write:0', 'settle:0', 'abort']);
+    }),
+  }), { signal: ac.signal });
+  assert.rejects(pipe, { name: 'AbortError' }).then(common.mustCall());
+}