| 1 | import { assert, describe, it } from "@effect/vitest" |
| 2 | import { Effect, Fiber, Latch, PrimaryKey, Schema } from "effect" |
| 3 | import { ClusterSchema, Entity } from "effect/cluster" |
| 4 | import { Rpc } from "effect/rpc" |
| 5 | import { type Backend, make } from "./harness.ts" |
| 6 | |
| 7 | class Request extends Schema.Class<Request>("ClusterTransportRequest")({ |
| 8 | id: Schema.String |
| 9 | }) { |
| 10 | [PrimaryKey.symbol]() { |
| 11 | return this.id |
| 12 | } |
| 13 | } |
| 14 | |
| 15 | const TransportEntity = Entity.make("ClusterIntegrationTransport", [ |
| 16 | Rpc.make("Persisted", { |
| 17 | payload: Request, |
| 18 | success: Schema.String |
| 19 | }), |
| 20 | Rpc.make("Volatile", { |
| 21 | payload: Request, |
| 22 | success: Schema.String |
| 23 | }).annotate(ClusterSchema.Persisted, false) |
| 24 | ]).annotateRpcs(ClusterSchema.Persisted, true) |
| 25 | |
| 26 | describe("cluster transport integration", () => { |
| 27 | for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) { |
| 28 | it.live(`${backend}: treats a directional socket cut as RunnerUnavailable for volatile and recovers persisted`, () => |
| 29 | Effect.gen(function*() { |
| 30 | const volatileEntered = Latch.makeUnsafe() |
| 31 | const volatileGate = Latch.makeUnsafe() |
| 32 | const persistedEntered = Latch.makeUnsafe() |
| 33 | const persistedGate = Latch.makeUnsafe() |
| 34 | let volatileAttempts = 0 |
| 35 | let persistedAttempts = 0 |
| 36 | |
| 37 | const TransportEntityLayer = TransportEntity.toLayer({ |
| 38 | Persisted: Effect.fnUntraced(function*({ payload }) { |
| 39 | persistedAttempts++ |
| 40 | persistedEntered.openUnsafe() |
| 41 | yield* persistedGate.await |
| 42 | return `persisted:${payload.id}` |
| 43 | }), |
| 44 | Volatile: Effect.fnUntraced(function*({ payload }) { |
| 45 | volatileAttempts++ |
| 46 | volatileEntered.openUnsafe() |
| 47 | yield* volatileGate.await |
| 48 | return `volatile:${payload.id}` |
| 49 | }) |
| 50 | }) |
| 51 | const cluster = yield* make({ backend, entities: TransportEntityLayer }) |
| 52 | const [runner] = yield* cluster.start(1) |
| 53 | yield* cluster.waitForStableAssignments() |
| 54 | const client = yield* cluster.getClient(TransportEntity) |
| 55 | |
| 56 | const volatile = yield* client("volatile").Volatile( |
| 57 | new Request({ id: `${backend}-volatile` }) |
| 58 | ).pipe(Effect.forkChild({ startImmediately: true })) |
| 59 | const persisted = yield* client("persisted").Persisted( |
| 60 | new Request({ id: `${backend}-persisted` }) |
| 61 | ).pipe(Effect.forkChild({ startImmediately: true })) |
| 62 | yield* cluster.waitUntil( |
| 63 | "The volatile request did not enter the runner before the socket cut", |
| 64 | Effect.as(volatileEntered.await, true) |
| 65 | ) |
| 66 | yield* cluster.waitUntil( |
| 67 | "The persisted request did not enter the runner before the socket cut", |
| 68 | Effect.as(persistedEntered.await, true) |
| 69 | ) |
| 70 | yield* cluster.cutSocket(runner, { peer: "client", direction: "inbound" }) |
| 71 | volatileGate.openUnsafe() |
| 72 | persistedGate.openUnsafe() |
| 73 | |
| 74 | assert.strictEqual(yield* Fiber.join(volatile), `volatile:${backend}-volatile`) |
| 75 | assert.strictEqual(yield* Fiber.join(persisted), `persisted:${backend}-persisted`) |
| 76 | assert.strictEqual(runner.state(), "running") |
| 77 | // Volatile requests have no storage fallback, so the second attempt |
| 78 | // proves the cut surfaced as RunnerUnavailable and was retried. |
| 79 | assert.strictEqual(volatileAttempts, 2) |
| 80 | // The persisted caller instead recovered the stored reply. |
| 81 | assert.strictEqual(persistedAttempts, 1) |
| 82 | })) |
| 83 | } |
| 84 | }) |