Scope cluster reply serialization failures and peer-delivered defects to their own request

.changeset/fix-cluster-reply-defect-isolation.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Scope cluster reply serialization failures and peer-delivered defects to their own request instead of the whole runner connection

packages/effect/src/unstable/cluster/HttpRunner.ts

@@ -150,7 +150,8 @@ export const toHttpEffect: Effect.Effect<
   const handlers = yield* Layer.build(RunnerServer.layerHandlers)
   return yield* RpcServer.toHttpEffect(Runners.Rpcs, {
     spanPrefix: "RunnerServer",
-    disableTracing: true
+    disableTracing: true,
+    disableFatalDefects: true
   }).pipe(Effect.provideContext(handlers))
 })
 
@@ -173,7 +174,8 @@ export const toHttpEffectWebsocket: Effect.Effect<
   const handlers = yield* Layer.build(RunnerServer.layerHandlers)
   return yield* RpcServer.toHttpEffectWebsocket(Runners.Rpcs, {
     spanPrefix: "RunnerServer",
-    disableTracing: true
+    disableTracing: true,
+    disableFatalDefects: true
   }).pipe(Effect.provideContext(handlers))
 })
 

packages/effect/src/unstable/cluster/RunnerServer.ts

@@ -10,13 +10,15 @@
  *
  * @since 4.0.0
  */
+import type * as Cause from "../../Cause.ts"
 import * as Effect from "../../Effect.ts"
 import type * as Exit from "../../Exit.ts"
 import * as Fiber from "../../Fiber.ts"
 import { constant } from "../../Function.ts"
 import * as Layer from "../../Layer.ts"
 import * as Option from "../../Option.ts"
 import * as Queue from "../../Queue.ts"
+import type * as Rpc from "../rpc/Rpc.ts"
 import * as RpcServer from "../rpc/RpcServer.ts"
 import type * as ClusterError from "./ClusterError.ts"
 import * as Message from "./Message.ts"
@@ -30,6 +32,30 @@ import { ShardingConfig } from "./ShardingConfig.ts"
 
 const constVoid = constant(Effect.void)
 
+// The original reply never left this runner, so its id can be reused for the
+// fallback reply without requiring a Snowflake.Generator here.
+const serializeDefectReply = <R extends Rpc.Any>(
+  reply: Reply.ReplyWithContext<R>,
+  defect: unknown
+): Effect.Effect<Reply.Encoded> =>
+  Effect.orDie(Reply.serialize(Reply.ReplyWithContext.fromDefect({
+    id: reply.reply.id,
+    requestId: reply.reply.requestId,
+    defect
+  })))
+
+// A reply that cannot be serialized is delivered to its own request as a
+// defect, instead of failing the handler fiber and poisoning every request
+// multiplexed on the runner connection.
+const serializeReply = <R extends Rpc.Any>(
+  reply: Reply.ReplyWithContext<R>
+): Effect.Effect<Reply.Encoded> =>
+  Effect.catchTag(
+    Reply.serialize(reply),
+    "MalformedMessage",
+    (error) => serializeDefectReply(reply, error)
+  )
+
 /**
  * Layer that handles runner protocol RPCs by forwarding requests to `Sharding`
  * and `MessageStorage`.
@@ -63,7 +89,7 @@ export const layerHandlers = Runners.Rpcs.toLayer(Effect.gen(function*() {
         envelope: request,
         lastSentReply: Option.none(),
         respond(reply) {
-          resume(Effect.orDie(Reply.serialize(reply)))
+          resume(serializeReply(reply))
           return Effect.void
         }
       })
@@ -108,23 +134,34 @@ export const layerHandlers = Runners.Rpcs.toLayer(Effect.gen(function*() {
     },
     Stream: ({ persisted, request }) =>
       Effect.flatMap(
-        Queue.make<Reply.Encoded, ClusterError.EntityNotAssignedToRunner>(),
+        Queue.make<Reply.Encoded, ClusterError.EntityNotAssignedToRunner | Cause.Done>(),
         (queue) => {
           const message = new Message.IncomingRequest({
             envelope: request,
             lastSentReply: Option.none(),
             respond(reply) {
-              return Effect.flatMap(Reply.serialize(reply), (reply) => {
-                Queue.offerUnsafe(queue, reply)
-                return Effect.void
-              })
+              return Reply.serialize(reply).pipe(
+                Effect.flatMap((reply) => {
+                  Queue.offerUnsafe(queue, reply)
+                  return Effect.void
+                }),
+                Effect.catchTag("MalformedMessage", (error) =>
+                  Effect.flatMap(serializeDefectReply(reply, error), (reply) => {
+                    // the fallback defect reply is terminal, so end the stream
+                    Queue.offerUnsafe(queue, reply)
+                    Queue.endUnsafe(queue)
+                    return Effect.void
+                  }))
+              )
             }
           })
           return Effect.as(
             persisted ?
               Effect.andThen(
                 storage.registerReplyHandler(message).pipe(
-                  Effect.onError((cause) => Queue.failCause(queue, cause)),
+                  Effect.onError((cause) =>
+                    Queue.failCause(queue, cause)
+                  ),
                   Effect.forkScoped
                 ),
                 sharding.notify(message, constWaitUntilRead)
@@ -168,7 +205,10 @@ export const layer: Layer.Layer<
   RpcServer.Protocol | Sharding.Sharding | MessageStorage.MessageStorage
 > = RpcServer.layer(Runners.Rpcs, {
   spanPrefix: "RunnerServer",
-  disableTracing: true
+  disableTracing: true,
+  // a handler defect concerns a single request, so it must not tear down
+  // every request multiplexed on the runner connection
+  disableFatalDefects: true
 }).pipe(Layer.provide(layerHandlers))
 
 /**

packages/effect/src/unstable/cluster/Runners.ts

@@ -574,11 +574,25 @@ export const makeRpc: Effect.Effect<
               persisted: isPersisted
             })
           ),
-          Effect.catchTag("RpcClientError", Effect.die),
           Effect.scoped,
-          Effect.catchDefect(() => Effect.fail(new RunnerUnavailable({ address })))
+          Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address })))
         )
       }
+      // A defect here was delivered by the peer, not a transport failure - those
+      // surface as `RpcClientError`. Persisted requests recover their reply from
+      // storage via the `RunnerUnavailable` path. Volatile requests have no stored
+      // reply and duplicate delivery is rejected by the entity's dedup guard, so
+      // the defect becomes the request's reply.
+      const respondDefect = (defect: unknown) =>
+        isPersisted
+          ? Effect.fail(new RunnerUnavailable({ address }))
+          : message.respond(
+            new Reply.WithExit({
+              id: snowflakeGen.nextUnsafe(),
+              requestId: message.envelope.requestId,
+              exit: Exit.die(defect)
+            })
+          )
       const isStream = RpcSchema.isStreamSchema(rpc.successSchema)
       if (!isStream) {
         return Effect.matchEffect(Message.serializeRequest(message), {
@@ -590,7 +604,6 @@ export const makeRpc: Effect.Effect<
                   persisted: isPersisted
                 })
               ),
-              Effect.catchTag("RpcClientError", Effect.die),
               Effect.flatMap((reply) =>
                 Schema.decodeEffect(Reply.Reply(message.rpc))(reply).pipe(
                   Effect.provideContext(message.context),
@@ -599,7 +612,8 @@ export const makeRpc: Effect.Effect<
               ),
               Effect.flatMap(message.respond),
               Effect.scoped,
-              Effect.catchDefect(() => Effect.fail(new RunnerUnavailable({ address })))
+              Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address }))),
+              Effect.catchDefect(respondDefect)
             ),
           onFailure: (error) =>
             message.respond(
@@ -626,10 +640,10 @@ export const makeRpc: Effect.Effect<
                 Effect.flatMap((reply) => Effect.orDie(decode(reply))),
                 Effect.flatMap(message.respond),
                 Effect.forever,
-                Effect.catchTag("RpcClientError", Effect.die),
                 Effect.provideContext(message.context),
                 Effect.catchTag("Done", (_) => Effect.void),
-                Effect.catchDefect(() => Effect.fail(new RunnerUnavailable({ address })))
+                Effect.catchTag("RpcClientError", () => Effect.fail(new RunnerUnavailable({ address }))),
+                Effect.catchDefect(respondDefect)
               )
             }),
             Effect.scoped