Release a blocked stream chunk when its RpcClient consumer is interrupted

.changeset/rpc-abandoned-stream-buffer.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Fix `RpcClient` stream interruption leaving chunk delivery blocked on a full buffer and stalling the shared protocol receive loop.

packages/effect/src/rpc/RpcClient.ts

@@ -454,9 +454,11 @@ export const makeNoSerialization: <Rpcs extends Rpc.Any, E, const Flatten extend
     const id = generateRequestId()
 
     const scope = Context.getUnsafe(fiber.context, Scope.Scope)
+    const queue = yield* Queue.bounded<any, any>(streamBufferSize)
     yield* Scope.addFinalizerExit(
       scope,
       (exit) => {
+        Queue.shutdownUnsafe(queue)
         if (!entries.has(id)) return Effect.void
         entries.delete(id)
         return sendInterrupt(
@@ -469,7 +471,6 @@ export const makeNoSerialization: <Rpcs extends Rpc.Any, E, const Flatten extend
       }
     )
 
-    const queue = yield* Queue.bounded<any, any>(streamBufferSize)
     entries.set(id, {
       _tag: "Queue",
       rpc,

packages/effect/test/rpc/RpcClient.test.ts

@@ -146,6 +146,24 @@ describe("RpcClient", () => {
       assert.isTrue(observedInterrupt)
     }))
 
+  it.effect("releases a chunk blocked on a full buffer when its stream consumer is interrupted", () =>
+    Effect.gen(function*() {
+      const consuming = yield* Deferred.make<void>()
+      const { client, handle, sent } = yield* makeChunkProtocol()
+      const reader = yield* client.Good(undefined, { streamBufferSize: 1 }).pipe(
+        Stream.runForEach(() => Deferred.succeed(consuming, void 0).pipe(Effect.andThen(Effect.never))),
+        Effect.forkChild
+      )
+      const requestId = takeRequestId(yield* Queue.take(sent))
+      const receiver = yield* handle({ _tag: "Chunk", requestId, values: ["a", "b", "c"] }).pipe(Effect.forkChild)
+      yield* Deferred.await(consuming)
+      assert.isUndefined(receiver.pollUnsafe())
+      yield* Fiber.interrupt(reader)
+
+      assert(Exit.isSuccess(yield* Fiber.await(receiver)))
+      assert.deepStrictEqual(yield* Queue.take(sent), { _tag: "Interrupt", requestId })
+    }))
+
   for (const consumer of ["queue", "stream"] as const) {
     it.effect(`releases the ${consumer} consumer when the request write is interrupted`, () =>
       Effect.gen(function*() {