From 4f0a73ee34862760c50e9c0b624b31e70a26bb37 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 09:52:34 +1200 Subject: [PATCH 1/9] Add cluster integration test harness --- .github/workflows/cluster-integration.yml | 26 +++++ .../ClusterHarness.test.ts | 37 ++++++ .../cluster-integration/ClusterHarness.ts | 110 ++++++++++++++++++ vitest.config.ts | 2 + 4 files changed, 175 insertions(+) create mode 100644 .github/workflows/cluster-integration.yml create mode 100644 packages/platform-node/test/cluster-integration/ClusterHarness.test.ts create mode 100644 packages/platform-node/test/cluster-integration/ClusterHarness.ts diff --git a/.github/workflows/cluster-integration.yml b/.github/workflows/cluster-integration.yml new file mode 100644 index 00000000000..dc3bd98c5f2 --- /dev/null +++ b/.github/workflows/cluster-integration.yml @@ -0,0 +1,26 @@ +name: Cluster Integration +on: + workflow_dispatch: + +permissions: {} + +jobs: + test: + name: Test + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + EFFECT_CLUSTER_INTEGRATION_TESTS: "1" + permissions: + contents: read + steps: + - uses: actions/checkout@v6 + - name: Pre-pull test container images + run: | + docker pull testcontainers/ryuk:0.14.0 & + docker pull postgres:alpine & + wait + - name: Install dependencies + uses: ./.github/actions/setup + - name: Test + run: pnpm test --run packages/platform-node/test/cluster-integration diff --git a/packages/platform-node/test/cluster-integration/ClusterHarness.test.ts b/packages/platform-node/test/cluster-integration/ClusterHarness.test.ts new file mode 100644 index 00000000000..f2d929a45ef --- /dev/null +++ b/packages/platform-node/test/cluster-integration/ClusterHarness.test.ts @@ -0,0 +1,37 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, PrimaryKey, Schema } from "effect" +import { ClusterSchema, Entity } from "effect/unstable/cluster" +import { Rpc } from "effect/unstable/rpc" +import { make } from "./ClusterHarness.ts" + +class Ping extends Schema.Class("Ping")({ + id: Schema.String +}) { + [PrimaryKey.symbol]() { + return this.id + } +} + +const TestEntity = Entity.make("ClusterIntegrationTestEntity", [ + Rpc.make("Ping", { + payload: Ping, + success: Schema.String + }) +]).annotateRpcs(ClusterSchema.Persisted, true) + +const TestEntityLayer = TestEntity.toLayer({ + Ping: ({ payload }) => Effect.succeed(`pong:${payload.id}`) +}) + +describe("ClusterHarness", () => { + it.live("runs a persisted entity message on a real multi-runner cluster", () => + Effect.gen(function*() { + const harness = yield* make(TestEntityLayer) + yield* harness.start(2) + + const client = yield* harness.getClient(TestEntity) + const result = yield* client("entity-1").Ping(new Ping({ id: "request-1" })) + + assert.strictEqual(result, "pong:request-1") + }).pipe(Effect.scoped), 120_000) +}) diff --git a/packages/platform-node/test/cluster-integration/ClusterHarness.ts b/packages/platform-node/test/cluster-integration/ClusterHarness.ts new file mode 100644 index 00000000000..3ca56396b7f --- /dev/null +++ b/packages/platform-node/test/cluster-integration/ClusterHarness.ts @@ -0,0 +1,110 @@ +import { NodeClusterSocket, NodeCrypto } from "@effect/platform-node" +import { Effect, Exit, Layer, Option, Scope } from "effect" +import type { Entity, Sharding } from "effect/unstable/cluster" +import { + RunnerAddress, + RunnerHealth, + ShardingConfig, + SocketRunner, + SqlMessageStorage, + SqlRunnerStorage +} from "effect/unstable/cluster" +import type { Rpc } from "effect/unstable/rpc" +import { RpcSerialization } from "effect/unstable/rpc" +import { PgContainer } from "../fixtures/pg-utils.ts" + +const clusterConfig = { + entityMessagePollInterval: 100, + entityReplyPollInterval: 50, + entityTerminationTimeout: 0, + refreshAssignmentsInterval: 100, + sendRetryInterval: 50 +} + +const StorageLive = Layer.mergeAll( + SqlMessageStorage.layer, + SqlRunnerStorage.layer +).pipe( + Layer.provide(PgContainer.layerClient), + Layer.provide(NodeCrypto.layer), + Layer.provide(ShardingConfig.layer(clusterConfig)), + Layer.orDie +) + +let nextPort = 40_000 + (process.pid % 1000) * 16 + +const runnerLayer = ( + address: RunnerAddress.RunnerAddress, + entities: Layer.Layer +) => + entities.pipe( + Layer.provideMerge(SocketRunner.layer), + Layer.provide(RunnerHealth.layerNoop), + Layer.provide(NodeClusterSocket.layerSocketServer), + Layer.provide(NodeClusterSocket.layerClientProtocol), + Layer.provide(ShardingConfig.layer({ + ...clusterConfig, + runnerAddress: Option.some(address) + })), + Layer.provide(RpcSerialization.layerMsgPack) + ) + +const clientLayer = SocketRunner.layerClientOnly.pipe( + Layer.provide(NodeClusterSocket.layerClientProtocol), + Layer.provide(ShardingConfig.layer(clusterConfig)), + Layer.provide(RpcSerialization.layerMsgPack) +) + +export const make = Effect.fnUntraced(function*( + entities: Layer.Layer +) { + const parentScope = yield* Effect.scope + const storageScope = yield* Scope.fork(parentScope) + const storage = yield* Layer.buildWithScope(StorageLive, storageScope) + const clientScope = yield* Scope.fork(parentScope) + const client = yield* clientLayer.pipe( + Layer.buildWithScope(clientScope), + Effect.provide(storage) + ) + const runners = new Map() + const firstPort = nextPort + nextPort += 16 + + const startRunner = Effect.fnUntraced(function*(index: number) { + if (runners.has(index)) return + const scope = yield* Scope.fork(parentScope) + const address = RunnerAddress.make("localhost", firstPort + index) + yield* runnerLayer(address, entities).pipe( + Layer.buildWithScope(scope), + Effect.provide(storage) + ) + runners.set(index, scope) + }) + + const kill = Effect.fnUntraced(function*(index: number) { + const scope = runners.get(index) + if (scope === undefined) return + yield* Scope.close(scope, Exit.void) + runners.delete(index) + }) + + const start = Effect.fnUntraced(function*(runnerCount: number) { + yield* Effect.forEach( + Array.from({ length: runnerCount }, (_, index) => index), + startRunner, + { discard: true } + ) + yield* Effect.sleep(1000) + }) + + const restart = Effect.fnUntraced(function*(index: number) { + yield* kill(index) + yield* startRunner(index) + yield* Effect.sleep(1000) + }) + + const getClient = (entity: Entity.Entity) => + entity.client.pipe(Effect.provide(client)) + + return { start, kill, restart, getClient } as const +}) diff --git a/vitest.config.ts b/vitest.config.ts index 2e6a2f07847..1be92940a1e 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -9,6 +9,7 @@ const isNode = typeof process !== "undefined" && !isDeno && !isBun const integrationTestsEnabled = process.env.EFFECT_INTEGRATION_TESTS === "1" +const clusterIntegrationTestsEnabled = process.env.EFFECT_CLUSTER_INTEGRATION_TESTS === "1" const project = (name: string, directory: string, include: boolean = true, config: ViteUserConfig = {}) => { if (!include) { @@ -32,6 +33,7 @@ export const exclude = [ "**/typetest/**", "**/coverage/**", "**/test/utils/**", + ...(!clusterIntegrationTestsEnabled ? ["**/cluster-integration/**"] : []), ...(!integrationTestsEnabled ? ["**/*.integration.test.{ts,tsx}"] : []), "**/*.d.ts", "**/*.config.*", From eb84c817652236824614a0b7c918389c76b9c7d9 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 11:28:09 +1200 Subject: [PATCH 2/9] Complete cluster integration test harness --- .../{cluster-integration.yml => cluster.yml} | 9 +- package.json | 1 + .../ClusterHarness.test.ts | 37 -- .../cluster-integration/ClusterHarness.ts | 110 ----- .../test/cluster-integration/README.md | 35 ++ .../test/cluster-integration/Smoke.test.ts | 56 +++ .../test/cluster-integration/globalSetup.ts | 27 ++ .../test/cluster-integration/harness.ts | 412 ++++++++++++++++++ vitest.config.ts | 42 +- 9 files changed, 575 insertions(+), 154 deletions(-) rename .github/workflows/{cluster-integration.yml => cluster.yml} (76%) delete mode 100644 packages/platform-node/test/cluster-integration/ClusterHarness.test.ts delete mode 100644 packages/platform-node/test/cluster-integration/ClusterHarness.ts create mode 100644 packages/platform-node/test/cluster-integration/README.md create mode 100644 packages/platform-node/test/cluster-integration/Smoke.test.ts create mode 100644 packages/platform-node/test/cluster-integration/globalSetup.ts create mode 100644 packages/platform-node/test/cluster-integration/harness.ts diff --git a/.github/workflows/cluster-integration.yml b/.github/workflows/cluster.yml similarity index 76% rename from .github/workflows/cluster-integration.yml rename to .github/workflows/cluster.yml index dc3bd98c5f2..1ca596b8d2c 100644 --- a/.github/workflows/cluster-integration.yml +++ b/.github/workflows/cluster.yml @@ -8,19 +8,22 @@ jobs: test: name: Test runs-on: ubuntu-latest - timeout-minutes: 15 + timeout-minutes: 30 env: - EFFECT_CLUSTER_INTEGRATION_TESTS: "1" + EFFECT_CLUSTER_TESTS: "1" permissions: contents: read steps: - uses: actions/checkout@v6 + - name: Pre-pull test container images run: | docker pull testcontainers/ryuk:0.14.0 & docker pull postgres:alpine & + docker pull mysql:lts & wait + - name: Install dependencies uses: ./.github/actions/setup - name: Test - run: pnpm test --run packages/platform-node/test/cluster-integration + run: pnpm test-cluster diff --git a/package.json b/package.json index b071186b5bb..3d970777913 100644 --- a/package.json +++ b/package.json @@ -13,6 +13,7 @@ "bundle-compare-selected": "bash scripts/bundle-compare-selected.sh", "circular": "node scripts/circular.mjs", "test": "vitest", + "test-cluster": "vitest run --project cluster-integration", "doctest": "vitest --config vitest.docs.ts", "coverage": "vitest --coverage", "check": "tsc -b tsconfig.json", diff --git a/packages/platform-node/test/cluster-integration/ClusterHarness.test.ts b/packages/platform-node/test/cluster-integration/ClusterHarness.test.ts deleted file mode 100644 index f2d929a45ef..00000000000 --- a/packages/platform-node/test/cluster-integration/ClusterHarness.test.ts +++ /dev/null @@ -1,37 +0,0 @@ -import { assert, describe, it } from "@effect/vitest" -import { Effect, PrimaryKey, Schema } from "effect" -import { ClusterSchema, Entity } from "effect/unstable/cluster" -import { Rpc } from "effect/unstable/rpc" -import { make } from "./ClusterHarness.ts" - -class Ping extends Schema.Class("Ping")({ - id: Schema.String -}) { - [PrimaryKey.symbol]() { - return this.id - } -} - -const TestEntity = Entity.make("ClusterIntegrationTestEntity", [ - Rpc.make("Ping", { - payload: Ping, - success: Schema.String - }) -]).annotateRpcs(ClusterSchema.Persisted, true) - -const TestEntityLayer = TestEntity.toLayer({ - Ping: ({ payload }) => Effect.succeed(`pong:${payload.id}`) -}) - -describe("ClusterHarness", () => { - it.live("runs a persisted entity message on a real multi-runner cluster", () => - Effect.gen(function*() { - const harness = yield* make(TestEntityLayer) - yield* harness.start(2) - - const client = yield* harness.getClient(TestEntity) - const result = yield* client("entity-1").Ping(new Ping({ id: "request-1" })) - - assert.strictEqual(result, "pong:request-1") - }).pipe(Effect.scoped), 120_000) -}) diff --git a/packages/platform-node/test/cluster-integration/ClusterHarness.ts b/packages/platform-node/test/cluster-integration/ClusterHarness.ts deleted file mode 100644 index 3ca56396b7f..00000000000 --- a/packages/platform-node/test/cluster-integration/ClusterHarness.ts +++ /dev/null @@ -1,110 +0,0 @@ -import { NodeClusterSocket, NodeCrypto } from "@effect/platform-node" -import { Effect, Exit, Layer, Option, Scope } from "effect" -import type { Entity, Sharding } from "effect/unstable/cluster" -import { - RunnerAddress, - RunnerHealth, - ShardingConfig, - SocketRunner, - SqlMessageStorage, - SqlRunnerStorage -} from "effect/unstable/cluster" -import type { Rpc } from "effect/unstable/rpc" -import { RpcSerialization } from "effect/unstable/rpc" -import { PgContainer } from "../fixtures/pg-utils.ts" - -const clusterConfig = { - entityMessagePollInterval: 100, - entityReplyPollInterval: 50, - entityTerminationTimeout: 0, - refreshAssignmentsInterval: 100, - sendRetryInterval: 50 -} - -const StorageLive = Layer.mergeAll( - SqlMessageStorage.layer, - SqlRunnerStorage.layer -).pipe( - Layer.provide(PgContainer.layerClient), - Layer.provide(NodeCrypto.layer), - Layer.provide(ShardingConfig.layer(clusterConfig)), - Layer.orDie -) - -let nextPort = 40_000 + (process.pid % 1000) * 16 - -const runnerLayer = ( - address: RunnerAddress.RunnerAddress, - entities: Layer.Layer -) => - entities.pipe( - Layer.provideMerge(SocketRunner.layer), - Layer.provide(RunnerHealth.layerNoop), - Layer.provide(NodeClusterSocket.layerSocketServer), - Layer.provide(NodeClusterSocket.layerClientProtocol), - Layer.provide(ShardingConfig.layer({ - ...clusterConfig, - runnerAddress: Option.some(address) - })), - Layer.provide(RpcSerialization.layerMsgPack) - ) - -const clientLayer = SocketRunner.layerClientOnly.pipe( - Layer.provide(NodeClusterSocket.layerClientProtocol), - Layer.provide(ShardingConfig.layer(clusterConfig)), - Layer.provide(RpcSerialization.layerMsgPack) -) - -export const make = Effect.fnUntraced(function*( - entities: Layer.Layer -) { - const parentScope = yield* Effect.scope - const storageScope = yield* Scope.fork(parentScope) - const storage = yield* Layer.buildWithScope(StorageLive, storageScope) - const clientScope = yield* Scope.fork(parentScope) - const client = yield* clientLayer.pipe( - Layer.buildWithScope(clientScope), - Effect.provide(storage) - ) - const runners = new Map() - const firstPort = nextPort - nextPort += 16 - - const startRunner = Effect.fnUntraced(function*(index: number) { - if (runners.has(index)) return - const scope = yield* Scope.fork(parentScope) - const address = RunnerAddress.make("localhost", firstPort + index) - yield* runnerLayer(address, entities).pipe( - Layer.buildWithScope(scope), - Effect.provide(storage) - ) - runners.set(index, scope) - }) - - const kill = Effect.fnUntraced(function*(index: number) { - const scope = runners.get(index) - if (scope === undefined) return - yield* Scope.close(scope, Exit.void) - runners.delete(index) - }) - - const start = Effect.fnUntraced(function*(runnerCount: number) { - yield* Effect.forEach( - Array.from({ length: runnerCount }, (_, index) => index), - startRunner, - { discard: true } - ) - yield* Effect.sleep(1000) - }) - - const restart = Effect.fnUntraced(function*(index: number) { - yield* kill(index) - yield* startRunner(index) - yield* Effect.sleep(1000) - }) - - const getClient = (entity: Entity.Entity) => - entity.client.pipe(Effect.provide(client)) - - return { start, kill, restart, getClient } as const -}) diff --git a/packages/platform-node/test/cluster-integration/README.md b/packages/platform-node/test/cluster-integration/README.md new file mode 100644 index 00000000000..b49b389d050 --- /dev/null +++ b/packages/platform-node/test/cluster-integration/README.md @@ -0,0 +1,35 @@ +# Cluster integration tests + +This suite runs multi-runner clusters against shared PostgreSQL and MySQL +containers. It is excluded from the default test projects and only registered +when `EFFECT_CLUSTER_TESTS=1`. + +```sh +EFFECT_CLUSTER_TESTS=1 pnpm test-cluster +``` + +Docker must be running. Vitest global setup starts one `postgres:alpine` and one +`mysql:lts` container for the entire project. Every cluster uses a unique table +prefix, and every runner listens on an operating-system-assigned port. + +## Harness + +`harness.ts` exposes: + +- `make({ backend, entities, lockMode })` to create a scoped cluster harness. +- `start(count)` to start in-process socket runners and a client over msgpack. +- `stop(runner)` for graceful deregistration and shard handoff. +- `kill(runner)` for abrupt teardown without deregistration or explicit lock cleanup. +- `freeze(runner)` to suspend SQL heartbeats and lock refresh while leaving the runner's sockets and reserved SQL connection open. +- `waitUntil`, `waitForStableAssignments`, and `waitForEntityOwner` for deadline-based polling with cluster diagnostics on failure. +- `messageCounts`, `unprocessedMessageCount`, `repliedMessageCount`, and `failedMessageCount` for storage assertions scoped to the cluster prefix. + +Advisory locks are owned by the reserved database session. A frozen advisory-lock +runner therefore keeps its locks until it is stopped or killed. Row-lock mode +uses expiry-driven takeover while frozen. + +## Adding a test + +Add `*.test.ts` under this directory. Test files define entities and assertions; +all cluster startup, lifecycle, waiting, and storage inspection belongs in the +harness. Use the harness polling helpers instead of calling `Effect.sleep`. diff --git a/packages/platform-node/test/cluster-integration/Smoke.test.ts b/packages/platform-node/test/cluster-integration/Smoke.test.ts new file mode 100644 index 00000000000..e4dd18c534e --- /dev/null +++ b/packages/platform-node/test/cluster-integration/Smoke.test.ts @@ -0,0 +1,56 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, PrimaryKey, Schema } from "effect" +import { ClusterSchema, Entity } from "effect/unstable/cluster" +import { Rpc } from "effect/unstable/rpc" +import { type Backend, make } from "./harness.ts" + +class Ping extends Schema.Class("Ping")({ + id: Schema.String +}) { + [PrimaryKey.symbol]() { + return this.id + } +} + +const TestEntity = Entity.make("ClusterIntegrationSmoke", [ + Rpc.make("Ping", { + payload: Ping, + success: Schema.String + }) +]).annotateRpcs(ClusterSchema.Persisted, true) + +const TestEntityLayer = TestEntity.toLayer({ + Ping: ({ payload }) => Effect.succeed(`pong:${payload.id}`) +}) + +describe("cluster integration smoke", () => { + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: persists messages and rebalances after an abrupt runner death`, () => + Effect.gen(function*() { + const cluster = yield* make({ backend, entities: TestEntityLayer }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + + const client = yield* cluster.getClient(TestEntity) + assert.strictEqual( + yield* client("entity-1").Ping(new Ping({ id: "request-1" })), + "pong:request-1" + ) + + yield* cluster.waitUntil( + "The smoke entity had no active owner", + Effect.map(cluster.ownerOfEntity(TestEntity, "entity-1"), (owner) => owner !== undefined) + ) + const owner = yield* cluster.ownerOfEntity(TestEntity, "entity-1") + yield* cluster.freeze(owner!) + yield* cluster.kill(owner!) + yield* cluster.waitForStableAssignments() + + assert.strictEqual( + yield* client("entity-1").Ping(new Ping({ id: "request-2" })), + "pong:request-2" + ) + assert.strictEqual(yield* cluster.repliedMessageCount, 2) + }).pipe(Effect.scoped)) + } +}) diff --git a/packages/platform-node/test/cluster-integration/globalSetup.ts b/packages/platform-node/test/cluster-integration/globalSetup.ts new file mode 100644 index 00000000000..cc7e5666c42 --- /dev/null +++ b/packages/platform-node/test/cluster-integration/globalSetup.ts @@ -0,0 +1,27 @@ +import { MySqlContainer } from "@testcontainers/mysql" +import { PostgreSqlContainer } from "@testcontainers/postgresql" +import type { TestProject } from "vitest/node" + +export interface ClusterDatabases { + readonly mysql: string + readonly pg: string +} + +declare module "vitest" { + export interface ProvidedContext { + readonly clusterDatabases: ClusterDatabases + } +} + +export default function setup(project: TestProject) { + return Promise.all([ + new PostgreSqlContainer("postgres:alpine").start(), + new MySqlContainer("mysql:lts").start() + ]).then(([pg, mysql]) => { + project.provide("clusterDatabases", { + mysql: mysql.getConnectionUri(), + pg: pg.getConnectionUri() + }) + return () => Promise.all([pg.stop(), mysql.stop()]).then(() => undefined) + }) +} diff --git a/packages/platform-node/test/cluster-integration/harness.ts b/packages/platform-node/test/cluster-integration/harness.ts new file mode 100644 index 00000000000..1e83abe80be --- /dev/null +++ b/packages/platform-node/test/cluster-integration/harness.ts @@ -0,0 +1,412 @@ +import { NodeClusterSocket, NodeCrypto, NodeSocketServer } from "@effect/platform-node" +import { MysqlClient } from "@effect/sql-mysql2" +import { PgClient } from "@effect/sql-pg" +import { Clock, Context, Duration, Effect, Exit, Latch, Layer, Option, Redacted, Scope } from "effect" +import { + type Entity, + EntityId, + type Runner as RunnerModel, + RunnerAddress, + RunnerHealth, + Runners, + RunnerStorage, + ShardId, + Sharding, + ShardingConfig, + SocketRunner, + SqlMessageStorage, + SqlRunnerStorage +} from "effect/unstable/cluster" +import type { Rpc } from "effect/unstable/rpc" +import { RpcSerialization } from "effect/unstable/rpc" +import * as SocketServer from "effect/unstable/socket/SocketServer" +import { SqlClient } from "effect/unstable/sql" +import { inject } from "vitest" + +export type Backend = "mysql" | "pg" +export type LockMode = "advisory" | "row" + +export interface ClusterRunner { + readonly address: RunnerAddress.RunnerAddress + readonly index: number + readonly sharding: Sharding.Sharding["Service"] + readonly state: () => "frozen" | "killed" | "running" | "stopped" +} + +export interface MessageCounts { + readonly failed: number + readonly replied: number + readonly unprocessed: number +} + +export interface MakeOptions { + readonly backend: Backend + readonly entities: Layer.Layer + readonly lockMode?: LockMode | undefined + readonly runnerLayer?: RunnerLayer | undefined +} + +interface RegistrationRow { + readonly address: string + readonly healthy: boolean | number + readonly last_heartbeat: Date | string + readonly runner: unknown +} + +interface MessageRow { + readonly processed: boolean | number + readonly reply_payload: string | Record | null +} + +interface RunnerEntry extends ClusterRunner { + readonly controller: ReturnType + readonly scope: Scope.Closeable + setState(state: ReturnType): void +} + +const clusterConfig = { + entityMaxIdleTime: 3_000, + entityMessagePollInterval: 500, + refreshAssignmentsInterval: 150, + shardLockExpiration: 1_750, + shardLockRefreshInterval: 500 +} as const + +let nextCluster = 0 + +const makeRunnerStorageController = (storage: RunnerStorage.RunnerStorage["Service"]) => { + const gate = Latch.makeUnsafe(true) + const refreshPaused = Latch.makeUnsafe() + const syncPaused = Latch.makeUnsafe() + let mode: "frozen" | "killed" | "running" = "running" + let lastRunners: Array = [] + + const waitWhileFrozen = ( + paused: Latch.Latch, + effect: Effect.Effect, + onKilled: () => A + ): Effect.Effect => + Effect.suspend(() => { + if (mode === "running") return effect + if (mode === "killed") return Effect.succeed(onKilled()) + paused.openUnsafe() + return gate.await.pipe( + Effect.uninterruptible, + Effect.andThen(Effect.suspend(() => mode === "running" ? effect : Effect.succeed(onKilled()))) + ) + }) + + const controlled = RunnerStorage.RunnerStorage.of({ + ...storage, + getRunners: waitWhileFrozen( + syncPaused, + storage.getRunners.pipe(Effect.tap((runners) => Effect.sync(() => lastRunners = runners))), + () => lastRunners + ), + refresh: (address, shardIds) => { + const shards = Array.from(shardIds) + return waitWhileFrozen(refreshPaused, storage.refresh(address, shards), () => shards) + }, + release: (address, shardId) => mode === "killed" ? Effect.void : storage.release(address, shardId), + releaseAll: (address) => mode === "killed" ? Effect.void : storage.releaseAll(address), + unregister: (address) => mode === "killed" ? Effect.void : storage.unregister(address) + }) + + return { + controlled, + freeze: Effect.sync(() => { + mode = "frozen" + gate.closeUnsafe() + }).pipe(Effect.andThen(Effect.all([refreshPaused.await, syncPaused.await], { discard: true }))), + kill: Effect.sync(() => { + mode = "killed" + gate.openUnsafe() + }), + resume: Effect.sync(() => { + mode = "running" + gate.openUnsafe() + }) + } +} + +const RunnerHealthLive = RunnerHealth.layerPing.pipe( + Layer.provide(Runners.layerRpc), + Layer.provide(NodeClusterSocket.layerClientProtocol) +) + +export const socketRunnerLayer = ( + address: RunnerAddress.RunnerAddress, + entities: Layer.Layer, + socketServer: SocketServer.SocketServer["Service"], + config: typeof clusterConfig & { readonly shardLockDisableAdvisory: boolean } +) => + entities.pipe( + Layer.provideMerge(SocketRunner.layer), + Layer.provide(RunnerHealthLive), + Layer.provide(Layer.succeed(SocketServer.SocketServer, socketServer)), + Layer.provide(NodeClusterSocket.layerClientProtocol), + Layer.provide(ShardingConfig.layer({ + ...config, + runnerAddress: Option.some(address) + })), + Layer.provide(RpcSerialization.layerMsgPack) + ) + +export type RunnerLayer = typeof socketRunnerLayer + +const clientLayer = ( + config: typeof clusterConfig & { readonly shardLockDisableAdvisory: boolean } +) => + SocketRunner.layerClientOnly.pipe( + Layer.provide(NodeClusterSocket.layerClientProtocol), + Layer.provide(ShardingConfig.layer(config)), + Layer.provide(RpcSerialization.layerMsgPack) + ) + +const parseReply = (payload: MessageRow["reply_payload"]): Record | undefined => { + if (payload === null) return undefined + return typeof payload === "string" ? JSON.parse(payload) : payload +} + +export const make = Effect.fnUntraced(function*(options: MakeOptions) { + const parentScope = yield* Effect.scope + const prefix = `cluster_${process.pid}_${nextCluster++}` + const config = { + ...clusterConfig, + shardLockDisableAdvisory: options.lockMode === "row" + } + const databases = inject("clusterDatabases") + const databaseLayer = options.backend === "pg" + ? PgClient.layer({ url: Redacted.make(databases.pg), maxConnections: 32 }) + : MysqlClient.layer({ url: Redacted.make(databases.mysql), maxConnections: 32 }) + const database = yield* Layer.buildWithScope(databaseLayer, parentScope) + const sql = Context.get(database, SqlClient.SqlClient).withoutTransforms() + const messageStorage = yield* SqlMessageStorage.layerWith({ prefix }).pipe( + Layer.provide(NodeCrypto.layer), + Layer.provide(ShardingConfig.layer(config)), + Layer.buildWithScope(parentScope), + Effect.provide(database) + ) + const shared = Context.merge(database, messageStorage) + const runners: Array = [] + + const makeRunnerStorage = Effect.fnUntraced(function*(scope: Scope.Closeable) { + return yield* SqlRunnerStorage.layerWith({ prefix }).pipe( + Layer.provide(ShardingConfig.layer(config)), + Layer.orDie, + Layer.buildWithScope(scope), + Effect.provide(database) + ) + }) + + const clientScope = yield* Scope.fork(parentScope) + const clientStorage = yield* makeRunnerStorage(clientScope) + const client = yield* clientLayer(config).pipe( + Layer.buildWithScope(clientScope), + Effect.provide(Context.merge(shared, clientStorage)) + ) + const startRunner = Effect.fnUntraced(function*(index: number) { + const scope = yield* Scope.fork(parentScope) + const serverContext = yield* NodeSocketServer.layer({ host: "127.0.0.1", port: 0 }).pipe( + Layer.buildWithScope(scope) + ) + const socketServer = Context.get(serverContext, SocketServer.SocketServer) + if (socketServer.address._tag !== "TcpAddress") { + return yield* Effect.die("Expected a TCP socket server") + } + const address = RunnerAddress.make("127.0.0.1", socketServer.address.port) + const rawStorageContext = yield* makeRunnerStorage(scope) + const controller = makeRunnerStorageController(Context.get(rawStorageContext, RunnerStorage.RunnerStorage)) + const storage = Context.make(RunnerStorage.RunnerStorage, controller.controlled) + const context = yield* (options.runnerLayer ?? socketRunnerLayer)( + address, + options.entities, + socketServer, + config + ).pipe( + Layer.buildWithScope(scope), + Effect.provide(Context.mergeAll(shared, storage)) + ) + const sharding = Context.get(context, Sharding.Sharding) + let state: ReturnType = "running" + const runner: RunnerEntry = { + address, + controller, + index, + scope, + setState(next) { + state = next + }, + sharding, + state: () => state + } + runners.push(runner) + return runner as ClusterRunner + }) + + const start = Effect.fnUntraced(function*(runnerCount: number) { + return yield* Effect.forEach( + Array.from({ length: runnerCount }, (_, index) => index), + startRunner + ) + }) + + const entryFor = (runner: ClusterRunner) => runners.find((entry) => entry === runner)! + + const stop = Effect.fnUntraced(function*(runner: ClusterRunner) { + const entry = entryFor(runner) + if (entry.state() === "stopped" || entry.state() === "killed") return + if (entry.state() === "frozen") yield* entry.controller.resume + entry.setState("stopped") + yield* Scope.close(entry.scope, Exit.void) + }) + + const kill = Effect.fnUntraced(function*(runner: ClusterRunner) { + const entry = entryFor(runner) + if (entry.state() === "stopped" || entry.state() === "killed") return + entry.setState("killed") + yield* entry.controller.kill + yield* Scope.close(entry.scope, Exit.void) + }) + + const freeze = Effect.fnUntraced(function*(runner: ClusterRunner) { + const entry = entryFor(runner) + if (entry.state() !== "running") return + entry.setState("frozen") + yield* entry.controller.freeze + }) + + const assignmentMap = () => { + const assignments: Record> = {} + for (let id = 1; id <= ShardingConfig.defaults.shardsPerGroup; id++) { + const shard = ShardId.make("default", id) + assignments[shard.toString()] = runners + .filter((runner) => runner.state() === "running" && runner.sharding.hasShardId(shard)) + .map((runner) => `${runner.address.host}:${runner.address.port}`) + } + return assignments + } + + const messageCounts = Effect.fnUntraced(function*() { + const messages = sql(`${prefix}_messages`) + const replies = sql(`${prefix}_replies`) + const rows = yield* sql` + SELECT m.processed, r.payload AS reply_payload + FROM ${messages} m + LEFT JOIN ${replies} r ON r.id = m.last_reply_id + WHERE m.kind = 0 + ` + let failed = 0 + let replied = 0 + let unprocessed = 0 + for (const row of rows) { + if (!row.processed) { + unprocessed++ + continue + } + const reply = parseReply(row.reply_payload) + if (reply?._tag === "Success") replied++ + if (reply?._tag === "Failure") failed++ + } + return { failed, replied, unprocessed } satisfies MessageCounts + }) + + const diagnostics = Effect.fnUntraced(function*() { + const table = sql(`${prefix}_runners`) + const registrations = yield* sql` + SELECT address, runner, healthy, last_heartbeat + FROM ${table} + ORDER BY address + ` + return { + assignments: assignmentMap(), + messageCounts: yield* messageCounts(), + registrations + } + }) + + const waitUntil = Effect.fnUntraced(function*( + description: string, + condition: Effect.Effect, + timeout: Duration.Input = "15 seconds" + ) { + const started = yield* Clock.currentTimeMillis + const deadline = started + Duration.toMillis(Duration.fromInputUnsafe(timeout)) + if (yield* Effect.provide(condition, client)) return + while ((yield* Clock.currentTimeMillis) < deadline) { + yield* Effect.sleep(100) + if (yield* Effect.provide(condition, client)) return + } + const state = yield* diagnostics() + return yield* Effect.fail(new Error(`${description}\n${JSON.stringify(state, null, 2)}`)) + }) + + const waitForStableAssignments = Effect.fnUntraced(function*(timeout?: Duration.Input) { + let previous = "" + let stablePolls = 0 + yield* waitUntil( + "Shard assignments did not stabilize before the deadline", + Effect.sync(() => { + const current = assignmentMap() + const owners = new Set(Object.values(current).flat()) + const complete = Object.values(current).every((owners) => owners.length === 1) && + runners.every((runner) => + runner.state() !== "running" || owners.has(`${runner.address.host}:${runner.address.port}`) + ) + const encoded = complete ? JSON.stringify(current) : "" + stablePolls = encoded !== "" && encoded === previous ? stablePolls + 1 : 0 + previous = encoded + return stablePolls >= 3 + }), + timeout + ) + return assignmentMap() + }) + + const ownerOfEntity = Effect.fnUntraced(function*( + entity: Entity.Entity, + entityId: string + ) { + const shardId = yield* entity.getShardId(EntityId.make(entityId)).pipe(Effect.provide(client)) + return runners.find((runner) => runner.state() === "running" && runner.sharding.hasShardId(shardId)) as + | ClusterRunner + | undefined + }) + + const waitForEntityOwner = ( + entity: Entity.Entity, + entityId: string, + runner: ClusterRunner, + timeout?: Duration.Input + ) => + waitUntil( + `Entity ${entity.type}/${entityId} was not owned by runner ${runner.index} before the deadline`, + Effect.map(ownerOfEntity(entity, entityId), (owner) => owner === runner), + timeout + ) + + const getClient = (entity: Entity.Entity) => + entity.client.pipe(Effect.provide(client)) + + return { + assignmentMap, + backend: options.backend, + diagnostics, + failedMessageCount: Effect.map(messageCounts(), (counts) => counts.failed), + freeze, + getClient, + kill, + lockMode: options.lockMode ?? "advisory", + messageCounts, + ownerOfEntity, + prefix, + repliedMessageCount: Effect.map(messageCounts(), (counts) => counts.replied), + runners: runners as ReadonlyArray, + start, + stop, + unprocessedMessageCount: Effect.map(messageCounts(), (counts) => counts.unprocessed), + waitForEntityOwner, + waitForStableAssignments, + waitUntil + } as const +}) diff --git a/vitest.config.ts b/vitest.config.ts index 1be92940a1e..3104bd82e64 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -9,9 +9,16 @@ const isNode = typeof process !== "undefined" && !isDeno && !isBun const integrationTestsEnabled = process.env.EFFECT_INTEGRATION_TESTS === "1" -const clusterIntegrationTestsEnabled = process.env.EFFECT_CLUSTER_INTEGRATION_TESTS === "1" +const clusterTestsEnabled = process.env.EFFECT_CLUSTER_TESTS === "1" -const project = (name: string, directory: string, include: boolean = true, config: ViteUserConfig = {}) => { +const project = ( + name: string, + directory: string, + include: boolean = true, + config: ViteUserConfig = {}, + projectExclude?: ReadonlyArray, + projectInclude?: ReadonlyArray +) => { if (!include) { return [] } @@ -21,7 +28,14 @@ const project = (name: string, directory: string, include: boolean = true, confi test: { name } }, config) - return [mergeConfig(shared, cfg)] + const merged = mergeConfig(shared, cfg) + if (projectExclude !== undefined) { + merged.test!.exclude = [...projectExclude] + } + if (projectInclude !== undefined) { + merged.test!.include = [...projectInclude] + } + return [merged] } export const exclude = [ @@ -33,7 +47,7 @@ export const exclude = [ "**/typetest/**", "**/coverage/**", "**/test/utils/**", - ...(!clusterIntegrationTestsEnabled ? ["**/cluster-integration/**"] : []), + "**/test/cluster-integration/**", ...(!integrationTestsEnabled ? ["**/*.integration.test.{ts,tsx}"] : []), "**/*.d.ts", "**/*.config.*", @@ -119,6 +133,26 @@ export default defineConfig({ }), ...project("@effect/platform-deno", "packages/platform-deno", isDeno), ...project("@effect/platform-node", "packages/platform-node", isNode), + ...project( + "cluster-integration", + "packages/platform-node", + isNode && clusterTestsEnabled, + { + test: { + globalSetup: [path.join(__dirname, "packages/platform-node/test/cluster-integration/globalSetup.ts")], + include: ["test/cluster-integration/**/*.test.ts"], + retry: 0, + sequence: { + concurrent: false + }, + testTimeout: 60_000 + } + }, + exclude.filter((path) => path !== "**/test/cluster-integration/**"), + [ + "test/cluster-integration/**/*.test.ts" + ] + ), ...project("@effect/platform-node-shared", "packages/platform-node-shared", !isDeno), ...project("@effect/vitest", "packages/vitest"), ...project("@effect/sql-clickhouse", "packages/sql/clickhouse"), From 68e2521a17139ffd59d2a0558b443f0bc5f053bf Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 11:58:12 +1200 Subject: [PATCH 3/9] Add cluster entity integration tests --- .../test/cluster-integration/Entity.test.ts | 428 ++++++++++++++++++ .../test/cluster-integration/README.md | 5 +- .../test/cluster-integration/Smoke.test.ts | 1 - .../test/cluster-integration/harness.ts | 76 +++- 4 files changed, 489 insertions(+), 21 deletions(-) create mode 100644 packages/platform-node/test/cluster-integration/Entity.test.ts diff --git a/packages/platform-node/test/cluster-integration/Entity.test.ts b/packages/platform-node/test/cluster-integration/Entity.test.ts new file mode 100644 index 00000000000..54832000789 --- /dev/null +++ b/packages/platform-node/test/cluster-integration/Entity.test.ts @@ -0,0 +1,428 @@ +import { assert, describe, it } from "@effect/vitest" +import { Clock, Effect, Fiber, Latch, Layer, PrimaryKey, Schema, Scope } from "effect" +import { ClusterSchema, Entity, EntityResource, Singleton } from "effect/unstable/cluster" +import { Rpc } from "effect/unstable/rpc" +import { type Backend, type ClusterRunner, make } from "./harness.ts" + +class Request extends Schema.Class("ClusterEntityRequest")({ + id: Schema.String, + sequence: Schema.Number +}) { + [PrimaryKey.symbol]() { + return this.id + } +} + +const StateReply = Schema.Struct({ + generation: Schema.Number, + runner: Schema.String, + value: Schema.Number +}) + +const StateEntity = Entity.make("ClusterIntegrationState", [ + Rpc.make("Increment", { + payload: Request, + success: StateReply + }), + Rpc.make("Ordered", { + payload: Request, + success: Schema.Number + }) +]).annotateRpcs(ClusterSchema.Persisted, true) + +let orderGate = Latch.makeUnsafe(true) +let orderEntered = Latch.makeUnsafe() +let order: Array = [] +const generations = new Map() + +const addressString = (address: { readonly host: string; readonly port: number }) => `${address.host}:${address.port}` + +const StateEntityLayer = StateEntity.toLayer( + Effect.gen(function*() { + const address = yield* Entity.CurrentAddress + const runner = yield* Entity.CurrentRunnerAddress + const entityId = String(address.entityId) + const generation = (generations.get(entityId) ?? 0) + 1 + generations.set(entityId, generation) + let value = 0 + return { + Increment: () => + Effect.sync(() => ({ + generation, + runner: addressString(runner), + value: ++value + })), + Ordered: ({ payload }) => + Effect.gen(function*() { + order.push(payload.sequence) + if (payload.sequence === 1) { + orderEntered.openUnsafe() + yield* orderGate.await + } + return payload.sequence + }) + } + }), + { maxIdleTime: "1 second" } +) + +const MailboxEntity = Entity.make("ClusterIntegrationMailbox", [ + Rpc.make("Hold", { + payload: Request, + success: Schema.Number + }) +]) + +let mailboxGate = Latch.makeUnsafe(true) +let mailboxEntered = Latch.makeUnsafe() + +const MailboxEntityLayer = MailboxEntity.toLayer({ + Hold: ({ payload }) => + Effect.gen(function*() { + mailboxEntered.openUnsafe() + yield* mailboxGate.await + return payload.sequence + }) +}, { mailboxCapacity: 1 }) + +const GroupEntity = Entity.make("ClusterIntegrationSpecialGroup", [ + Rpc.make("Runner", { success: Schema.String }) +]).annotate(ClusterSchema.ShardGroup, () => "special") + +const GroupEntityLayer = GroupEntity.toLayer(Effect.gen(function*() { + const runner = yield* Entity.CurrentRunnerAddress + return { Runner: () => Effect.succeed(addressString(runner)) } +})) + +const ResourceEntity = Entity.make("ClusterIntegrationResource", [ + Rpc.make("Get", { success: Schema.Number }), + Rpc.make("Close", { success: Schema.Void }) +]) + +const resourceState = { acquired: 0, released: 0 } + +const ResourceEntityLayer = ResourceEntity.toLayer(Effect.gen(function*() { + const resource = yield* EntityResource.make({ + acquire: Effect.gen(function*() { + const closeScope = yield* EntityResource.CloseScope + return yield* Effect.acquireRelease( + Effect.sync(() => ++resourceState.acquired), + () => Effect.sync(() => resourceState.released++) + ).pipe(Scope.provide(closeScope)) + }) + }) + return { + Close: () => resource.close, + Get: () => Effect.scoped(resource.get) + } +})) + +const StandardEntities = Layer.mergeAll(StateEntityLayer, MailboxEntityLayer, ResourceEntityLayer) + +const resetOrder = () => { + orderGate = Latch.makeUnsafe() + orderEntered = Latch.makeUnsafe() + order = [] +} + +const resetMailbox = () => { + mailboxGate = Latch.makeUnsafe() + mailboxEntered = Latch.makeUnsafe() +} + +const findIdsByRunner = Effect.fnUntraced(function*( + cluster: Effect.Success>, + runners: ReadonlyArray +) { + const found = new Map() + for (let index = 0; index < 2_000 && found.size < runners.length; index++) { + const id = `entity-${index}` + const owner = yield* cluster.ownerOfEntity(StateEntity, id) + if (owner && runners.includes(owner) && !found.has(owner)) found.set(owner, id) + } + assert.strictEqual(found.size, runners.length) + return found +}) + +describe("cluster entity integration", () => { + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: routes by entity id, isolates state, and preserves mailbox order`, () => + Effect.gen(function*() { + generations.clear() + resetOrder() + const cluster = yield* make({ backend, entities: StandardEntities }) + const runners = yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const ids = yield* findIdsByRunner(cluster, runners) + const client = yield* cluster.getClient(StateEntity) + + for (const [runner, id] of ids) { + const first = yield* client(id).Increment(new Request({ id: `${id}-1`, sequence: 0 })) + const second = yield* client(id).Increment(new Request({ id: `${id}-2`, sequence: 0 })) + assert.strictEqual(first.runner, addressString(runner.address)) + assert.strictEqual(second.runner, first.runner) + assert.strictEqual(first.value, 1) + assert.strictEqual(second.value, 2) + assert.isFalse(cluster.clientSharding.hasShardId(yield* cluster.shardOfEntity(StateEntity, id))) + } + + const orderedId = ids.values().next().value! + const first = yield* client(orderedId).Ordered( + new Request({ id: `${backend}-ordered-1`, sequence: 1 }) + ).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil("The first ordered request did not start", Effect.as(orderEntered.await, true)) + const second = yield* client(orderedId).Ordered( + new Request({ id: `${backend}-ordered-2`, sequence: 2 }) + ).pipe(Effect.forkChild({ startImmediately: true })) + orderGate.openUnsafe() + assert.strictEqual(yield* Fiber.join(first), 1) + assert.strictEqual(yield* Fiber.join(second), 2) + assert.deepStrictEqual(order, [1, 2]) + + const registrations = (yield* cluster.diagnostics()).registrations + assert.strictEqual(registrations.length, runners.length) + }).pipe(Effect.scoped)) + + it.live(`${backend}: reports mailbox saturation and revives idle entities with fresh state`, () => + Effect.gen(function*() { + generations.clear() + resetMailbox() + const cluster = yield* make({ backend, entities: StandardEntities }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + + const mailbox = yield* cluster.getClient(MailboxEntity) + const held = yield* mailbox("full").Hold( + new Request({ id: `${backend}-held`, sequence: 1 }) + ).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil("The mailbox request did not start", Effect.as(mailboxEntered.await, true)) + const error = yield* mailbox("full").Hold( + new Request({ id: `${backend}-rejected`, sequence: 2 }) + ).pipe(Effect.flip) + assert.strictEqual(error._tag, "MailboxFull") + mailboxGate.openUnsafe() + assert.strictEqual(yield* Fiber.join(held), 1) + + const state = yield* cluster.getClient(StateEntity) + const first = yield* state("idle").Increment( + new Request({ id: `${backend}-idle-1`, sequence: 0 }) + ) + const owner = yield* cluster.ownerOfEntity(StateEntity, "idle") + yield* cluster.waitUntil( + "The idle entity was not reaped", + Effect.map(owner!.sharding.activeEntityCount, (count) => count === 0), + "12 seconds" + ) + const revived = yield* state("idle").Increment( + new Request({ id: `${backend}-idle-2`, sequence: 0 }) + ) + assert.strictEqual(first.generation, 1) + assert.strictEqual(revived.generation, 2) + assert.strictEqual(revived.value, 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: rebalances on runner addition, graceful stop, and abrupt death`, () => + Effect.gen(function*() { + resetOrder() + const cluster = yield* make({ backend, entities: StandardEntities }) + const initial = yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const before = new Map() + for (let index = 0; index < 2_000; index++) { + const id = `moving-${index}` + before.set(id, (yield* cluster.ownerOfEntity(StateEntity, id))!) + } + + const [added] = yield* cluster.start(1) + assert.strictEqual(added.index, 2) + yield* cluster.waitForStableAssignments() + let movedId: string | undefined + for (const [id, old] of before) { + if (old !== added && (yield* cluster.ownerOfEntity(StateEntity, id)) === added) { + movedId = id + break + } + } + assert.isDefined(movedId) + const client = yield* cluster.getClient(StateEntity) + const moved = yield* client(movedId!).Increment( + new Request({ id: `${backend}-moved`, sequence: 0 }) + ) + assert.strictEqual(moved.runner, addressString(added.address)) + + const stopId = (yield* findIdsByRunner(cluster, initial)).get(initial[0])! + const request = yield* client(stopId).Ordered( + new Request({ id: `${backend}-stop`, sequence: 1 }) + ).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil("The handover request did not start", Effect.as(orderEntered.await, true)) + const stopping = yield* cluster.stop(initial[0]).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil( + "The stopped runner did not hand over its entity", + Effect.map(cluster.ownerOfEntity(StateEntity, stopId), (owner) => owner !== undefined && owner !== initial[0]) + ) + orderGate.openUnsafe() + assert.strictEqual(yield* Fiber.join(request), 1) + yield* Fiber.join(stopping) + + const killId = `kill-${backend}` + const killed = yield* cluster.ownerOfEntity(StateEntity, killId) + yield* cluster.kill(killed!) + const reply = yield* client(killId).Increment( + new Request({ id: `${backend}-kill`, sequence: 0 }) + ) + assert.notStrictEqual(reply.runner, addressString(killed!.address)) + yield* cluster.waitForStableAssignments() + assert.strictEqual((yield* cluster.messageCounts()).unprocessed, 0) + }).pipe(Effect.scoped)) + + it.live(`${backend}: transfers frozen row locks after expiry`, () => + Effect.gen(function*() { + const cluster = yield* make({ backend, entities: StandardEntities, lockMode: "row" }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const id = `frozen-${backend}` + const oldOwner = yield* cluster.ownerOfEntity(StateEntity, id) + const shard = yield* cluster.shardOfEntity(StateEntity, id) + yield* cluster.freeze(oldOwner!) + yield* cluster.waitUntil( + "The frozen runner's row lock did not expire", + Effect.map(cluster.ownerOfEntity(StateEntity, id), (owner) => owner !== undefined && owner !== oldOwner), + "12 seconds" + ) + const nextOwner = yield* cluster.ownerOfEntity(StateEntity, id) + assert.strictEqual(cluster.ownersOfShard(shard).length, 1) + assert.strictEqual(cluster.ownersOfShard(shard)[0], nextOwner) + const client = yield* cluster.getClient(StateEntity) + const reply = yield* client(id).Increment(new Request({ id: `${backend}-freeze`, sequence: 0 })) + assert.strictEqual(reply.runner, addressString(nextOwner!.address)) + yield* cluster.kill(oldOwner!) + }).pipe(Effect.scoped)) + + it.live(`${backend}: retains frozen advisory locks until the session closes`, () => + Effect.gen(function*() { + const cluster = yield* make({ backend, entities: StandardEntities }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const id = `frozen-advisory-${backend}` + const oldOwner = yield* cluster.ownerOfEntity(StateEntity, id) + const shard = yield* cluster.shardOfEntity(StateEntity, id) + yield* cluster.freeze(oldOwner!) + const deadline = (yield* Clock.currentTimeMillis) + 3_000 + yield* cluster.waitUntil( + "The advisory-lock observation window did not elapse", + Effect.map(Clock.currentTimeMillis, (now) => now >= deadline), + "5 seconds" + ) + assert.strictEqual(cluster.ownersOfShard(shard).length, 0) + assert.deepStrictEqual(cluster.ownersOfShard(shard, true), [oldOwner]) + yield* cluster.kill(oldOwner!) + yield* cluster.waitUntil( + "The advisory lock was not handed over after its session closed", + Effect.map(cluster.ownerOfEntity(StateEntity, id), (owner) => owner !== undefined && owner !== oldOwner) + ) + }).pipe(Effect.scoped)) + } + + it.live("assigns annotated entities only to runners in their shard group", () => + Effect.gen(function*() { + const entities = Layer.mergeAll(StateEntityLayer, GroupEntityLayer) + const cluster = yield* make({ + backend: "pg", + config: { availableShardGroups: ["default", "special"], shardsPerGroup: 30 }, + entities + }) + const [defaultRunner] = yield* cluster.start(1, { assignedShardGroups: ["default"] }) + const [specialRunner] = yield* cluster.start(1, { assignedShardGroups: ["special"] }) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(GroupEntity) + assert.strictEqual(yield* client("grouped").Runner(), addressString(specialRunner.address)) + assert.strictEqual(yield* cluster.ownerOfEntity(GroupEntity, "grouped"), specialRunner) + assert.strictEqual(yield* cluster.ownerOfEntity(StateEntity, "default"), defaultRunner) + }).pipe(Effect.scoped)) + + it.live("runs one singleton and migrates it after owner death", () => + Effect.gen(function*() { + const singleton = { active: 0, maxActive: 0, starts: 0 } + const singletonLayer = Singleton.make( + "cluster-integration-singleton", + Effect.acquireRelease( + Effect.sync(() => { + singleton.active++ + singleton.starts++ + singleton.maxActive = Math.max(singleton.maxActive, singleton.active) + }), + () => Effect.sync(() => singleton.active--) + ) + ) + const cluster = yield* make({ + backend: "pg", + entities: Layer.merge(StateEntityLayer, singletonLayer) + }) + const runners = yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil("The singleton did not start", Effect.sync(() => singleton.active === 1)) + const firstOwner = yield* cluster.ownerOfEntity(StateEntity, "cluster-integration-singleton") + assert.isTrue(runners.includes(firstOwner!)) + yield* cluster.kill(firstOwner!) + yield* cluster.waitUntil( + "The singleton did not migrate", + Effect.sync(() => singleton.starts >= 2 && singleton.active === 1) + ) + assert.notStrictEqual( + yield* cluster.ownerOfEntity(StateEntity, "cluster-integration-singleton"), + firstOwner + ) + assert.strictEqual(singleton.maxActive, 1) + }).pipe(Effect.scoped)) + + it.live("keeps EntityResource alive during movement and releases it explicitly", () => + Effect.gen(function*() { + resourceState.acquired = 0 + resourceState.released = 0 + const cluster = yield* make({ backend: "pg", entities: ResourceEntityLayer }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(ResourceEntity) + assert.strictEqual(yield* client("resource").Get(), 1) + const owner = yield* cluster.ownerOfEntity(ResourceEntity, "resource") + yield* cluster.stop(owner!) + yield* cluster.waitUntil( + "The resource entity did not move", + Effect.map(cluster.ownerOfEntity(ResourceEntity, "resource"), (next) => next !== undefined && next !== owner) + ) + assert.strictEqual(resourceState.released, 0) + assert.strictEqual(yield* client("resource").Get(), 2) + yield* client("resource").Close() + yield* cluster.waitUntil( + "The entity resource was not released", + Effect.sync(() => resourceState.released === 1) + ) + }).pipe(Effect.scoped)) + + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: isolates clusters with different table prefixes`, () => + Effect.gen(function*() { + const first = yield* make({ backend, entities: StateEntityLayer, config: { shardsPerGroup: 30 } }) + const second = yield* make({ backend, entities: StateEntityLayer, config: { shardsPerGroup: 30 } }) + const [firstRunner] = yield* first.start(1) + const [secondRunner] = yield* second.start(1) + yield* first.waitForStableAssignments() + yield* second.waitForStableAssignments() + assert.notStrictEqual(first.prefix, second.prefix) + const firstRegistrations = (yield* first.diagnostics()).registrations + const secondRegistrations = (yield* second.diagnostics()).registrations + assert.deepStrictEqual(firstRegistrations.map((row) => row.address), [addressString(firstRunner.address)]) + assert.deepStrictEqual(secondRegistrations.map((row) => row.address), [addressString(secondRunner.address)]) + const firstClient = yield* first.getClient(StateEntity) + const secondClient = yield* second.getClient(StateEntity) + const firstReply = yield* firstClient("same-id").Increment( + new Request({ id: `${backend}-prefix-first`, sequence: 0 }) + ) + const secondReply = yield* secondClient("same-id").Increment( + new Request({ id: `${backend}-prefix-second`, sequence: 0 }) + ) + assert.strictEqual(firstReply.runner, addressString(firstRunner.address)) + assert.strictEqual(secondReply.runner, addressString(secondRunner.address)) + }).pipe(Effect.scoped)) + } +}) diff --git a/packages/platform-node/test/cluster-integration/README.md b/packages/platform-node/test/cluster-integration/README.md index b49b389d050..590f6da6330 100644 --- a/packages/platform-node/test/cluster-integration/README.md +++ b/packages/platform-node/test/cluster-integration/README.md @@ -16,12 +16,13 @@ prefix, and every runner listens on an operating-system-assigned port. `harness.ts` exposes: -- `make({ backend, entities, lockMode })` to create a scoped cluster harness. -- `start(count)` to start in-process socket runners and a client over msgpack. +- `make({ backend, entities, lockMode, config })` to create a scoped cluster harness. +- `start(count, { assignedShardGroups, runnerShardWeight })` to start in-process socket runners and a client over msgpack. - `stop(runner)` for graceful deregistration and shard handoff. - `kill(runner)` for abrupt teardown without deregistration or explicit lock cleanup. - `freeze(runner)` to suspend SQL heartbeats and lock refresh while leaving the runner's sockets and reserved SQL connection open. - `waitUntil`, `waitForStableAssignments`, and `waitForEntityOwner` for deadline-based polling with cluster diagnostics on failure. +- `clientSharding` and `ownersOfShard` for direct shard ownership assertions. - `messageCounts`, `unprocessedMessageCount`, `repliedMessageCount`, and `failedMessageCount` for storage assertions scoped to the cluster prefix. Advisory locks are owned by the reserved database session. A frozen advisory-lock diff --git a/packages/platform-node/test/cluster-integration/Smoke.test.ts b/packages/platform-node/test/cluster-integration/Smoke.test.ts index e4dd18c534e..d1813c16b9c 100644 --- a/packages/platform-node/test/cluster-integration/Smoke.test.ts +++ b/packages/platform-node/test/cluster-integration/Smoke.test.ts @@ -42,7 +42,6 @@ describe("cluster integration smoke", () => { Effect.map(cluster.ownerOfEntity(TestEntity, "entity-1"), (owner) => owner !== undefined) ) const owner = yield* cluster.ownerOfEntity(TestEntity, "entity-1") - yield* cluster.freeze(owner!) yield* cluster.kill(owner!) yield* cluster.waitForStableAssignments() diff --git a/packages/platform-node/test/cluster-integration/harness.ts b/packages/platform-node/test/cluster-integration/harness.ts index 1e83abe80be..804b0b9a7cb 100644 --- a/packages/platform-node/test/cluster-integration/harness.ts +++ b/packages/platform-node/test/cluster-integration/harness.ts @@ -29,6 +29,7 @@ export type LockMode = "advisory" | "row" export interface ClusterRunner { readonly address: RunnerAddress.RunnerAddress readonly index: number + readonly shardGroups: ReadonlyArray readonly sharding: Sharding.Sharding["Service"] readonly state: () => "frozen" | "killed" | "running" | "stopped" } @@ -41,11 +42,21 @@ export interface MessageCounts { export interface MakeOptions { readonly backend: Backend + readonly config?: Partial | undefined readonly entities: Layer.Layer readonly lockMode?: LockMode | undefined readonly runnerLayer?: RunnerLayer | undefined } +export interface StartOptions { + readonly assignedShardGroups?: ReadonlyArray | undefined + readonly runnerShardWeight?: number | undefined +} + +type HarnessConfig = Partial & { + readonly shardLockDisableAdvisory: boolean +} + interface RegistrationRow { readonly address: string readonly healthy: boolean | number @@ -138,7 +149,7 @@ export const socketRunnerLayer = ( address: RunnerAddress.RunnerAddress, entities: Layer.Layer, socketServer: SocketServer.SocketServer["Service"], - config: typeof clusterConfig & { readonly shardLockDisableAdvisory: boolean } + config: HarnessConfig ) => entities.pipe( Layer.provideMerge(SocketRunner.layer), @@ -154,9 +165,7 @@ export const socketRunnerLayer = ( export type RunnerLayer = typeof socketRunnerLayer -const clientLayer = ( - config: typeof clusterConfig & { readonly shardLockDisableAdvisory: boolean } -) => +const clientLayer = (config: HarnessConfig) => SocketRunner.layerClientOnly.pipe( Layer.provide(NodeClusterSocket.layerClientProtocol), Layer.provide(ShardingConfig.layer(config)), @@ -173,6 +182,7 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { const prefix = `cluster_${process.pid}_${nextCluster++}` const config = { ...clusterConfig, + ...options.config, shardLockDisableAdvisory: options.lockMode === "row" } const databases = inject("clusterDatabases") @@ -190,9 +200,12 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { const shared = Context.merge(database, messageStorage) const runners: Array = [] - const makeRunnerStorage = Effect.fnUntraced(function*(scope: Scope.Closeable) { + const makeRunnerStorage = Effect.fnUntraced(function*( + scope: Scope.Closeable, + storageConfig: HarnessConfig = config + ) { return yield* SqlRunnerStorage.layerWith({ prefix }).pipe( - Layer.provide(ShardingConfig.layer(config)), + Layer.provide(ShardingConfig.layer(storageConfig)), Layer.orDie, Layer.buildWithScope(scope), Effect.provide(database) @@ -205,8 +218,19 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { Layer.buildWithScope(clientScope), Effect.provide(Context.merge(shared, clientStorage)) ) - const startRunner = Effect.fnUntraced(function*(index: number) { + const clientSharding = Context.get(client, Sharding.Sharding) + let nextRunnerIndex = 0 + const startRunner = Effect.fnUntraced(function*(index: number, startOptions?: StartOptions) { const scope = yield* Scope.fork(parentScope) + const runnerConfig: HarnessConfig = { + ...config, + ...(startOptions?.assignedShardGroups === undefined + ? undefined + : { assignedShardGroups: startOptions.assignedShardGroups }), + ...(startOptions?.runnerShardWeight === undefined + ? undefined + : { runnerShardWeight: startOptions.runnerShardWeight }) + } const serverContext = yield* NodeSocketServer.layer({ host: "127.0.0.1", port: 0 }).pipe( Layer.buildWithScope(scope) ) @@ -215,14 +239,14 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { return yield* Effect.die("Expected a TCP socket server") } const address = RunnerAddress.make("127.0.0.1", socketServer.address.port) - const rawStorageContext = yield* makeRunnerStorage(scope) + const rawStorageContext = yield* makeRunnerStorage(scope, runnerConfig) const controller = makeRunnerStorageController(Context.get(rawStorageContext, RunnerStorage.RunnerStorage)) const storage = Context.make(RunnerStorage.RunnerStorage, controller.controlled) const context = yield* (options.runnerLayer ?? socketRunnerLayer)( address, options.entities, socketServer, - config + runnerConfig ).pipe( Layer.buildWithScope(scope), Effect.provide(Context.mergeAll(shared, storage)) @@ -233,6 +257,7 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { address, controller, index, + shardGroups: runnerConfig.assignedShardGroups ?? ShardingConfig.defaults.assignedShardGroups, scope, setState(next) { state = next @@ -244,10 +269,10 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { return runner as ClusterRunner }) - const start = Effect.fnUntraced(function*(runnerCount: number) { + const start = Effect.fnUntraced(function*(runnerCount: number, startOptions?: StartOptions) { return yield* Effect.forEach( - Array.from({ length: runnerCount }, (_, index) => index), - startRunner + Array.from({ length: runnerCount }, () => nextRunnerIndex++), + (index) => startRunner(index, startOptions) ) }) @@ -278,15 +303,22 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { const assignmentMap = () => { const assignments: Record> = {} - for (let id = 1; id <= ShardingConfig.defaults.shardsPerGroup; id++) { - const shard = ShardId.make("default", id) - assignments[shard.toString()] = runners - .filter((runner) => runner.state() === "running" && runner.sharding.hasShardId(shard)) - .map((runner) => `${runner.address.host}:${runner.address.port}`) + const groups = config.availableShardGroups ?? ShardingConfig.defaults.availableShardGroups + const shardsPerGroup = config.shardsPerGroup ?? ShardingConfig.defaults.shardsPerGroup + for (const group of groups) { + for (let id = 1; id <= shardsPerGroup; id++) { + const shard = ShardId.make(group, id) + assignments[shard.toString()] = runners + .filter((runner) => runner.state() === "running" && runner.sharding.hasShardId(shard)) + .map((runner) => `${runner.address.host}:${runner.address.port}`) + } } return assignments } + const ownersOfShard = (shard: ShardId.ShardId, includeInactive = false) => + runners.filter((runner) => (includeInactive || runner.state() === "running") && runner.sharding.hasShardId(shard)) + const messageCounts = Effect.fnUntraced(function*() { const messages = sql(`${prefix}_messages`) const replies = sql(`${prefix}_replies`) @@ -363,11 +395,16 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { return assignmentMap() }) + const shardOfEntity = ( + entity: Entity.Entity, + entityId: string + ) => entity.getShardId(EntityId.make(entityId)).pipe(Effect.provide(client)) + const ownerOfEntity = Effect.fnUntraced(function*( entity: Entity.Entity, entityId: string ) { - const shardId = yield* entity.getShardId(EntityId.make(entityId)).pipe(Effect.provide(client)) + const shardId = yield* shardOfEntity(entity, entityId) return runners.find((runner) => runner.state() === "running" && runner.sharding.hasShardId(shardId)) as | ClusterRunner | undefined @@ -391,6 +428,7 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { return { assignmentMap, backend: options.backend, + clientSharding, diagnostics, failedMessageCount: Effect.map(messageCounts(), (counts) => counts.failed), freeze, @@ -399,9 +437,11 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { lockMode: options.lockMode ?? "advisory", messageCounts, ownerOfEntity, + ownersOfShard, prefix, repliedMessageCount: Effect.map(messageCounts(), (counts) => counts.replied), runners: runners as ReadonlyArray, + shardOfEntity, start, stop, unprocessedMessageCount: Effect.map(messageCounts(), (counts) => counts.unprocessed), From 57698085b0848ad49aa26206bda59143738a27dc Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 12:17:33 +1200 Subject: [PATCH 4/9] Add cluster persistence integration tests --- .../cluster-integration/Persistence.test.ts | 407 ++++++++++++++++++ 1 file changed, 407 insertions(+) create mode 100644 packages/platform-node/test/cluster-integration/Persistence.test.ts diff --git a/packages/platform-node/test/cluster-integration/Persistence.test.ts b/packages/platform-node/test/cluster-integration/Persistence.test.ts new file mode 100644 index 00000000000..bb1ae8c2823 --- /dev/null +++ b/packages/platform-node/test/cluster-integration/Persistence.test.ts @@ -0,0 +1,407 @@ +import { assert, describe, it } from "@effect/vitest" +import { Cause, Clock, DateTime, Effect, Exit, Fiber, Latch, PrimaryKey, Schema, Stream } from "effect" +import { ClusterSchema, DeliverAt, Entity } from "effect/unstable/cluster" +import { Rpc, RpcSchema } from "effect/unstable/rpc" +import { type Backend, make } from "./harness.ts" + +class KeyedPayload extends Schema.Class("ClusterPersistenceKeyedPayload")({ + id: Schema.String +}) { + [PrimaryKey.symbol]() { + return this.id + } +} + +class ScheduledPayload extends Schema.Class("ClusterPersistenceScheduledPayload")({ + deliverAt: Schema.Number, + id: Schema.String +}) { + [PrimaryKey.symbol]() { + return this.id + } + + [DeliverAt.symbol]() { + return DateTime.makeUnsafe(this.deliverAt) + } +} + +const PersistedRpc = Rpc.make("Persisted", { + payload: KeyedPayload, + success: Schema.String +}) + +const StoredReplyRpc = Rpc.make("StoredReply", { + payload: KeyedPayload, + success: Schema.String +}) + +const UninterruptibleRpc = Rpc.make("Uninterruptible", { + payload: KeyedPayload, + success: Schema.String +}).annotate(ClusterSchema.Uninterruptible, true) + +const VolatileRpc = Rpc.make("Volatile", { + payload: KeyedPayload, + success: Schema.String +}).annotate(ClusterSchema.Persisted, false) + +const TypedFailureRpc = Rpc.make("TypedFailure", { + error: Schema.String, + payload: KeyedPayload, + success: Schema.Never +}) + +const DefectRpc = Rpc.make("Defect", { + payload: KeyedPayload, + success: Schema.Never +}) + +const HealthyRpc = Rpc.make("Healthy", { + payload: KeyedPayload, + success: Schema.String +}) + +const StreamedRpc = Rpc.make("Streamed", { + payload: KeyedPayload, + success: RpcSchema.Stream(Schema.Number, Schema.Never) +}) + +const ScheduledRpc = Rpc.make("Scheduled", { + payload: ScheduledPayload, + success: Schema.Number +}) + +const PersistenceEntity = Entity.make("ClusterIntegrationPersistence", [ + PersistedRpc, + StoredReplyRpc, + UninterruptibleRpc, + VolatileRpc, + TypedFailureRpc, + DefectRpc, + HealthyRpc, + StreamedRpc, + ScheduledRpc +]).annotateRpcs(ClusterSchema.Persisted, true) + +const freshState = () => ({ + completedUninterruptible: 0, + completedVolatile: 0, + counts: new Map(), + scheduledDeliveries: [] as Array, + uninterruptibleEntered: Latch.makeUnsafe(), + uninterruptibleGate: Latch.makeUnsafe(), + volatileEntered: Latch.makeUnsafe(), + volatileGate: Latch.makeUnsafe() +}) + +let state = freshState() + +const resetState = () => { + state = freshState() +} + +const increment = (tag: string, id: string) => { + const key = `${tag}:${id}` + const next = (state.counts.get(key) ?? 0) + 1 + state.counts.set(key, next) + return next +} + +const count = (tag: string, id: string) => state.counts.get(`${tag}:${id}`) ?? 0 + +const PersistenceEntityLayer = PersistenceEntity.toLayer({ + Defect: ({ payload }) => + Effect.sync(() => increment("Defect", payload.id)).pipe( + Effect.andThen(Effect.die(`defect:${payload.id}`)) + ), + Healthy: ({ payload }) => + Effect.sync(() => { + increment("Healthy", payload.id) + return `healthy:${payload.id}` + }), + Persisted: ({ payload }) => + Effect.sync(() => { + increment("Persisted", payload.id) + return `persisted:${payload.id}` + }), + Scheduled: ({ payload }) => + Effect.gen(function*() { + increment("Scheduled", payload.id) + const deliveredAt = yield* Clock.currentTimeMillis + state.scheduledDeliveries.push(deliveredAt) + return deliveredAt + }), + StoredReply: ({ payload }) => + Effect.sync(() => { + increment("StoredReply", payload.id) + return `stored:${payload.id}` + }), + Streamed: (request) => { + increment("Streamed", request.payload.id) + return Stream.fromIterable([0, 1, 2, 3, 4]).pipe(Stream.rechunk(1)) + }, + TypedFailure: ({ payload }) => + Effect.sync(() => increment("TypedFailure", payload.id)).pipe( + Effect.andThen(Effect.fail(`typed:${payload.id}`)) + ), + Uninterruptible: ({ payload }) => + Effect.gen(function*() { + increment("Uninterruptible", payload.id) + state.uninterruptibleEntered.openUnsafe() + yield* state.uninterruptibleGate.await + state.completedUninterruptible++ + return `uninterruptible:${payload.id}` + }), + Volatile: ({ payload }) => + Effect.gen(function*() { + increment("Volatile", payload.id) + state.volatileEntered.openUnsafe() + yield* state.volatileGate.await + state.completedVolatile++ + return `volatile:${payload.id}` + }) +}, { disableFatalDefects: true }) + +describe("cluster message persistence integration", () => { + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: delivers a persisted request sent while its runner is down exactly once`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ backend, entities: PersistenceEntityLayer }) + yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + const owner = yield* cluster.ownerOfEntity(PersistenceEntity, "restart") + yield* cluster.kill(owner!) + + const client = yield* cluster.getClient(PersistenceEntity) + const replyFiber = yield* client("restart").Persisted( + new KeyedPayload({ id: `${backend}-restart` }) + ).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil( + "The request was not persisted while the runner was down", + Effect.map(cluster.unprocessedMessageCount, (value) => value === 1) + ) + + yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + assert.strictEqual(yield* Fiber.join(replyFiber), `persisted:${backend}-restart`) + yield* cluster.waitUntil( + "The persisted reply was not recorded", + Effect.map(cluster.repliedMessageCount, (value) => value === 1) + ) + assert.strictEqual(count("Persisted", `${backend}-restart`), 1) + assert.deepStrictEqual(yield* cluster.messageCounts(), { + failed: 0, + replied: 1, + unprocessed: 0 + }) + }).pipe(Effect.scoped)) + + it.live(`${backend}: serves primary-key duplicates from the stored reply`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ backend, entities: PersistenceEntityLayer }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const id = `${backend}-stored` + const payload = new KeyedPayload({ id }) + + assert.strictEqual(yield* client("stored").StoredReply(payload), `stored:${id}`) + const firstOwner = yield* cluster.ownerOfEntity(PersistenceEntity, "stored") + yield* cluster.kill(firstOwner!) + yield* cluster.waitForStableAssignments() + + assert.strictEqual(yield* client("stored").StoredReply(payload), `stored:${id}`) + assert.strictEqual(count("StoredReply", id), 1) + assert.deepStrictEqual(yield* cluster.messageCounts(), { + failed: 0, + replied: 1, + unprocessed: 0 + }) + }).pipe(Effect.scoped)) + + it.live(`${backend}: does not lose an uninterruptible request during runner shutdown`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ + backend, + config: { entityTerminationTimeout: 100 }, + entities: PersistenceEntityLayer + }) + const [owner] = yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const id = `${backend}-uninterruptible` + const replyFiber = yield* client("uninterruptible").Uninterruptible( + new KeyedPayload({ id }) + ).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil( + "The uninterruptible handler did not start", + Effect.as(state.uninterruptibleEntered.await, true) + ) + + const stopping = yield* cluster.stop(owner).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The uninterruptible request was not resumed by the replacement runner", + Effect.sync(() => count("Uninterruptible", id) >= 2) + ) + state.uninterruptibleGate.openUnsafe() + yield* cluster.waitUntil( + "The resumed uninterruptible request did not complete", + Effect.sync(() => state.completedUninterruptible === 1) + ) + + assert.strictEqual(yield* Fiber.join(replyFiber), `uninterruptible:${id}`) + yield* Fiber.join(stopping) + assert.strictEqual(yield* cluster.repliedMessageCount, 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: does not store or redeliver a volatile request after runner failure`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ + backend, + config: { entityTerminationTimeout: 100 }, + entities: PersistenceEntityLayer + }) + const [owner] = yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const id = `${backend}-volatile` + const requestFiber = yield* client("volatile").Volatile( + new KeyedPayload({ id }), + { discard: true } + ).pipe(Effect.forkChild({ startImmediately: true })) + yield* cluster.waitUntil( + "The volatile handler did not start", + Effect.as(state.volatileEntered.await, true) + ) + + yield* cluster.kill(owner) + yield* Fiber.interrupt(requestFiber) + state.volatileGate.openUnsafe() + yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + const observationDeadline = (yield* Clock.currentTimeMillis) + 1_000 + yield* cluster.waitUntil( + "The volatile redelivery observation window did not elapse", + Effect.map(Clock.currentTimeMillis, (now) => now >= observationDeadline), + "2 seconds" + ) + + assert.strictEqual(count("Volatile", id), 1) + assert.strictEqual(state.completedVolatile, 0) + assert.deepStrictEqual(yield* cluster.messageCounts(), { + failed: 0, + replied: 0, + unprocessed: 0 + }) + }).pipe(Effect.scoped)) + + it.live(`${backend}: persists typed failures and defects without wedging the mailbox`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ backend, entities: PersistenceEntityLayer }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const typedId = `${backend}-typed` + const defectId = `${backend}-defect` + + assert.strictEqual( + yield* client("failures").TypedFailure(new KeyedPayload({ id: typedId })).pipe(Effect.flip), + `typed:${typedId}` + ) + assert.strictEqual( + yield* client("failures").TypedFailure(new KeyedPayload({ id: typedId })).pipe(Effect.flip), + `typed:${typedId}` + ) + assert.strictEqual(count("TypedFailure", typedId), 1) + + const firstDefect = yield* client("failures").Defect( + new KeyedPayload({ id: defectId }) + ).pipe(Effect.exit) + assert.isTrue(Exit.isFailure(firstDefect)) + if (Exit.isFailure(firstDefect)) { + assert.include(Cause.pretty(firstDefect.cause), `defect:${defectId}`) + } + const storedDefect = yield* client("failures").Defect( + new KeyedPayload({ id: defectId }) + ).pipe(Effect.exit) + assert.isTrue(Exit.isFailure(storedDefect)) + if (Exit.isFailure(storedDefect)) { + assert.include(Cause.pretty(storedDefect.cause), `defect:${defectId}`) + } + assert.strictEqual(count("Defect", defectId), 1) + + assert.strictEqual( + yield* client("failures").Healthy(new KeyedPayload({ id: `${backend}-healthy` })), + `healthy:${backend}-healthy` + ) + yield* cluster.waitUntil( + "The failure replies were not persisted", + Effect.map(cluster.messageCounts(), (counts) => counts.failed === 2 && counts.replied === 1) + ) + assert.deepStrictEqual(yield* cluster.messageCounts(), { + failed: 2, + replied: 1, + unprocessed: 0 + }) + }).pipe(Effect.scoped)) + + it.live(`${backend}: round-trips a chunked reply through storage`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ backend, entities: PersistenceEntityLayer }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const id = `${backend}-stream` + const values = yield* client("stream").Streamed(new KeyedPayload({ id })).pipe(Stream.runCollect) + assert.deepStrictEqual(Array.from(values), [0, 1, 2, 3, 4]) + yield* cluster.waitUntil( + "The terminal stream reply was not persisted", + Effect.map(cluster.repliedMessageCount, (value) => value === 1) + ) + assert.strictEqual(count("Streamed", id), 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: delivers scheduled messages only after their deadline`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ backend, entities: PersistenceEntityLayer }) + yield* cluster.start(2) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const id = `${backend}-scheduled` + const deliverAt = (yield* Clock.currentTimeMillis) + 1_500 + const replyFiber = yield* client("scheduled").Scheduled( + new ScheduledPayload({ deliverAt, id }) + ).pipe(Effect.forkChild({ startImmediately: true })) + + yield* cluster.waitUntil( + "The early-delivery observation point was not reached", + Effect.map(Clock.currentTimeMillis, (now) => now >= deliverAt - 500), + "2 seconds" + ) + assert.strictEqual(count("Scheduled", id), 0) + assert.deepStrictEqual(state.scheduledDeliveries, []) + + yield* cluster.waitUntil( + "The scheduled message was not delivered after its deadline", + Effect.sync(() => state.scheduledDeliveries.length === 1), + "5 seconds" + ) + yield* cluster.waitUntil( + "The scheduled reply was not persisted", + Effect.map(cluster.repliedMessageCount, (value) => value === 1) + ) + const deliveredAt = yield* Fiber.join(replyFiber) + assert.isAtLeast(deliveredAt, deliverAt) + assert.strictEqual(state.scheduledDeliveries[0], deliveredAt) + }).pipe(Effect.scoped)) + } +}) From 3524b8976f3b13c6c623b1704532e4b23b387e93 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 12:20:55 +1200 Subject: [PATCH 5/9] Add cluster workflow integration tests --- .../test/cluster-integration/Workflow.test.ts | 426 ++++++++++++++++++ .../test/cluster-integration/harness.ts | 27 +- 2 files changed, 448 insertions(+), 5 deletions(-) create mode 100644 packages/platform-node/test/cluster-integration/Workflow.test.ts diff --git a/packages/platform-node/test/cluster-integration/Workflow.test.ts b/packages/platform-node/test/cluster-integration/Workflow.test.ts new file mode 100644 index 00000000000..43a97b16541 --- /dev/null +++ b/packages/platform-node/test/cluster-integration/Workflow.test.ts @@ -0,0 +1,426 @@ +import { assert, describe, it } from "@effect/vitest" +import { Cause, Clock, Duration, Effect, Exit, Fiber, Latch, Layer, Option, Schema } from "effect" +import { ClusterWorkflowEngine, Entity, EntityId } from "effect/unstable/cluster" +import { PersistedQueue } from "effect/unstable/persistence" +import { Rpc } from "effect/unstable/rpc" +import { + Activity, + DurableClock, + DurableDeferred, + DurableQueue, + Workflow, + WorkflowEngine +} from "effect/unstable/workflow" +import { type Backend, type ClusterRunner, make } from "./harness.ts" + +const EndToEndWorkflow = Workflow.make("ClusterIntegrationEndToEnd", { + payload: { + id: Schema.String, + value: Schema.Number + }, + success: Schema.Number, + idempotencyKey: ({ id }) => id +}) + +let endToEndGate = Latch.makeUnsafe(true) +let endToEndEntered = Latch.makeUnsafe() +const endToEndRuns = new Map() + +const EndToEndWorkflowLayer = EndToEndWorkflow.toLayer(({ id, value }) => + Activity.make({ + name: "EndToEnd", + success: Schema.Number, + execute: Effect.gen(function*() { + endToEndRuns.set(id, (endToEndRuns.get(id) ?? 0) + 1) + endToEndEntered.openUnsafe() + yield* endToEndGate.await + return value + 1 + }) + }) +) + +const ReplayGate = DurableDeferred.make("ClusterIntegrationReplayGate", { + success: Schema.String +}) + +const ReplayWorkflow = Workflow.make("ClusterIntegrationReplay", { + payload: { id: Schema.String }, + success: Schema.String, + idempotencyKey: ({ id }) => id +}) + +const replayRuns = new Map() + +const ReplayWorkflowLayer = ReplayWorkflow.toLayer(Effect.fnUntraced(function*({ id }) { + yield* Activity.make({ + name: "BeforeSuspension", + execute: Effect.sync(() => replayRuns.set(id, (replayRuns.get(id) ?? 0) + 1)) + }) + return yield* DurableDeferred.await(ReplayGate) +})) + +const RestartGate = DurableDeferred.make("ClusterIntegrationRestartGate", { + success: Schema.String +}) + +const RestartWorkflow = Workflow.make("ClusterIntegrationRestart", { + payload: { id: Schema.String }, + success: Schema.String, + idempotencyKey: ({ id }) => id +}) + +const RestartWorkflowLayer = RestartWorkflow.toLayer(() => DurableDeferred.await(RestartGate)) + +class RetryError extends Schema.ErrorClass("ClusterIntegrationRetryError")({ + _tag: Schema.tag("ClusterIntegrationRetryError"), + attempt: Schema.Number +}) {} + +const RetryWorkflow = Workflow.make("ClusterIntegrationRetry", { + payload: { + id: Schema.String, + succeed: Schema.Boolean + }, + success: Schema.Number, + error: RetryError, + idempotencyKey: ({ id }) => id +}) + +const retryAttempts = new Map>() + +const RetryWorkflowLayer = RetryWorkflow.toLayer(({ id, succeed }) => + Activity.make({ + name: "Retry", + success: Schema.Number, + error: RetryError, + execute: Effect.gen(function*() { + const attempt = yield* Activity.CurrentAttempt + const attempts = retryAttempts.get(id) ?? [] + attempts.push(attempt) + retryAttempts.set(id, attempts) + if (succeed && attempt === 3) return attempt + return yield* new RetryError({ attempt }) + }) + }).pipe(Activity.retry({ times: 2 })) +) + +const ClockWorkflow = Workflow.make("ClusterIntegrationClock", { + payload: { id: Schema.String }, + success: Schema.Number, + idempotencyKey: ({ id }) => id +}) + +const ClockWorkflowLayer = ClockWorkflow.toLayer(() => + Effect.gen(function*() { + yield* DurableClock.sleep({ + name: "RestartSleep", + duration: "1 second", + inMemoryThreshold: Duration.zero + }) + return yield* Clock.currentTimeMillis + }) +) + +const Queue = DurableQueue.make({ + name: "ClusterIntegrationQueue", + payload: { id: Schema.String }, + success: Schema.String, + idempotencyKey: ({ id }) => id +}) + +const QueueWorkflow = Workflow.make("ClusterIntegrationQueue", { + payload: { id: Schema.String }, + success: Schema.String, + idempotencyKey: ({ id }) => id +}) + +const QueueWorkflowLayer = QueueWorkflow.toLayer(({ id }) => DurableQueue.process(Queue, { id })) +const queueRuns = new Map() +let queueWorkerGate = Latch.makeUnsafe() + +const QueueWorkerLayer = Layer.effectDiscard( + Effect.forkScoped( + Effect.suspend(() => queueWorkerGate.await).pipe( + Effect.andThen(DurableQueue.makeWorker(Queue, ({ id }) => + Effect.sync(() => { + queueRuns.set(id, (queueRuns.get(id) ?? 0) + 1) + return `processed:${id}` + }))) + ) + ) +) + +const InterruptGate = DurableDeferred.make("ClusterIntegrationInterruptGate") + +const InterruptWorkflow = Workflow.make("ClusterIntegrationInterrupt", { + payload: { id: Schema.String }, + idempotencyKey: ({ id }) => id +}) + +const InterruptWorkflowLayer = InterruptWorkflow.toLayer(() => DurableDeferred.await(InterruptGate)) + +const CompleteDeferred = Rpc.make("CompleteDeferred", { + payload: { + token: DurableDeferred.Token, + value: Schema.String + }, + success: Schema.String +}) + +const DeferredControl = Entity.make("ClusterIntegrationDeferredControl", [CompleteDeferred]) + +const DeferredControlLayer = DeferredControl.toLayer(Effect.gen(function*() { + const runner = yield* Entity.CurrentRunnerAddress + return { + CompleteDeferred: ({ payload }) => + DurableDeferred.succeed(RestartGate, payload).pipe( + Effect.as(`${runner.host}:${runner.port}`) + ) + } +})) + +const Workflows = Layer.mergeAll( + EndToEndWorkflowLayer, + ReplayWorkflowLayer, + RestartWorkflowLayer, + RetryWorkflowLayer, + ClockWorkflowLayer, + QueueWorkflowLayer, + QueueWorkerLayer, + InterruptWorkflowLayer, + DeferredControlLayer +) + +const entities = ({ prefix }: { readonly prefix: string }) => { + const queue = PersistedQueue.layer.pipe( + Layer.provideMerge(PersistedQueue.layerStoreSql({ + tableName: `${prefix}_workflow_queue`, + pollInterval: 100, + lockRefreshInterval: 500, + lockExpiration: 1_750 + })) + ) + return Workflows.pipe( + Layer.provide(queue), + Layer.provide(ClusterWorkflowEngine.layer), + Layer.orDie + ) +} + +type Cluster = Effect.Success> + +const withWorkflow = ( + cluster: Cluster, + effect: Effect.Effect +) => Effect.provideService(effect, WorkflowEngine.WorkflowEngine, cluster.workflowEngine) + +const waitForSuspended = Effect.fnUntraced(function*< + Name extends string, + Payload extends Workflow.AnyStructSchema, + Success extends Schema.Top, + Error extends Schema.Top +>( + cluster: Cluster, + workflow: Workflow.Workflow, + executionId: string +) { + yield* cluster.waitUntil( + `${workflow._tag}/${executionId} did not suspend`, + Effect.map(workflow.poll(executionId), (result) => Option.isSome(result) && result.value._tag === "Suspended") + ) +}) + +const waitForComplete = Effect.fnUntraced(function*< + Name extends string, + Payload extends Workflow.AnyStructSchema, + Success extends Schema.Top, + Error extends Schema.Top +>( + cluster: Cluster, + workflow: Workflow.Workflow, + executionId: string +) { + let complete: Workflow.Complete | undefined + yield* cluster.waitUntil( + `${workflow._tag}/${executionId} did not complete`, + Effect.map(workflow.poll(executionId), (result) => { + if (Option.isNone(result) || result.value._tag !== "Complete") return false + complete = result.value + return true + }) + ) + return complete! +}) + +const restart = Effect.fnUntraced(function*(cluster: Cluster) { + const running = cluster.runners.filter((runner) => runner.state() === "running") + yield* Effect.forEach(running, cluster.kill, { concurrency: "unbounded", discard: true }) + const replacements = yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + return replacements +}) + +const workflowOwner = (cluster: Cluster, executionId: string) => { + const shard = cluster.clientSharding.getShardId(EntityId.make(executionId), "default") + return cluster.ownersOfShard(shard)[0] +} + +const findControlOnAnotherRunner = Effect.fnUntraced(function*( + cluster: Cluster, + workflowRunner: ClusterRunner +) { + for (let index = 0; index < 2_000; index++) { + const id = `control-${index}` + const owner = yield* cluster.ownerOfEntity(DeferredControl, id) + if (owner !== undefined && owner !== workflowRunner) return [id, owner] as const + } + return yield* Effect.die("Could not route deferred control to another runner") +}) + +describe("cluster workflow integration", () => { + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: executes end to end and deduplicates concurrent callers`, () => + Effect.gen(function*() { + const id = `${backend}-end-to-end` + endToEndGate = Latch.makeUnsafe() + endToEndEntered = Latch.makeUnsafe() + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + + const first = yield* withWorkflow(cluster, EndToEndWorkflow.execute({ id, value: 41 })).pipe( + Effect.forkChild({ startImmediately: true }) + ) + yield* cluster.waitUntil("The end-to-end activity did not start", Effect.as(endToEndEntered.await, true)) + const second = yield* withWorkflow(cluster, EndToEndWorkflow.execute({ id, value: 41 })).pipe( + Effect.forkChild({ startImmediately: true }) + ) + endToEndGate.openUnsafe() + + assert.strictEqual(yield* Fiber.join(first), 42) + assert.strictEqual(yield* Fiber.join(second), 42) + assert.strictEqual(endToEndRuns.get(id), 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: replays completed activities after the owner dies`, () => + Effect.gen(function*() { + const id = `${backend}-replay` + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const executionId = yield* withWorkflow(cluster, ReplayWorkflow.execute({ id }, { discard: true })) + yield* waitForSuspended(cluster, ReplayWorkflow, executionId) + assert.strictEqual(replayRuns.get(id), 1) + + const owner = workflowOwner(cluster, executionId) + assert.isDefined(owner) + yield* cluster.kill(owner!) + yield* cluster.waitForStableAssignments() + const token = DurableDeferred.tokenFromExecutionId(ReplayGate, { workflow: ReplayWorkflow, executionId }) + yield* withWorkflow(cluster, DurableDeferred.succeed(ReplayGate, { token, value: "resumed" })) + const result = yield* waitForComplete(cluster, ReplayWorkflow, executionId) + + assert.deepStrictEqual(result.exit, Exit.succeed("resumed")) + assert.strictEqual(replayRuns.get(id), 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: resumes deferred workflows across a whole-cluster restart from another runner`, () => + Effect.gen(function*() { + const id = `${backend}-restart` + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const executionId = yield* withWorkflow(cluster, RestartWorkflow.execute({ id }, { discard: true })) + yield* waitForSuspended(cluster, RestartWorkflow, executionId) + yield* restart(cluster) + + const owner = workflowOwner(cluster, executionId) + assert.isDefined(owner) + const [controlId, controlOwner] = yield* findControlOnAnotherRunner(cluster, owner!) + const control = yield* cluster.getClient(DeferredControl) + const token = DurableDeferred.tokenFromExecutionId(RestartGate, { workflow: RestartWorkflow, executionId }) + const completedBy = yield* control(controlId).CompleteDeferred({ token, value: "after-restart" }) + assert.strictEqual(completedBy, `${controlOwner.address.host}:${controlOwner.address.port}`) + + const result = yield* waitForComplete(cluster, RestartWorkflow, executionId) + assert.deepStrictEqual(result.exit, Exit.succeed("after-restart")) + }).pipe(Effect.scoped)) + + it.live(`${backend}: applies activity retry policy and preserves the exhausted error`, () => + Effect.gen(function*() { + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const successId = `${backend}-retry-success` + const failureId = `${backend}-retry-failure` + + assert.strictEqual( + yield* withWorkflow(cluster, RetryWorkflow.execute({ id: successId, succeed: true })), + 3 + ) + const error = yield* withWorkflow( + cluster, + RetryWorkflow.execute({ id: failureId, succeed: false }) + ).pipe(Effect.flip) + + assert.deepStrictEqual(retryAttempts.get(successId), [1, 2, 3]) + assert.deepStrictEqual(retryAttempts.get(failureId), [1, 2, 3]) + assert.strictEqual(error._tag, "ClusterIntegrationRetryError") + assert.strictEqual(error.attempt, 3) + }).pipe(Effect.scoped)) + + it.live(`${backend}: wakes a durable clock after a whole-cluster restart`, () => + Effect.gen(function*() { + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const started = yield* Clock.currentTimeMillis + const executionId = yield* withWorkflow( + cluster, + ClockWorkflow.execute({ id: `${backend}-clock` }, { discard: true }) + ) + yield* waitForSuspended(cluster, ClockWorkflow, executionId) + yield* restart(cluster) + + const result = yield* waitForComplete(cluster, ClockWorkflow, executionId) + assert(Exit.isSuccess(result.exit)) + assert.isAtLeast(result.exit.value - started, 900) + }).pipe(Effect.scoped)) + + it.live(`${backend}: persists queued work across restart and consumes it once`, () => + Effect.gen(function*() { + const id = `${backend}-queue` + queueWorkerGate = Latch.makeUnsafe() + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const executionId = yield* withWorkflow(cluster, QueueWorkflow.execute({ id }, { discard: true })) + yield* waitForSuspended(cluster, QueueWorkflow, executionId) + yield* restart(cluster) + queueWorkerGate.openUnsafe() + + const result = yield* waitForComplete(cluster, QueueWorkflow, executionId) + assert.deepStrictEqual(result.exit, Exit.succeed(`processed:${id}`)) + assert.strictEqual(queueRuns.get(id), 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: persists interruption across a whole-cluster restart`, () => + Effect.gen(function*() { + const cluster = yield* make({ backend, entities }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + const executionId = yield* withWorkflow( + cluster, + InterruptWorkflow.execute({ id: `${backend}-interrupt` }, { discard: true }) + ) + yield* waitForSuspended(cluster, InterruptWorkflow, executionId) + yield* withWorkflow(cluster, InterruptWorkflow.interrupt(executionId)) + yield* restart(cluster) + + const result = yield* waitForComplete(cluster, InterruptWorkflow, executionId) + assert(Exit.isFailure(result.exit)) + assert.isTrue(Exit.hasInterrupts(result.exit)) + assert.isTrue(Cause.hasInterrupts(result.exit.cause)) + }).pipe(Effect.scoped)) + } +}) diff --git a/packages/platform-node/test/cluster-integration/harness.ts b/packages/platform-node/test/cluster-integration/harness.ts index 804b0b9a7cb..11992f7ea95 100644 --- a/packages/platform-node/test/cluster-integration/harness.ts +++ b/packages/platform-node/test/cluster-integration/harness.ts @@ -3,8 +3,10 @@ import { MysqlClient } from "@effect/sql-mysql2" import { PgClient } from "@effect/sql-pg" import { Clock, Context, Duration, Effect, Exit, Latch, Layer, Option, Redacted, Scope } from "effect" import { + ClusterWorkflowEngine, type Entity, EntityId, + type MessageStorage, type Runner as RunnerModel, RunnerAddress, RunnerHealth, @@ -21,6 +23,7 @@ import type { Rpc } from "effect/unstable/rpc" import { RpcSerialization } from "effect/unstable/rpc" import * as SocketServer from "effect/unstable/socket/SocketServer" import { SqlClient } from "effect/unstable/sql" +import { WorkflowEngine } from "effect/unstable/workflow" import { inject } from "vitest" export type Backend = "mysql" | "pg" @@ -43,7 +46,9 @@ export interface MessageCounts { export interface MakeOptions { readonly backend: Backend readonly config?: Partial | undefined - readonly entities: Layer.Layer + readonly entities: + | RunnerEntities + | ((options: { readonly prefix: string }) => RunnerEntities) readonly lockMode?: LockMode | undefined readonly runnerLayer?: RunnerLayer | undefined } @@ -57,6 +62,12 @@ type HarnessConfig = Partial & { readonly shardLockDisableAdvisory: boolean } +type RunnerEntities = Layer.Layer< + never, + never, + Sharding.Sharding | MessageStorage.MessageStorage | SqlClient.SqlClient +> + interface RegistrationRow { readonly address: string readonly healthy: boolean | number @@ -147,7 +158,7 @@ const RunnerHealthLive = RunnerHealth.layerPing.pipe( export const socketRunnerLayer = ( address: RunnerAddress.RunnerAddress, - entities: Layer.Layer, + entities: RunnerEntities, socketServer: SocketServer.SocketServer["Service"], config: HarnessConfig ) => @@ -198,6 +209,7 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { Effect.provide(database) ) const shared = Context.merge(database, messageStorage) + const entities = typeof options.entities === "function" ? options.entities({ prefix }) : options.entities const runners: Array = [] const makeRunnerStorage = Effect.fnUntraced(function*( @@ -214,10 +226,14 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { const clientScope = yield* Scope.fork(parentScope) const clientStorage = yield* makeRunnerStorage(clientScope) - const client = yield* clientLayer(config).pipe( + const clientBase = yield* clientLayer(config).pipe( Layer.buildWithScope(clientScope), Effect.provide(Context.merge(shared, clientStorage)) ) + const workflowEngine = yield* ClusterWorkflowEngine.make.pipe( + Effect.provide(Context.merge(shared, clientBase)) + ) + const client = Context.add(clientBase, WorkflowEngine.WorkflowEngine, workflowEngine) const clientSharding = Context.get(client, Sharding.Sharding) let nextRunnerIndex = 0 const startRunner = Effect.fnUntraced(function*(index: number, startOptions?: StartOptions) { @@ -244,7 +260,7 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { const storage = Context.make(RunnerStorage.RunnerStorage, controller.controlled) const context = yield* (options.runnerLayer ?? socketRunnerLayer)( address, - options.entities, + entities, socketServer, runnerConfig ).pipe( @@ -447,6 +463,7 @@ export const make = Effect.fnUntraced(function*(options: MakeOptions) { unprocessedMessageCount: Effect.map(messageCounts(), (counts) => counts.unprocessed), waitForEntityOwner, waitForStableAssignments, - waitUntil + waitUntil, + workflowEngine } as const }) From e6de6e7e5ab99d30bc58e5be4a22f0c61e5f9a26 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 12:21:02 +1200 Subject: [PATCH 6/9] Add cluster cron integration tests --- .../cluster-integration/ClusterCron.test.ts | 297 ++++++++++++++++++ 1 file changed, 297 insertions(+) create mode 100644 packages/platform-node/test/cluster-integration/ClusterCron.test.ts diff --git a/packages/platform-node/test/cluster-integration/ClusterCron.test.ts b/packages/platform-node/test/cluster-integration/ClusterCron.test.ts new file mode 100644 index 00000000000..9a44551110e --- /dev/null +++ b/packages/platform-node/test/cluster-integration/ClusterCron.test.ts @@ -0,0 +1,297 @@ +import { assert, describe, it } from "@effect/vitest" +import { Clock, Context, Cron, DateTime, Duration, Effect, Latch, Layer } from "effect" +import { ClusterCron, ClusterSchema, Entity } from "effect/unstable/cluster" +import { type Backend, make } from "./harness.ts" + +interface Tick { + readonly at: number + readonly runner: string + readonly scheduled: string +} + +const everySecond = Cron.parseUnsafe("* * * * * *", "UTC") +const testConfig = { shardsPerGroup: 12 } as const + +const addressString = (address: { readonly host: string; readonly port: number }) => `${address.host}:${address.port}` + +const recordTick = (ticks: Array) => + Effect.contextWith((context: Context.Context) => + Effect.gen(function*() { + const address = Context.getUnsafe(context, Entity.CurrentAddress) + const runner = Context.getUnsafe(context, Entity.CurrentRunnerAddress) + const at = yield* Clock.currentTimeMillis + const tick = { + at, + runner: addressString(runner), + scheduled: String(address.entityId) + } + const isFirst = ticks.length === 0 + ticks.push(tick) + return isFirst + }) + ) + +const cronProbe = (name: string, shardGroup = "default") => + Entity.make(`ClusterCron/${name}`, []).annotate(ClusterSchema.ShardGroup, () => shardGroup) + +const nextScheduled = (cron: Cron.Cron, after: DateTime.DateTime.Input) => + DateTime.formatIso(DateTime.fromDateUnsafe(Cron.next(cron, after))) + +const assertScheduledFromExecutionTime = ( + cron: Cron.Cron, + ticks: ReadonlyArray, + startIndex = 1 +) => { + for (let index = startIndex; index < ticks.length; index++) { + assert.strictEqual(ticks[index].scheduled, nextScheduled(cron, ticks[index - 1].at)) + } +} + +const assertScheduledFromPrevious = (cron: Cron.Cron, ticks: ReadonlyArray) => { + assert.strictEqual(ticks[0].scheduled, "initial") + for (let index = 2; index < ticks.length; index++) { + assert.strictEqual(ticks[index].scheduled, nextScheduled(cron, ticks[index - 1].scheduled)) + } +} + +describe("cluster cron integration", () => { + for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { + it.live(`${backend}: runs once per scheduled instant and continues after failure`, () => + Effect.gen(function*() { + const ticks: Array = [] + let failingAttempts = 0 + let successfulAttempts = 0 + const cron = ClusterCron.make({ + name: `basic-${backend}`, + cron: everySecond, + execute: recordTick(ticks) + }) + const failingCron = ClusterCron.make({ + name: `failure-${backend}`, + cron: everySecond, + execute: Effect.suspend(() => { + failingAttempts++ + if (failingAttempts === 1) return Effect.fail("expected cron failure") + successfulAttempts++ + return Effect.void + }) + }) + const cluster = yield* make({ + backend, + config: testConfig, + entities: Layer.merge(cron, failingCron) + }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The cron jobs did not continue through four scheduled instants", + Effect.sync(() => ticks.length >= 4 && failingAttempts >= 4) + ) + + const firstFour = ticks.slice(0, 4) + assert.strictEqual(new Set(firstFour.map((tick) => tick.scheduled)).size, firstFour.length) + assertScheduledFromExecutionTime(everySecond, firstFour) + assert.strictEqual(successfulAttempts, failingAttempts - 1) + }).pipe(Effect.scoped)) + + it.live(`${backend}: calculates the next run from the previous instant or the current time`, () => + Effect.gen(function*() { + const previousTicks: Array = [] + const currentTicks: Array = [] + const gate = Latch.makeUnsafe() + let entered = 0 + const blockedExecution = (ticks: Array) => + Effect.gen(function*() { + if (yield* recordTick(ticks)) { + entered++ + yield* gate.await + } + }).pipe(Effect.uninterruptible) + const previousCron = ClusterCron.make({ + name: `previous-${backend}`, + cron: everySecond, + calculateNextRunFromPrevious: true, + execute: blockedExecution(previousTicks) + }) + const currentCron = ClusterCron.make({ + name: `current-${backend}`, + cron: everySecond, + execute: blockedExecution(currentTicks) + }) + const cluster = yield* make({ + backend, + config: testConfig, + entities: Layer.merge(previousCron, currentCron) + }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "Both cron executions did not enter their first run", + Effect.sync(() => entered === 2) + ) + const blockedAt = yield* Clock.currentTimeMillis + yield* cluster.waitUntil( + "The cron executions were not held across several scheduled instants", + Effect.map(Clock.currentTimeMillis, (now) => now >= blockedAt + 3_200), + "5 seconds" + ) + gate.openUnsafe() + yield* cluster.waitUntil( + "The cron jobs did not resume after their first execution", + Effect.sync(() => previousTicks.length >= 4 && currentTicks.length >= 4), + "12 seconds" + ) + + const previousSecond = DateTime.toEpochMillis(DateTime.makeUnsafe(previousTicks[1].scheduled)) + const currentSecond = DateTime.toEpochMillis(DateTime.makeUnsafe(currentTicks[1].scheduled)) + assertScheduledFromPrevious(everySecond, previousTicks) + assertScheduledFromExecutionTime(everySecond, currentTicks, 2) + assert.isAtLeast(previousTicks[1].at - previousSecond, 2_000) + assert.isAtMost(currentTicks[1].at - currentSecond, 1_000) + }).pipe(Effect.scoped)) + + it.live(`${backend}: catches up or skips stale runs and preserves the schedule across restart`, () => + Effect.gen(function*() { + const catchUpTicks: Array = [] + const skipTicks: Array = [] + const catchUpCron = ClusterCron.make({ + name: `catch-up-${backend}`, + cron: everySecond, + calculateNextRunFromPrevious: true, + execute: recordTick(catchUpTicks) + }) + const skipCron = ClusterCron.make({ + name: `skip-stale-${backend}`, + cron: everySecond, + calculateNextRunFromPrevious: true, + skipIfOlderThan: Duration.millis(500), + execute: recordTick(skipTicks) + }) + const cluster = yield* make({ + backend, + config: testConfig, + entities: Layer.merge(catchUpCron, skipCron) + }) + const runners = yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The cron schedules did not start", + Effect.sync(() => catchUpTicks.length >= 2 && skipTicks.length >= 2) + ) + yield* Effect.forEach(runners, cluster.kill, { discard: true }) + const catchUpBefore = catchUpTicks.length + const skipBefore = skipTicks.length + const lastScheduledBeforeRestart = catchUpTicks[catchUpBefore - 1].scheduled + const stoppedAt = yield* Clock.currentTimeMillis + yield* cluster.waitUntil( + "The cluster downtime window did not elapse", + Effect.map(Clock.currentTimeMillis, (now) => now >= stoppedAt + 3_200), + "5 seconds" + ) + assert.strictEqual(catchUpTicks.length, catchUpBefore) + assert.strictEqual(skipTicks.length, skipBefore) + + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The restarted cluster did not expose the catch-up and stale-skip difference", + Effect.sync(() => { + const caughtUp = catchUpTicks.length - catchUpBefore + const skipped = skipTicks.length - skipBefore + return caughtUp >= 3 && caughtUp >= skipped + 2 + }), + "15 seconds" + ) + yield* cluster.waitUntil( + "The stale-skipping cron did not resume", + Effect.sync(() => skipTicks.length > skipBefore) + ) + + const firstCatchUp = catchUpTicks[catchUpBefore] + const firstAfterSkip = skipTicks[skipBefore] + const firstAfterSkipScheduledAt = DateTime.toEpochMillis(DateTime.makeUnsafe(firstAfterSkip.scheduled)) + assert.strictEqual( + firstCatchUp.scheduled, + nextScheduled(everySecond, lastScheduledBeforeRestart) + ) + assert.isAtLeast(firstCatchUp.at - DateTime.toEpochMillis(DateTime.makeUnsafe(firstCatchUp.scheduled)), 2_000) + assert.isAtMost(firstAfterSkip.at - firstAfterSkipScheduledAt, 750) + assertScheduledFromPrevious(everySecond, catchUpTicks) + }).pipe(Effect.scoped)) + + it.live(`${backend}: resumes without duplicate or missing ticks after the singleton owner dies`, () => + Effect.gen(function*() { + const name = `owner-failover-${backend}` + const ticks: Array = [] + const cron = ClusterCron.make({ + name, + cron: everySecond, + calculateNextRunFromPrevious: true, + execute: recordTick(ticks) + }) + const cluster = yield* make({ backend, config: testConfig, entities: cron }) + yield* cluster.start(3) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The cron did not execute before failover", + Effect.sync(() => ticks.length >= 2) + ) + const owner = yield* cluster.ownerOfEntity(cronProbe(name), name) + assert.isDefined(owner) + yield* cluster.kill(owner!) + const afterKill = ticks.length + yield* cluster.waitUntil( + "The cron did not resume after its singleton owner died", + Effect.sync(() => ticks.length >= afterKill + 3), + "12 seconds" + ) + yield* cluster.waitForStableAssignments() + + assertScheduledFromPrevious(everySecond, ticks) + assert.strictEqual(new Set(ticks.map((tick) => tick.scheduled)).size, ticks.length) + assert.isTrue(ticks.slice(afterKill).every((tick) => tick.runner !== addressString(owner!.address))) + }).pipe(Effect.scoped)) + + it.live(`${backend}: assigns cron singletons and executions to their shard groups`, () => + Effect.gen(function*() { + const defaultName = `default-group-${backend}` + const specialName = `special-group-${backend}` + const defaultTicks: Array = [] + const specialTicks: Array = [] + const defaultCron = ClusterCron.make({ + name: defaultName, + cron: everySecond, + execute: recordTick(defaultTicks) + }) + const specialCron = ClusterCron.make({ + name: specialName, + cron: everySecond, + shardGroup: "special", + execute: recordTick(specialTicks) + }) + const cluster = yield* make({ + backend, + config: { + availableShardGroups: ["default", "special"], + shardsPerGroup: 12 + }, + entities: Layer.merge(defaultCron, specialCron) + }) + const [defaultRunner] = yield* cluster.start(1, { assignedShardGroups: ["default"] }) + const [specialRunner] = yield* cluster.start(1, { assignedShardGroups: ["special"] }) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The shard-group cron jobs did not execute", + Effect.sync(() => defaultTicks.length >= 2 && specialTicks.length >= 2) + ) + + assert.strictEqual(yield* cluster.ownerOfEntity(cronProbe(defaultName), defaultName), defaultRunner) + assert.strictEqual( + yield* cluster.ownerOfEntity(cronProbe(specialName, "special"), specialName), + specialRunner + ) + assert.isTrue(defaultTicks.every((tick) => tick.runner === addressString(defaultRunner.address))) + assert.isTrue(specialTicks.every((tick) => tick.runner === addressString(specialRunner.address))) + }).pipe(Effect.scoped)) + } +}) From f26c376798f16b04e8f37553fe57982dc8a16e77 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 12:41:20 +1200 Subject: [PATCH 7/9] Fix persisted stream recovery --- .../src/unstable/cluster/SqlMessageStorage.ts | 2 +- .../cluster-integration/Persistence.test.ts | 66 ++++++++++++++++++- 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts index c63c403675e..e8c2ed4c0d2 100644 --- a/packages/effect/src/unstable/cluster/SqlMessageStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlMessageStorage.ts @@ -1041,7 +1041,7 @@ const replyKind = { } as const satisfies Record["_tag"], number | null> const replyFromRow = (row: ReplyRow): Reply.Encoded => - Number(row.kind) === replyKind.WithExit ? + row.kind !== null && Number(row.kind) === replyKind.WithExit ? { _tag: "WithExit", id: String(row.id), diff --git a/packages/platform-node/test/cluster-integration/Persistence.test.ts b/packages/platform-node/test/cluster-integration/Persistence.test.ts index bb1ae8c2823..b445811811e 100644 --- a/packages/platform-node/test/cluster-integration/Persistence.test.ts +++ b/packages/platform-node/test/cluster-integration/Persistence.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest" -import { Cause, Clock, DateTime, Effect, Exit, Fiber, Latch, PrimaryKey, Schema, Stream } from "effect" +import { Cause, Clock, DateTime, Effect, Exit, Fiber, Latch, Option, PrimaryKey, Schema, Stream } from "effect" import { ClusterSchema, DeliverAt, Entity } from "effect/unstable/cluster" import { Rpc, RpcSchema } from "effect/unstable/rpc" import { type Backend, make } from "./harness.ts" @@ -88,6 +88,8 @@ const freshState = () => ({ completedVolatile: 0, counts: new Map(), scheduledDeliveries: [] as Array, + streamThirdEntered: Latch.makeUnsafe(), + streamThirdGate: Latch.makeUnsafe(), uninterruptibleEntered: Latch.makeUnsafe(), uninterruptibleGate: Latch.makeUnsafe(), volatileEntered: Latch.makeUnsafe(), @@ -138,7 +140,20 @@ const PersistenceEntityLayer = PersistenceEntity.toLayer({ }), Streamed: (request) => { increment("Streamed", request.payload.id) - return Stream.fromIterable([0, 1, 2, 3, 4]).pipe(Stream.rechunk(1)) + const start = Option.match(request.lastSentChunkValue, { + onNone: () => 0, + onSome: (value) => value + 1 + }) + return Stream.fromIterable([0, 1, 2, 3, 4].slice(start)).pipe( + Stream.mapEffect((value) => { + if (request.payload.id.endsWith("-restart") && value === 2) { + state.streamThirdEntered.openUnsafe() + return Effect.as(state.streamThirdGate.await, value) + } + return Effect.succeed(value) + }), + Stream.rechunk(1) + ) }, TypedFailure: ({ payload }) => Effect.sync(() => increment("TypedFailure", payload.id)).pipe( @@ -369,6 +384,53 @@ describe("cluster message persistence integration", () => { assert.strictEqual(count("Streamed", id), 1) }).pipe(Effect.scoped)) + it.live(`${backend}: resumes a persisted stream after its runner is killed`, () => + Effect.gen(function*() { + resetState() + const cluster = yield* make({ backend, entities: PersistenceEntityLayer }) + const [owner] = yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + const client = yield* cluster.getClient(PersistenceEntity) + const id = `${backend}-stream-restart` + const received: Array = [] + const valuesFiber = yield* client("stream-restart").Streamed(new KeyedPayload({ id })).pipe( + Stream.tap((value) => Effect.sync(() => received.push(value))), + Stream.runCollect, + Effect.forkChild({ startImmediately: true }) + ) + yield* cluster.waitUntil( + "The stream did not deliver two chunks before blocking the third", + Effect.sync(() => received.length === 2 && received[0] === 0 && received[1] === 1) + ) + yield* cluster.waitUntil( + "The stream handler did not block before delivering its third chunk", + Effect.as(state.streamThirdEntered.await, true) + ) + + yield* cluster.kill(owner) + yield* cluster.start(1) + yield* cluster.waitForStableAssignments() + yield* cluster.waitUntil( + "The replacement runner did not resume the persisted stream", + Effect.sync(() => count("Streamed", id) === 2) + ) + state.streamThirdGate.openUnsafe() + + assert.deepStrictEqual(Array.from(yield* Fiber.join(valuesFiber)), [0, 1, 2, 3, 4]) + yield* cluster.waitUntil( + "The terminal stream reply was not persisted after recovery", + Effect.map( + cluster.messageCounts(), + (counts) => counts.replied === 1 && counts.unprocessed === 0 + ) + ) + assert.deepStrictEqual(yield* cluster.messageCounts(), { + failed: 0, + replied: 1, + unprocessed: 0 + }) + }).pipe(Effect.scoped)) + it.live(`${backend}: delivers scheduled messages only after their deadline`, () => Effect.gen(function*() { resetState() From f2da9e4f1e5a8043002db5dc0705d7f3ced5542e Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 13:56:52 +1200 Subject: [PATCH 8/9] Address cluster integration review feedback --- .changeset/fix-cluster-stream-recovery.md | 5 +++++ .../cluster-integration/ClusterCron.test.ts | 10 +++++----- .../test/cluster-integration/Entity.test.ts | 18 +++++++++--------- .../cluster-integration/Persistence.test.ts | 16 ++++++++-------- .../test/cluster-integration/Smoke.test.ts | 2 +- .../test/cluster-integration/Workflow.test.ts | 14 +++++++------- 6 files changed, 35 insertions(+), 30 deletions(-) create mode 100644 .changeset/fix-cluster-stream-recovery.md diff --git a/.changeset/fix-cluster-stream-recovery.md b/.changeset/fix-cluster-stream-recovery.md new file mode 100644 index 00000000000..7c54c638156 --- /dev/null +++ b/.changeset/fix-cluster-stream-recovery.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Fix persisted cluster stream recovery when SQL drivers return a null reply kind. diff --git a/packages/platform-node/test/cluster-integration/ClusterCron.test.ts b/packages/platform-node/test/cluster-integration/ClusterCron.test.ts index 9a44551110e..4223d7e540d 100644 --- a/packages/platform-node/test/cluster-integration/ClusterCron.test.ts +++ b/packages/platform-node/test/cluster-integration/ClusterCron.test.ts @@ -92,7 +92,7 @@ describe("cluster cron integration", () => { assert.strictEqual(new Set(firstFour.map((tick) => tick.scheduled)).size, firstFour.length) assertScheduledFromExecutionTime(everySecond, firstFour) assert.strictEqual(successfulAttempts, failingAttempts - 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: calculates the next run from the previous instant or the current time`, () => Effect.gen(function*() { @@ -148,7 +148,7 @@ describe("cluster cron integration", () => { assertScheduledFromExecutionTime(everySecond, currentTicks, 2) assert.isAtLeast(previousTicks[1].at - previousSecond, 2_000) assert.isAtMost(currentTicks[1].at - currentSecond, 1_000) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: catches up or skips stale runs and preserves the schedule across restart`, () => Effect.gen(function*() { @@ -217,7 +217,7 @@ describe("cluster cron integration", () => { assert.isAtLeast(firstCatchUp.at - DateTime.toEpochMillis(DateTime.makeUnsafe(firstCatchUp.scheduled)), 2_000) assert.isAtMost(firstAfterSkip.at - firstAfterSkipScheduledAt, 750) assertScheduledFromPrevious(everySecond, catchUpTicks) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: resumes without duplicate or missing ticks after the singleton owner dies`, () => Effect.gen(function*() { @@ -250,7 +250,7 @@ describe("cluster cron integration", () => { assertScheduledFromPrevious(everySecond, ticks) assert.strictEqual(new Set(ticks.map((tick) => tick.scheduled)).size, ticks.length) assert.isTrue(ticks.slice(afterKill).every((tick) => tick.runner !== addressString(owner!.address))) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: assigns cron singletons and executions to their shard groups`, () => Effect.gen(function*() { @@ -292,6 +292,6 @@ describe("cluster cron integration", () => { ) assert.isTrue(defaultTicks.every((tick) => tick.runner === addressString(defaultRunner.address))) assert.isTrue(specialTicks.every((tick) => tick.runner === addressString(specialRunner.address))) - }).pipe(Effect.scoped)) + })) } }) diff --git a/packages/platform-node/test/cluster-integration/Entity.test.ts b/packages/platform-node/test/cluster-integration/Entity.test.ts index 54832000789..0864f128d2b 100644 --- a/packages/platform-node/test/cluster-integration/Entity.test.ts +++ b/packages/platform-node/test/cluster-integration/Entity.test.ts @@ -181,7 +181,7 @@ describe("cluster entity integration", () => { const registrations = (yield* cluster.diagnostics()).registrations assert.strictEqual(registrations.length, runners.length) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: reports mailbox saturation and revives idle entities with fresh state`, () => Effect.gen(function*() { @@ -219,7 +219,7 @@ describe("cluster entity integration", () => { assert.strictEqual(first.generation, 1) assert.strictEqual(revived.generation, 2) assert.strictEqual(revived.value, 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: rebalances on runner addition, graceful stop, and abrupt death`, () => Effect.gen(function*() { @@ -273,7 +273,7 @@ describe("cluster entity integration", () => { assert.notStrictEqual(reply.runner, addressString(killed!.address)) yield* cluster.waitForStableAssignments() assert.strictEqual((yield* cluster.messageCounts()).unprocessed, 0) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: transfers frozen row locks after expiry`, () => Effect.gen(function*() { @@ -296,7 +296,7 @@ describe("cluster entity integration", () => { const reply = yield* client(id).Increment(new Request({ id: `${backend}-freeze`, sequence: 0 })) assert.strictEqual(reply.runner, addressString(nextOwner!.address)) yield* cluster.kill(oldOwner!) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: retains frozen advisory locks until the session closes`, () => Effect.gen(function*() { @@ -320,7 +320,7 @@ describe("cluster entity integration", () => { "The advisory lock was not handed over after its session closed", Effect.map(cluster.ownerOfEntity(StateEntity, id), (owner) => owner !== undefined && owner !== oldOwner) ) - }).pipe(Effect.scoped)) + })) } it.live("assigns annotated entities only to runners in their shard group", () => @@ -338,7 +338,7 @@ describe("cluster entity integration", () => { assert.strictEqual(yield* client("grouped").Runner(), addressString(specialRunner.address)) assert.strictEqual(yield* cluster.ownerOfEntity(GroupEntity, "grouped"), specialRunner) assert.strictEqual(yield* cluster.ownerOfEntity(StateEntity, "default"), defaultRunner) - }).pipe(Effect.scoped)) + })) it.live("runs one singleton and migrates it after owner death", () => Effect.gen(function*() { @@ -373,7 +373,7 @@ describe("cluster entity integration", () => { firstOwner ) assert.strictEqual(singleton.maxActive, 1) - }).pipe(Effect.scoped)) + })) it.live("keeps EntityResource alive during movement and releases it explicitly", () => Effect.gen(function*() { @@ -397,7 +397,7 @@ describe("cluster entity integration", () => { "The entity resource was not released", Effect.sync(() => resourceState.released === 1) ) - }).pipe(Effect.scoped)) + })) for (const backend of ["pg", "mysql"] satisfies ReadonlyArray) { it.live(`${backend}: isolates clusters with different table prefixes`, () => @@ -423,6 +423,6 @@ describe("cluster entity integration", () => { ) assert.strictEqual(firstReply.runner, addressString(firstRunner.address)) assert.strictEqual(secondReply.runner, addressString(secondRunner.address)) - }).pipe(Effect.scoped)) + })) } }) diff --git a/packages/platform-node/test/cluster-integration/Persistence.test.ts b/packages/platform-node/test/cluster-integration/Persistence.test.ts index b445811811e..8e4b9194f4f 100644 --- a/packages/platform-node/test/cluster-integration/Persistence.test.ts +++ b/packages/platform-node/test/cluster-integration/Persistence.test.ts @@ -210,7 +210,7 @@ describe("cluster message persistence integration", () => { replied: 1, unprocessed: 0 }) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: serves primary-key duplicates from the stored reply`, () => Effect.gen(function*() { @@ -234,7 +234,7 @@ describe("cluster message persistence integration", () => { replied: 1, unprocessed: 0 }) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: does not lose an uninterruptible request during runner shutdown`, () => Effect.gen(function*() { @@ -272,7 +272,7 @@ describe("cluster message persistence integration", () => { assert.strictEqual(yield* Fiber.join(replyFiber), `uninterruptible:${id}`) yield* Fiber.join(stopping) assert.strictEqual(yield* cluster.repliedMessageCount, 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: does not store or redeliver a volatile request after runner failure`, () => Effect.gen(function*() { @@ -314,7 +314,7 @@ describe("cluster message persistence integration", () => { replied: 0, unprocessed: 0 }) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: persists typed failures and defects without wedging the mailbox`, () => Effect.gen(function*() { @@ -365,7 +365,7 @@ describe("cluster message persistence integration", () => { replied: 1, unprocessed: 0 }) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: round-trips a chunked reply through storage`, () => Effect.gen(function*() { @@ -382,7 +382,7 @@ describe("cluster message persistence integration", () => { Effect.map(cluster.repliedMessageCount, (value) => value === 1) ) assert.strictEqual(count("Streamed", id), 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: resumes a persisted stream after its runner is killed`, () => Effect.gen(function*() { @@ -429,7 +429,7 @@ describe("cluster message persistence integration", () => { replied: 1, unprocessed: 0 }) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: delivers scheduled messages only after their deadline`, () => Effect.gen(function*() { @@ -464,6 +464,6 @@ describe("cluster message persistence integration", () => { const deliveredAt = yield* Fiber.join(replyFiber) assert.isAtLeast(deliveredAt, deliverAt) assert.strictEqual(state.scheduledDeliveries[0], deliveredAt) - }).pipe(Effect.scoped)) + })) } }) diff --git a/packages/platform-node/test/cluster-integration/Smoke.test.ts b/packages/platform-node/test/cluster-integration/Smoke.test.ts index d1813c16b9c..58cc5215ebd 100644 --- a/packages/platform-node/test/cluster-integration/Smoke.test.ts +++ b/packages/platform-node/test/cluster-integration/Smoke.test.ts @@ -50,6 +50,6 @@ describe("cluster integration smoke", () => { "pong:request-2" ) assert.strictEqual(yield* cluster.repliedMessageCount, 2) - }).pipe(Effect.scoped)) + })) } }) diff --git a/packages/platform-node/test/cluster-integration/Workflow.test.ts b/packages/platform-node/test/cluster-integration/Workflow.test.ts index 43a97b16541..9a0283c01b0 100644 --- a/packages/platform-node/test/cluster-integration/Workflow.test.ts +++ b/packages/platform-node/test/cluster-integration/Workflow.test.ts @@ -300,7 +300,7 @@ describe("cluster workflow integration", () => { assert.strictEqual(yield* Fiber.join(first), 42) assert.strictEqual(yield* Fiber.join(second), 42) assert.strictEqual(endToEndRuns.get(id), 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: replays completed activities after the owner dies`, () => Effect.gen(function*() { @@ -322,7 +322,7 @@ describe("cluster workflow integration", () => { assert.deepStrictEqual(result.exit, Exit.succeed("resumed")) assert.strictEqual(replayRuns.get(id), 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: resumes deferred workflows across a whole-cluster restart from another runner`, () => Effect.gen(function*() { @@ -344,7 +344,7 @@ describe("cluster workflow integration", () => { const result = yield* waitForComplete(cluster, RestartWorkflow, executionId) assert.deepStrictEqual(result.exit, Exit.succeed("after-restart")) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: applies activity retry policy and preserves the exhausted error`, () => Effect.gen(function*() { @@ -367,7 +367,7 @@ describe("cluster workflow integration", () => { assert.deepStrictEqual(retryAttempts.get(failureId), [1, 2, 3]) assert.strictEqual(error._tag, "ClusterIntegrationRetryError") assert.strictEqual(error.attempt, 3) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: wakes a durable clock after a whole-cluster restart`, () => Effect.gen(function*() { @@ -385,7 +385,7 @@ describe("cluster workflow integration", () => { const result = yield* waitForComplete(cluster, ClockWorkflow, executionId) assert(Exit.isSuccess(result.exit)) assert.isAtLeast(result.exit.value - started, 900) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: persists queued work across restart and consumes it once`, () => Effect.gen(function*() { @@ -402,7 +402,7 @@ describe("cluster workflow integration", () => { const result = yield* waitForComplete(cluster, QueueWorkflow, executionId) assert.deepStrictEqual(result.exit, Exit.succeed(`processed:${id}`)) assert.strictEqual(queueRuns.get(id), 1) - }).pipe(Effect.scoped)) + })) it.live(`${backend}: persists interruption across a whole-cluster restart`, () => Effect.gen(function*() { @@ -421,6 +421,6 @@ describe("cluster workflow integration", () => { assert(Exit.isFailure(result.exit)) assert.isTrue(Exit.hasInterrupts(result.exit)) assert.isTrue(Cause.hasInterrupts(result.exit.cause)) - }).pipe(Effect.scoped)) + })) } }) From 10f570f0b0241f7fce91d1e0eff8294a68201002 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 14:10:59 +1200 Subject: [PATCH 9/9] Address remaining cluster review feedback --- .../cluster-integration/ClusterCron.test.ts | 12 ++++-- .../test/cluster-integration/Entity.test.ts | 28 ++++++------ .../cluster-integration/Persistence.test.ts | 43 +++++++++---------- 3 files changed, 41 insertions(+), 42 deletions(-) diff --git a/packages/platform-node/test/cluster-integration/ClusterCron.test.ts b/packages/platform-node/test/cluster-integration/ClusterCron.test.ts index 4223d7e540d..8ecb051d466 100644 --- a/packages/platform-node/test/cluster-integration/ClusterCron.test.ts +++ b/packages/platform-node/test/cluster-integration/ClusterCron.test.ts @@ -100,13 +100,15 @@ describe("cluster cron integration", () => { const currentTicks: Array = [] const gate = Latch.makeUnsafe() let entered = 0 - const blockedExecution = (ticks: Array) => - Effect.gen(function*() { + const blockedExecution = Effect.fnUntraced( + function*(ticks: Array) { if (yield* recordTick(ticks)) { entered++ yield* gate.await } - }).pipe(Effect.uninterruptible) + }, + Effect.uninterruptible + ) const previousCron = ClusterCron.make({ name: `previous-${backend}`, cron: everySecond, @@ -178,7 +180,9 @@ describe("cluster cron integration", () => { "The cron schedules did not start", Effect.sync(() => catchUpTicks.length >= 2 && skipTicks.length >= 2) ) - yield* Effect.forEach(runners, cluster.kill, { discard: true }) + for (const runner of runners) { + yield* cluster.kill(runner) + } const catchUpBefore = catchUpTicks.length const skipBefore = skipTicks.length const lastScheduledBeforeRestart = catchUpTicks[catchUpBefore - 1].scheduled diff --git a/packages/platform-node/test/cluster-integration/Entity.test.ts b/packages/platform-node/test/cluster-integration/Entity.test.ts index 0864f128d2b..a0f89435f43 100644 --- a/packages/platform-node/test/cluster-integration/Entity.test.ts +++ b/packages/platform-node/test/cluster-integration/Entity.test.ts @@ -52,15 +52,14 @@ const StateEntityLayer = StateEntity.toLayer( runner: addressString(runner), value: ++value })), - Ordered: ({ payload }) => - Effect.gen(function*() { - order.push(payload.sequence) - if (payload.sequence === 1) { - orderEntered.openUnsafe() - yield* orderGate.await - } - return payload.sequence - }) + Ordered: Effect.fnUntraced(function*({ payload }) { + order.push(payload.sequence) + if (payload.sequence === 1) { + orderEntered.openUnsafe() + yield* orderGate.await + } + return payload.sequence + }) } }), { maxIdleTime: "1 second" } @@ -77,12 +76,11 @@ let mailboxGate = Latch.makeUnsafe(true) let mailboxEntered = Latch.makeUnsafe() const MailboxEntityLayer = MailboxEntity.toLayer({ - Hold: ({ payload }) => - Effect.gen(function*() { - mailboxEntered.openUnsafe() - yield* mailboxGate.await - return payload.sequence - }) + Hold: Effect.fnUntraced(function*({ payload }) { + mailboxEntered.openUnsafe() + yield* mailboxGate.await + return payload.sequence + }) }, { mailboxCapacity: 1 }) const GroupEntity = Entity.make("ClusterIntegrationSpecialGroup", [ diff --git a/packages/platform-node/test/cluster-integration/Persistence.test.ts b/packages/platform-node/test/cluster-integration/Persistence.test.ts index 8e4b9194f4f..850de48f06e 100644 --- a/packages/platform-node/test/cluster-integration/Persistence.test.ts +++ b/packages/platform-node/test/cluster-integration/Persistence.test.ts @@ -126,13 +126,12 @@ const PersistenceEntityLayer = PersistenceEntity.toLayer({ increment("Persisted", payload.id) return `persisted:${payload.id}` }), - Scheduled: ({ payload }) => - Effect.gen(function*() { - increment("Scheduled", payload.id) - const deliveredAt = yield* Clock.currentTimeMillis - state.scheduledDeliveries.push(deliveredAt) - return deliveredAt - }), + Scheduled: Effect.fnUntraced(function*({ payload }) { + increment("Scheduled", payload.id) + const deliveredAt = yield* Clock.currentTimeMillis + state.scheduledDeliveries.push(deliveredAt) + return deliveredAt + }), StoredReply: ({ payload }) => Effect.sync(() => { increment("StoredReply", payload.id) @@ -159,22 +158,20 @@ const PersistenceEntityLayer = PersistenceEntity.toLayer({ Effect.sync(() => increment("TypedFailure", payload.id)).pipe( Effect.andThen(Effect.fail(`typed:${payload.id}`)) ), - Uninterruptible: ({ payload }) => - Effect.gen(function*() { - increment("Uninterruptible", payload.id) - state.uninterruptibleEntered.openUnsafe() - yield* state.uninterruptibleGate.await - state.completedUninterruptible++ - return `uninterruptible:${payload.id}` - }), - Volatile: ({ payload }) => - Effect.gen(function*() { - increment("Volatile", payload.id) - state.volatileEntered.openUnsafe() - yield* state.volatileGate.await - state.completedVolatile++ - return `volatile:${payload.id}` - }) + Uninterruptible: Effect.fnUntraced(function*({ payload }) { + increment("Uninterruptible", payload.id) + state.uninterruptibleEntered.openUnsafe() + yield* state.uninterruptibleGate.await + state.completedUninterruptible++ + return `uninterruptible:${payload.id}` + }), + Volatile: Effect.fnUntraced(function*({ payload }) { + increment("Volatile", payload.id) + state.volatileEntered.openUnsafe() + yield* state.volatileGate.await + state.completedVolatile++ + return `volatile:${payload.id}` + }) }, { disableFatalDefects: true }) describe("cluster message persistence integration", () => {