| 1 | import { assert, describe, it } from "@effect/vitest" |
| 2 | import { Effect, Fiber, Queue, Schema, Stream } from "effect" |
| 3 | import { ClusterSchema, Entity, EntityAddress, EntityId, EntityType, type Envelope, Snowflake } from "effect/cluster" |
| 4 | import { Headers } from "effect/http" |
| 5 | import { Rpc } from "effect/rpc" |
| 6 | import { make, makeRawRunnerClient } from "./harness.ts" |
| 7 | |
| 8 | const WatchRpc = Rpc.make("Watch", { payload: { id: Schema.String }, success: Schema.Int, stream: true }) |
| 9 | const WatchEntity = Entity.make("ClusterIntegrationStreamDisconnect", [WatchRpc]) |
| 10 | .annotateRpcs(ClusterSchema.Persisted, false) |
| 11 | |
| 12 | const setup = Effect.fnUntraced(function*() { |
| 13 | const counts = { starts: 0, stops: 0 } |
| 14 | const layer = WatchEntity.toLayer({ |
| 15 | Watch: () => { |
| 16 | counts.starts++ |
| 17 | return Rpc.fork(Stream.never.pipe(Stream.ensuring(Effect.sync(() => { |
| 18 | counts.stops++ |
| 19 | })))) |
| 20 | } |
| 21 | }) |
| 22 | const cluster = yield* make({ backend: "pg", entities: layer }) |
| 23 | const [owner] = yield* cluster.start(1) |
| 24 | yield* cluster.waitForStableAssignments() |
| 25 | const entityId = "watch-1" |
| 26 | return { cluster, counts, owner, entityId } |
| 27 | }) |
| 28 | |
| 29 | describe("cross-runner disconnect cleanup", () => { |
| 30 | it.live("dead caller: socket closed with no further messages", () => |
| 31 | Effect.gen(function*() { |
| 32 | const { cluster, counts, owner, entityId } = yield* setup() |
| 33 | const raw = yield* makeRawRunnerClient(owner) |
| 34 | const id = EntityId.make(entityId) |
| 35 | const snowflake = yield* Snowflake.Generator.pipe(Effect.provide(Snowflake.layerGenerator)) |
| 36 | const request: Envelope.PartialRequest = { |
| 37 | _tag: "Request", |
| 38 | requestId: snowflake.nextUnsafe(), |
| 39 | address: EntityAddress.make({ |
| 40 | shardId: owner.sharding.getShardId(id, WatchEntity.getShardGroup(id)), |
| 41 | entityType: EntityType.make(WatchEntity.type), |
| 42 | entityId: id |
| 43 | }), |
| 44 | tag: "Watch", |
| 45 | payload: yield* Schema.encodeEffect(raw.codecFor(WatchRpc.payloadSchema))({ id: "x" }), |
| 46 | headers: Headers.empty |
| 47 | } |
| 48 | const caller = yield* raw.client.Stream({ request, persisted: false }, { asQueue: true }).pipe( |
| 49 | Effect.flatMap((queue) => Effect.forever(Queue.take(queue))), |
| 50 | Effect.scoped, |
| 51 | Effect.forkChild({ startImmediately: true }) |
| 52 | ) |
| 53 | yield* cluster.waitUntil("host handler did not start", Effect.sync(() => counts.starts === 1)) |
| 54 | assert.deepStrictEqual(counts, { starts: 1, stops: 0 }) |
| 55 | |
| 56 | // The caller "dies": its socket closes and nothing else is ever sent. |
| 57 | yield* raw.disconnect |
| 58 | yield* cluster.waitUntil("host handler did not stop after the caller died", Effect.sync(() => counts.stops === 1)) |
| 59 | assert.deepStrictEqual(counts, { starts: 1, stops: 1 }) |
| 60 | yield* Fiber.interrupt(caller) |
| 61 | })) |
| 62 | }) |