| 1 | import { assert, describe, it } from "@effect/vitest" |
| 2 | import { Effect, Fiber, Latch, 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 | const ResidencyEntity = Entity.make("ClusterIntegrationResidency", [ |
| 8 | Rpc.make("Ping", { |
| 9 | payload: { id: Schema.String }, |
| 10 | success: Schema.String |
| 11 | }), |
| 12 | Rpc.make("Hold", { |
| 13 | payload: { id: Schema.String } |
| 14 | }) |
| 15 | ]).annotateRpcs(ClusterSchema.Persisted, true) |
| 16 | |
| 17 | let holdGate = Latch.makeUnsafe() |
| 18 | |
| 19 | const ResidencyEntityLayer = ResidencyEntity.toLayer({ |
| 20 | Ping: ({ payload }) => Effect.succeed(payload.id), |
| 21 | Hold: () => holdGate.await |
| 22 | }, { maxIdleTime: "1 second", concurrency: 2 }) |
| 23 | |
| 24 | describe("cluster residency integration", () => { |
| 25 | for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) { |
| 26 | it.live(`${backend}: stays at the entity cap and keeps residents moving`, () => |
| 27 | Effect.gen(function*() { |
| 28 | holdGate = Latch.makeUnsafe() |
| 29 | const cluster = yield* make({ |
| 30 | backend, |
| 31 | entities: ResidencyEntityLayer, |
| 32 | config: { maxResidentEntities: 3 } |
| 33 | }) |
| 34 | const [runner] = yield* cluster.start(1) |
| 35 | yield* cluster.waitForStableAssignments() |
| 36 | const client = yield* cluster.getClient(ResidencyEntity) |
| 37 | |
| 38 | // fill every entity slot with a held request |
| 39 | const held = yield* Effect.forEach( |
| 40 | ["held-1", "held-2", "held-3"], |
| 41 | (id) => client(id).Hold({ id }).pipe(Effect.forkChild({ startImmediately: true })) |
| 42 | ) |
| 43 | yield* cluster.waitUntil( |
| 44 | "The held entities did not become resident", |
| 45 | Effect.map(runner.sharding.activeEntityCount, (count) => count === 3) |
| 46 | ) |
| 47 | |
| 48 | // a backlog of messages for new entity ids stays in storage |
| 49 | for (let i = 1; i <= 8; i++) { |
| 50 | yield* client(`backlog-${i}`).Ping({ id: `backlog-${i}` }, { discard: true }) |
| 51 | } |
| 52 | |
| 53 | // resident entities keep receiving messages while at the cap |
| 54 | assert.strictEqual(yield* client("held-1").Ping({ id: "first" }), "first") |
| 55 | assert.strictEqual(yield* client("held-2").Ping({ id: "second" }), "second") |
| 56 | |
| 57 | // the runner never spawned entities for the backlog |
| 58 | assert.strictEqual(yield* runner.sharding.activeEntityCount, 3) |
| 59 | assert.isAtLeast(yield* cluster.unprocessedMessageCount, 8) |
| 60 | |
| 61 | // release the held entities: as slots free up, the backlog drains in |
| 62 | // waves without ever exceeding the cap |
| 63 | holdGate.openUnsafe() |
| 64 | yield* Fiber.joinAll(held) |
| 65 | yield* cluster.waitUntil( |
| 66 | "The backlog did not drain after entity slots freed up", |
| 67 | Effect.gen(function*() { |
| 68 | assert.isAtMost(yield* runner.sharding.activeEntityCount, 3) |
| 69 | const counts = yield* cluster.messageCounts() |
| 70 | return counts.unprocessed === 0 && counts.failed === 0 |
| 71 | }), |
| 72 | "45 seconds" |
| 73 | ) |
| 74 | })) |
| 75 | } |
| 76 | }) |