Don't cancel ReadableStreams that are locked.

kentonv · kentonv · commit 111a252d0f69 · 2026-02-12T19:34:28.000-06:00
I noticed that the `state.stream.cancel()` call in `ReadableStreamStubHook.dispose()` was often failing, throwing an exception because the stream was already locked. But we were ignoring the exceptions.

This actually fixes the problem in two ways:
1. When we pipe from a ReadableStream (which locks it), we also take a reference on its StubHook, which we dispose then the pipe completes. This makes sense and solves the case seen in the tests.
2. I also just made it skip the cancel() call if the stream is locked. Throwing the exception and ignoring it is just a waste of cycles.
diff --git a/src/rpc.ts b/src/rpc.ts
@@ -538,7 +538,7 @@ class RpcSessionImpl implements Importer, Exporter {
     return readable;
   }
 
-  createPipe(readable: ReadableStream): ImportId {
+  createPipe(readable: ReadableStream, readableHook: StubHook): ImportId {
     if (this.abortReason) throw this.abortReason;
 
     this.send(["pipe"]);
@@ -555,7 +555,7 @@ class RpcSessionImpl implements Importer, Exporter {
       // Errors are handled by the writable stream's error handling -- either the write fails
       // and the writable side reports it, or the readable side errors and pipeTo aborts the
       // writable side. Either way, the hook's disposal will handle cleanup.
-    });
+    }).finally(() => readableHook.dispose());
 
     return importId;
   }
diff --git a/src/serialize.ts b/src/serialize.ts
@@ -19,8 +19,9 @@ export interface Exporter {
   unexport(ids: Array<ExportId>): void;
 
   // Creates a pipe by sending a ["pipe"] message, then starts pumping the given ReadableStream
-  // into the pipe's writable end. Returns the import ID assigned to the pipe.
-  createPipe(readable: ReadableStream): ImportId;
+  // into the pipe's writable end. Returns the import ID assigned to the pipe. `hook` should be
+  // disposed when the pipe finishes.
+  createPipe(readable: ReadableStream, hook: StubHook): ImportId;
 
   onSendError(error: Error): Error | void;
 }
@@ -256,8 +257,11 @@ export class Devaluator {
           throw new Error("Can't serialize ReadableStream in this context.");
         }
 
+        let ws = <ReadableStream>value;
+        let hook = this.source.getHookForReadableStream(ws, parent);
+
         // Create a pipe and start pumping the ReadableStream into it.
-        let importId = this.exporter.createPipe(<ReadableStream>value);
+        let importId = this.exporter.createPipe(ws, hook);
 
         return ["readable", importId];
       }
diff --git a/src/streams.ts b/src/streams.ts
@@ -495,9 +495,20 @@ class ReadableStreamStubHook extends StubHook {
       if (--state.refcount === 0) {
         if (!state.canceled) {
           state.canceled = true;
-          state.stream.cancel(
-              new Error("ReadableStream RPC stub was disposed without being consumed"))
-              .catch(() => {});  // Ignore errors from cancel.
+
+          // Don't try to cancel the stream if it's locked. It won't work anyway -- it'll throw
+          // an exception, which we'd ignore anyway.
+          //
+          // This is a little janky but it makes some sense: If someone has locked the stream, they
+          // have taken responsibility for fully reading it. The only reason we really need to
+          // cancel when this hook is disposed is to handle the case where an application receives
+          // a ReadableStream but completely ignores it -- we want it to be canceled naturally when
+          // the payload is disposed.
+          if (!state.stream.locked) {
+            state.stream.cancel(
+                new Error("ReadableStream RPC stub was disposed without being consumed"))
+                .catch(() => {});  // Ignore errors from cancel.
+          }
         }
       }
     }

Original file line number	Diff line number	Diff line change
`@@ -538,7 +538,7 @@ class RpcSessionImpl implements Importer, Exporter {`
`538`	`538`	`return readable;`
`539`	`539`	`}`
`540`	`540`
`541`		`- createPipe(readable: ReadableStream): ImportId {`
	`541`	`+ createPipe(readable: ReadableStream, readableHook: StubHook): ImportId {`
`542`	`542`	`if (this.abortReason) throw this.abortReason;`
`543`	`543`
`544`	`544`	`this.send(["pipe"]);`
`@@ -555,7 +555,7 @@ class RpcSessionImpl implements Importer, Exporter {`
`555`	`555`	`// Errors are handled by the writable stream's error handling -- either the write fails`
`556`	`556`	`// and the writable side reports it, or the readable side errors and pipeTo aborts the`
`557`	`557`	`// writable side. Either way, the hook's disposal will handle cleanup.`
`558`		`- });`
	`558`	`+ }).finally(() => readableHook.dispose());`
`559`	`559`
`560`	`560`	`return importId;`
`561`	`561`	`}`