diff --git a/.changeset/fix-cluster-reply-defect-isolation.md b/.changeset/fix-cluster-reply-defect-isolation.md new file mode 100644 index 00000000000..62f0d84727f --- /dev/null +++ b/.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 diff --git a/packages/effect/src/unstable/cluster/HttpRunner.ts b/packages/effect/src/unstable/cluster/HttpRunner.ts index 05663cdfd0c..cb05b035814 100644 --- a/packages/effect/src/unstable/cluster/HttpRunner.ts +++ b/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)) }) diff --git a/packages/effect/src/unstable/cluster/RunnerServer.ts b/packages/effect/src/unstable/cluster/RunnerServer.ts index cea221b3837..7f4f6ef97c2 100644 --- a/packages/effect/src/unstable/cluster/RunnerServer.ts +++ b/packages/effect/src/unstable/cluster/RunnerServer.ts @@ -10,6 +10,7 @@ * * @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" @@ -17,6 +18,7 @@ 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 = ( + reply: Reply.ReplyWithContext, + defect: unknown +): Effect.Effect => + 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 = ( + reply: Reply.ReplyWithContext +): Effect.Effect => + 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(), + Queue.make(), (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)) /** diff --git a/packages/effect/src/unstable/cluster/Runners.ts b/packages/effect/src/unstable/cluster/Runners.ts index fea620c40df..a12d95fd0ed 100644 --- a/packages/effect/src/unstable/cluster/Runners.ts +++ b/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 diff --git a/packages/effect/test/cluster/Runners.test.ts b/packages/effect/test/cluster/Runners.test.ts new file mode 100644 index 00000000000..753f6938f44 --- /dev/null +++ b/packages/effect/test/cluster/Runners.test.ts @@ -0,0 +1,286 @@ +import { assert, describe, it } from "@effect/vitest" +import { Cause, Context, Effect, Exit, Layer, Option, Queue, Schema, Stream } from "effect" +import { TestClock } from "effect/testing" +import { + ClusterError, + ClusterSchema, + Entity, + EntityAddress, + EntityId, + EntityType, + Envelope, + Message, + MessageStorage, + type Reply, + RunnerAddress, + RunnerHealth, + Runners, + RunnerServer, + RunnerStorage, + ShardId, + Sharding, + ShardingConfig, + Snowflake +} from "effect/unstable/cluster" +import { Headers } from "effect/unstable/http" +import { Rpc, RpcClient, RpcTest } from "effect/unstable/rpc" +import { RpcClientError } from "effect/unstable/rpc/RpcClientError" +import type { FromClientEncoded, FromServerEncoded } from "effect/unstable/rpc/RpcMessage" +import { Socket } from "effect/unstable/socket" + +// An entity whose replies cannot be serialized: the handlers return +// non-integers for a `Schema.Int` success schema, so `Reply.serialize` fails +// on encode. +const BadReplyEntity = Entity.make("BadReplyEntity", [ + Rpc.make("BadReply", { + success: Schema.Int, + payload: { id: Schema.Number } + }), + Rpc.make("BadStream", { + success: Schema.Int, + payload: { id: Schema.Number }, + stream: true + }) +]).annotateRpcs(ClusterSchema.Persisted, false) + +const BadReplyEntityLayer = BadReplyEntity.toLayer({ + BadReply: () => Effect.succeed(1.5), + BadStream: () => Stream.make(2.5) +}) + +const TestShardingConfig = ShardingConfig.layer({ + entityMailboxCapacity: 10, + entityTerminationTimeout: 0, + entityMessagePollInterval: 5000, + sendRetryInterval: 100, + refreshAssignmentsInterval: 0 +}) + +const RunnerServerHandlers = RunnerServer.layerHandlers.pipe( + Layer.provideMerge(BadReplyEntityLayer), + Layer.provideMerge(Sharding.layer), + Layer.provideMerge(Snowflake.layerGenerator), + Layer.provide(RunnerStorage.layerMemory), + Layer.provide(RunnerHealth.layerNoop), + Layer.provide(Runners.layerNoop), + Layer.provideMerge(MessageStorage.layerMemory), + Layer.provide(TestShardingConfig) +) + +describe.concurrent("RunnerServer", () => { + const makeRequest = (options: { + readonly entityId: string + readonly tag: string + }) => + Effect.gen(function*() { + const sharding = yield* Sharding.Sharding + const snowflakeGen = yield* Snowflake.Generator + const entityId = EntityId.make(options.entityId) + const address = EntityAddress.make({ + shardId: sharding.getShardId(entityId, BadReplyEntity.getShardGroup(entityId)), + entityType: EntityType.make("BadReplyEntity"), + entityId + }) + return { + _tag: "Request", + requestId: snowflakeGen.nextUnsafe(), + address, + tag: options.tag, + payload: { id: 1 }, + headers: Headers.empty + } as Envelope.PartialRequest + }) + + it.effect("a reply that fails to serialize fails only its own request", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const client = yield* RpcTest.makeClient(Runners.Rpcs) + const request = yield* makeRequest({ entityId: "bad-1", tag: "BadReply" }) + + const exit = yield* Effect.exit(client.Effect({ request, persisted: false })) + // the serialize failure must be delivered as this request's reply, + // not defect the whole connection + if (!Exit.isSuccess(exit)) { + return assert.fail("Effect rpc must not defect on a reply serialization failure") + } + const reply = exit.value + if (reply._tag !== "WithExit" || reply.exit._tag !== "Failure") { + return assert.fail("expected a WithExit reply with a failure exit") + } + assert.strictEqual(reply.requestId, String(request.requestId)) + const die = reply.exit.cause.find((entry) => entry._tag === "Die") + assert.isDefined(die, "the reply exit must carry the encode failure as a defect") + assert.include( + JSON.stringify(die), + "MalformedMessage", + "the defect must identify the encode failure" + ) + }).pipe(Effect.provide(RunnerServerHandlers))) + + it.effect("a stream reply that fails to serialize ends the stream with the defect", () => + Effect.gen(function*() { + yield* TestClock.adjust(1) + const client = yield* RpcTest.makeClient(Runners.Rpcs) + const request = yield* makeRequest({ entityId: "bad-2", tag: "BadStream" }) + + const queue = yield* client.Stream({ request, persisted: false }, { asQueue: true }) + const replies: Array = [] + yield* Queue.take(queue).pipe( + Effect.flatMap((reply) => + Effect.sync(() => { + replies.push(reply) + }) + ), + Effect.forever, + Effect.catchTag("Done", () => Effect.void) + ) + + // the stream is terminated with a defect reply for the chunk that + // could not be serialized + assert.strictEqual(replies.length, 1) + const last = replies[0] + if (last._tag !== "WithExit" || last.exit._tag !== "Failure") { + return assert.fail("expected a terminal WithExit reply with a failure exit") + } + assert.strictEqual(last.requestId, String(request.requestId)) + assert.include( + JSON.stringify(last.exit.cause), + "MalformedMessage", + "the defect must identify the encode failure" + ) + }).pipe(Effect.provide(RunnerServerHandlers))) +}) + +describe.concurrent("Runners.makeRpc", () => { + const runnerAddress = RunnerAddress.make("localhost", 42_000) + + const TestRpc = Rpc.make("TestRpc", { + success: Schema.Number, + payload: { id: Schema.Number } + }).annotate(ClusterSchema.Persisted, false) + + const TestRpcPersisted = Rpc.make("TestRpcPersisted", { + success: Schema.Number, + payload: { id: Schema.Number } + }).annotate(ClusterSchema.Persisted, true) + + type SendRpc = typeof TestRpc | typeof TestRpcPersisted + + const makeOutgoingRequest = ( + rpc: SendRpc, + requestId: Snowflake.Snowflake, + respond: (reply: Reply.Reply) => Effect.Effect + ): Message.OutgoingRequest => + new Message.OutgoingRequest({ + envelope: Envelope.makeRequest({ + requestId, + address: EntityAddress.make({ + shardId: ShardId.make("default", 1), + entityType: EntityType.make("TestRpcEntity"), + entityId: EntityId.make("1") + }), + tag: rpc._tag, + payload: { id: 1 }, + headers: Headers.empty + }), + rpc, + context: Context.empty(), + lastReceivedReply: Option.none(), + respond, + annotations: Context.empty() + }) + + const layerFakeProtocol = ( + onRequest: ( + request: FromClientEncoded, + write: (data: FromServerEncoded) => Effect.Effect + ) => Effect.Effect + ) => + Layer.succeed(Runners.RpcClientProtocol)(() => + Effect.sync(() => { + let write!: (data: FromServerEncoded) => Effect.Effect + return RpcClient.Protocol.of({ + run(_clientId, f) { + write = f + return Effect.never + }, + send(_clientId, request) { + return onRequest(request, write) + }, + supportsAck: true, + supportsTransferables: false + }) + }) + ) + + const layerRunners = (protocol: Layer.Layer) => + Runners.layerRpc.pipe( + Layer.provideMerge(Snowflake.layerGenerator), + Layer.provide(protocol), + Layer.provideMerge(MessageStorage.layerNoop), + Layer.provide(TestShardingConfig) + ) + + const respondWithDefect = (request: FromClientEncoded, write: (data: FromServerEncoded) => Effect.Effect) => + request._tag === "Request" ? write({ _tag: "Defect", defect: "boom" }) : Effect.void + + it.effect("a server-delivered defect resolves the request instead of RunnerUnavailable", () => + Effect.gen(function*() { + const runners = yield* Runners.Runners + const snowflakeGen = yield* Snowflake.Generator + const replies: Array> = [] + const message = makeOutgoingRequest(TestRpc, snowflakeGen.nextUnsafe(), (reply) => + Effect.sync(() => { + replies.push(reply) + })) + + const exit = yield* Effect.exit(runners.send({ address: runnerAddress, message })) + // a defect the server delivered must not be treated as an unavailable + // runner, otherwise the message is re-sent into the dedup guard + assert.isTrue(Exit.isSuccess(exit), "send must not fail with RunnerUnavailable for a delivered defect") + assert.strictEqual(replies.length, 1) + const reply = replies[0] + if (reply._tag !== "WithExit" || !Exit.isFailure(reply.exit)) { + return assert.fail("expected a WithExit reply with a failure exit") + } + assert.include( + String(Cause.squash(reply.exit.cause)), + "boom", + "the reply must carry the server defect" + ) + }).pipe(Effect.provide(layerRunners(layerFakeProtocol(respondWithDefect))))) + + it.effect("transport failures still map to RunnerUnavailable", () => + Effect.gen(function*() { + const runners = yield* Runners.Runners + const snowflakeGen = yield* Snowflake.Generator + const message = makeOutgoingRequest(TestRpc, snowflakeGen.nextUnsafe(), () => Effect.void) + + const exit = yield* Effect.exit(runners.send({ address: runnerAddress, message })) + if (!Exit.isFailure(exit)) { + return assert.fail("send must fail for a transport failure") + } + assert.instanceOf(Cause.squash(exit.cause), ClusterError.RunnerUnavailable) + }).pipe(Effect.provide(layerRunners(layerFakeProtocol(() => + Effect.fail( + new RpcClientError({ + reason: new Socket.SocketCloseError({ code: 1006 }) + }) + ) + ))))) + + it.effect("a delivered defect for a persisted request maps to RunnerUnavailable for storage recovery", () => + Effect.gen(function*() { + const runners = yield* Runners.Runners + const snowflakeGen = yield* Snowflake.Generator + const message = makeOutgoingRequest(TestRpcPersisted, snowflakeGen.nextUnsafe(), () => Effect.void) + + const exit = yield* Effect.exit(runners.send({ address: runnerAddress, message })) + // persisted replies can be recovered from storage, so the send is + // reported as RunnerUnavailable to trigger the storage fallback + if (!Exit.isFailure(exit)) { + return assert.fail("send must fail for a delivered defect on a persisted request") + } + assert.instanceOf(Cause.squash(exit.cause), ClusterError.RunnerUnavailable) + }).pipe(Effect.provide(layerRunners(layerFakeProtocol(respondWithDefect))))) +}) diff --git a/packages/platform-node/test/RpcServer.test.ts b/packages/platform-node/test/RpcServer.test.ts index b4104c316cd..1f36177d076 100644 --- a/packages/platform-node/test/RpcServer.test.ts +++ b/packages/platform-node/test/RpcServer.test.ts @@ -1,6 +1,6 @@ import { NodeHttpServer, NodeSocket, NodeSocketServer } from "@effect/platform-node" import { assert, describe, it } from "@effect/vitest" -import { Cause, Deferred, Effect, Fiber, Layer, Ref, Schedule, Schema, Stream } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, Layer, Ref, Schedule, Schema, Stream } from "effect" import { Entity, EntityProxy, EntityProxyServer, Sharding } from "effect/unstable/cluster" import { HttpClient, HttpClientRequest, HttpRouter, HttpServer } from "effect/unstable/http" import { Rpc, RpcClient, RpcGroup, RpcSerialization, RpcServer, RpcTest } from "effect/unstable/rpc" @@ -211,11 +211,70 @@ describe("RpcServer", () => { })) }) - describe("unknown-tag isolation", () => { - const Ticker = Rpc.make("Ticker", { - success: Schema.Number, - stream: true + // Shared harness asserting that a failing sibling call does not affect an + // in-flight Ticker stream multiplexed on the same socket connection. + const Ticker = Rpc.make("Ticker", { + success: Schema.Number, + stream: true + }) + + const IsolationClient = RpcClient.layerProtocolSocket().pipe( + Layer.provide( + Effect.gen(function*() { + const server = yield* SocketServer.SocketServer + const address = server.address as SocketServer.TcpAddress + return NodeSocket.layerNet({ port: address.port }) + }).pipe(Layer.unwrap) + ), + Layer.provide(RpcSerialization.layerNdjson) + ) + + const assertTickerSurvives = ( + setup: Effect.Effect< + { + readonly ticker: Stream.Stream + readonly failingCall: Effect.Effect + }, + never, + R + >, + label: string + ) => + Effect.gen(function*() { + const { failingCall, ticker } = yield* setup + + const received = yield* Ref.make>([]) + + const tickerFiber = yield* ticker.pipe( + Stream.runForEach((value) => Ref.update(received, (xs) => [...xs, value])), + Effect.forkChild + ) + + yield* Effect.retry( + Effect.flatMap( + Ref.get(received), + (xs) => xs.length >= 2 ? Effect.void : Effect.fail("not enough ticks yet") + ), + { schedule: Schedule.spaced("50 millis"), times: 200 } + ) + + const ticksBefore = (yield* Ref.get(received)).length + assert.isAtLeast(ticksBefore, 2) + + yield* failingCall + + yield* Effect.sleep("300 millis") + + const ticksAfter = (yield* Ref.get(received)).length + const tickerStatus = tickerFiber.pollUnsafe() + + yield* Fiber.interrupt(tickerFiber) + + assert.isUndefined(tickerStatus, `Ticker stream must still be running after ${label}`) + assert.isAbove(ticksAfter, ticksBefore, `Ticker stream must keep emitting after ${label}`) }) + + describe("unknown-tag isolation", () => { const Ghost = Rpc.make("Ghost", { payload: { value: Schema.String }, success: Schema.String @@ -234,58 +293,66 @@ describe("RpcServer", () => { Layer.provideMerge(NodeSocketServer.layer({ port: 0 })), Layer.provide(RpcSerialization.layerNdjson) ) - const IsolationClient = RpcClient.layerProtocolSocket().pipe( - Layer.provide( - Effect.gen(function*() { - const server = yield* SocketServer.SocketServer - const address = server.address as SocketServer.TcpAddress - return NodeSocket.layerNet({ port: address.port }) - }).pipe(Layer.unwrap) - ), - Layer.provide(RpcSerialization.layerNdjson) - ) it.live( "an unknown request tag fails only its own request, not other in-flight streams on the same connection", () => - Effect.gen(function*() { - const client = yield* RpcClient.make(clientGroup) - - const received = yield* Ref.make>([]) - - const tickerFiber = yield* client.Ticker().pipe( - Stream.runForEach((value) => Ref.update(received, (xs) => [...xs, value])), - Effect.forkChild - ) - - yield* Effect.retry( - Effect.flatMap( - Ref.get(received), - (xs) => xs.length >= 2 ? Effect.void : Effect.fail("not enough ticks yet") - ), - { schedule: Schedule.spaced("50 millis"), times: 200 } - ) - - const ticksBeforeGhost = (yield* Ref.get(received)).length - assert.isAtLeast(ticksBeforeGhost, 2) + assertTickerSurvives( + Effect.map(RpcClient.make(clientGroup), (client) => ({ + ticker: client.Ticker(), + failingCall: Effect.gen(function*() { + const ghostExit = yield* client.Ghost({ value: "boo" }).pipe(Effect.exit) + assert.isTrue(Exit.isFailure(ghostExit), "Ghost call should fail with the routing miss") + }) + })), + "the unknown-tag failure" + ).pipe(Effect.provide(IsolationClient.pipe(Layer.provideMerge(IsolationServer)))), + { timeout: 30_000 } + ) + }) - const ghostExit = yield* client.Ghost({ value: "boo" }).pipe(Effect.exit) - assert.isTrue(ghostExit._tag === "Failure", "Ghost call should fail with the routing miss") + // Guards the behavior the cluster RunnerServer relies on: with + // `disableFatalDefects` a handler defect is delivered as a per-request + // exit instead of a connection-wide defect frame. + describe("fatal-defect isolation", () => { + const Boom = Rpc.make("Boom", { + success: Schema.String + }) - yield* Effect.sleep("300 millis") + const group = RpcGroup.make(Ticker, Boom) - const ticksAfterGhost = (yield* Ref.get(received)).length - const tickerStatus = tickerFiber.pollUnsafe() + const Handlers = group.toLayer({ + Ticker: () => Stream.fromSchedule(Schedule.spaced("60 millis")), + Boom: () => Effect.die("boom") + }) - yield* Fiber.interrupt(tickerFiber) + const DefectServer = RpcServer.layer(group, { disableFatalDefects: true }).pipe( + Layer.provide(Handlers), + Layer.provideMerge(RpcServer.layerProtocolSocketServer), + Layer.provideMerge(NodeSocketServer.layer({ port: 0 })), + Layer.provide(RpcSerialization.layerNdjson) + ) - assert.isUndefined(tickerStatus, "Ticker stream must still be running after the unknown-tag failure") - assert.isAbove( - ticksAfterGhost, - ticksBeforeGhost, - "Ticker stream must keep emitting after the unknown-tag failure" - ) - }).pipe(Effect.provide(IsolationClient.pipe(Layer.provideMerge(IsolationServer)))), + it.live( + "with disableFatalDefects a handler defect fails only its own request, not other in-flight streams", + () => + assertTickerSurvives( + Effect.map(RpcClient.make(group), (client) => ({ + ticker: client.Ticker(), + failingCall: Effect.gen(function*() { + const boomExit = yield* client.Boom().pipe(Effect.exit) + if (!Exit.isFailure(boomExit)) { + return assert.fail("Boom call must fail with the handler defect") + } + assert.include( + String(Cause.squash(boomExit.cause)), + "boom", + "the caller must receive the handler defect" + ) + }) + })), + "the handler defect" + ).pipe(Effect.provide(IsolationClient.pipe(Layer.provideMerge(DefectServer)))), { timeout: 30_000 } ) }) diff --git a/packages/platform-node/test/cluster/SocketRunner.test.ts b/packages/platform-node/test/cluster/SocketRunner.test.ts index 721cbf443d2..57d7fda93d2 100644 --- a/packages/platform-node/test/cluster/SocketRunner.test.ts +++ b/packages/platform-node/test/cluster/SocketRunner.test.ts @@ -1,6 +1,7 @@ import { NodeClusterSocket } from "@effect/platform-node" -import { describe, it } from "@effect/vitest" -import { BigDecimal, Effect, Layer, Option, PrimaryKey, Schema } from "effect" +import { assert, describe, it } from "@effect/vitest" +import { BigDecimal, Cause, Effect, Exit, Fiber, Layer, Option, PrimaryKey, Schema } from "effect" +import type { Sharding } from "effect/unstable/cluster" import { ClusterSchema, Entity, @@ -48,8 +49,8 @@ const SharedStorage = Layer.mergeAll( Layer.provide(ShardingConfig.layerDefaults) ) -const makeRunnerLayer = (port: number) => - TestEntityLayer.pipe( +const makeRunnerLayer = (port: number, entities: Layer.Layer) => + entities.pipe( Layer.provideMerge(SocketRunner.layer), Layer.provide(RunnerHealth.layerNoop), Layer.provide(NodeClusterSocket.layerSocketServer), @@ -76,6 +77,30 @@ const makeClientLayer = (port: number) => Layer.provide(RpcSerialization.layerMsgPack) ) +// An entity whose reply cannot be serialized: the handler returns a +// non-integer for a `Schema.Int` success schema, so `Reply.serialize` fails +// on the host runner when encoding the reply for the wire. +const IsolationEntity = Entity + .make("IsolationEntity", [ + Rpc.make("BadReply", { + payload: { id: Schema.Number }, + success: Schema.Int + }), + Rpc.make("Slow", { + success: Schema.String + }) + ]) + .annotateRpcs(ClusterSchema.Persisted, false) + +const IsolationEntityLayer = IsolationEntity.toLayer( + Effect.succeed({ + BadReply: () => Effect.succeed(1.5), + Slow: () => Effect.as(Effect.sleep("2 seconds"), "done") + }) +) + +const ISOLATION_PORT = 50_124 + // BigDecimal.normalize creates a circular `normalized` self-reference. // When a persisted message is sent with discard: true, the notify path in Runners.makeRpc // passes the raw envelope (with circular BigDecimal payload) to the runner via msgpack, @@ -86,7 +111,7 @@ describe("SocketRunner", () => { () => Effect.gen(function*() { // Start the runner (with socket server and entity handler) - yield* Layer.launch(makeRunnerLayer(RUNNER_PORT)).pipe(Effect.forkScoped) + yield* Layer.launch(makeRunnerLayer(RUNNER_PORT, TestEntityLayer)).pipe(Effect.forkScoped) // Give the runner time to start and acquire shards yield* Effect.sleep("2 seconds") @@ -117,4 +142,49 @@ describe("SocketRunner", () => { )), 30_000 ) + + it.live( + "a reply serialization failure fails only its own request", + () => + Effect.gen(function*() { + // Start the runner hosting the entities + yield* Layer.launch(makeRunnerLayer(ISOLATION_PORT, IsolationEntityLayer)).pipe(Effect.forkScoped) + yield* Effect.sleep("2 seconds") + + yield* Effect.gen(function*() { + const makeClient = yield* IsolationEntity.client + // Give the client time to discover the runner + yield* Effect.sleep("3 seconds") + + // a sibling request in flight on the same runner-to-runner connection + const slowFiber = yield* makeClient("slow-entity").Slow().pipe(Effect.forkChild) + yield* Effect.sleep("300 millis") + + const badExit = yield* makeClient("bad-entity").BadReply({ id: 1 }).pipe(Effect.exit) + assert.isTrue(Exit.isFailure(badExit), "the unencodable reply must fail the request") + const failure = Exit.isFailure(badExit) ? String(Cause.squash(badExit.cause)) : "" + assert.include(failure, "MalformedMessage", "the caller must receive the real encode error") + assert.notInclude( + failure, + "AlreadyProcessingMessage", + "the request must not be re-sent into the entity's dedup guard" + ) + + const slowExit = yield* Fiber.await(slowFiber) + assert.isTrue( + Exit.isSuccess(slowExit), + "a sibling in-flight request on the same connection must be unaffected" + ) + if (Exit.isSuccess(slowExit)) { + assert.strictEqual(slowExit.value, "done") + } + }).pipe( + Effect.provide(makeClientLayer(ISOLATION_PORT)), + Effect.scoped + ) + }).pipe(Effect.provide( + SharedStorage + )), + 30_000 + ) })