From 27127e213e8d3530e69bd7ca6adaa0b0d5047bbf Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Tue, 21 Jul 2026 21:42:24 +1200 Subject: [PATCH 01/10] Test refresh timeout for unresponsive reserved connections - Simulate a partitioned SQL connection and verify refresh completes before the shard lock expires --- .../test/cluster/SqlRunnerStorage.test.ts | 75 ++++++++++++++++++- 1 file changed, 73 insertions(+), 2 deletions(-) diff --git a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts index 913899875e5..8e5d6920c3a 100644 --- a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts @@ -1,7 +1,8 @@ import { NodeFileSystem } from "@effect/platform-node" import { SqliteClient } from "@effect/sql-sqlite-node" -import { describe, expect, it } from "@effect/vitest" -import { Effect, FileSystem, Layer } from "effect" +import { assert, describe, expect, it } from "@effect/vitest" +import { Effect, Fiber, FileSystem, Layer } from "effect" +import { TestClock } from "effect/testing" import { Runner, RunnerAddress, @@ -10,12 +11,49 @@ import { ShardingConfig, SqlRunnerStorage } from "effect/unstable/cluster" +import { SqlClient, type SqlConnection } from "effect/unstable/sql" import { MysqlContainer } from "../fixtures/mysql2-utils.ts" import { PgContainer } from "../fixtures/pg-utils.ts" const StorageLive = SqlRunnerStorage.layer describe("SqlRunnerStorage", () => { + it.effect("refresh does not hang when the reserved connection stops responding", () => { + const partitioned = { current: false } + const layer = StorageLive.pipe( + Layer.provideMerge(blackholeReservedConnection(partitioned)), + Layer.provide(ShardingConfig.layer({ + shardLockDisableAdvisory: true, + shardLockExpiration: 1000, + shardLockRefreshInterval: 100 + })) + ) + + return Effect.gen(function*() { + const storage = yield* RunnerStorage.RunnerStorage + const runner = Runner.make({ + address: runnerAddress1, + groups: ["default"], + weight: 1 + }) + const shards = [ShardId.make("default", 1)] + + yield* storage.register(runner, true) + yield* storage.acquire(runnerAddress1, shards) + partitioned.current = true + + const fiber = yield* storage.refresh(runnerAddress1, shards).pipe( + Effect.exit, + Effect.forkChild({ startImmediately: true }) + ) + yield* TestClock.adjust(1001) + const result = fiber.pollUnsafe() + partitioned.current = false + yield* Fiber.interrupt(fiber) + + assert.isDefined(result, "refresh should complete before the shard lock expires") + }).pipe(Effect.provide(layer)) + }, 60_000) ;([ ["pg", Layer.orDie(PgContainer.layerClient)], ["mysql", Layer.orDie(MysqlContainer.layerClient)], @@ -92,6 +130,39 @@ describe("SqlRunnerStorage", () => { const runnerAddress1 = RunnerAddress.make("localhost", 1234) +const blackholeReservedConnection = (partitioned: { readonly current: boolean }) => + Layer.effect( + SqlClient.SqlClient, + Effect.gen(function*() { + const sql = yield* SqlClient.SqlClient + const wrapConnection = (connection: SqlConnection.Connection): SqlConnection.Connection => { + const execute = (effect: Effect.Effect) => + Effect.suspend(() => partitioned.current ? Effect.never : effect) + return { + ...connection, + execute: (...args) => execute(connection.execute(...args)), + executeRaw: (...args) => execute(connection.executeRaw(...args)), + executeValues: (...args) => execute(connection.executeValues(...args)), + executeValuesUnprepared: (...args) => execute(connection.executeValuesUnprepared(...args)), + executeUnprepared: (...args) => execute(connection.executeUnprepared(...args)) + } + } + let client: SqlClient.SqlClient + client = new Proxy(sql, { + get(target, property, receiver) { + if (property === "reserve") { + return Effect.map(target.reserve, wrapConnection) + } + if (property === "withoutTransforms") { + return () => client + } + return Reflect.get(target, property, receiver) + } + }) + return client + }) + ).pipe(Layer.provide(PgContainer.layerClient)) + const SqliteLayer = Effect.gen(function*() { const fs = yield* FileSystem.FileSystem const dir = yield* fs.makeTempDirectoryScoped() From b15cbb553f445c7c3b554b06a12848fda97a352e Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 22 Jul 2026 09:33:11 +1200 Subject: [PATCH 02/10] Prevent SQL runner lock refreshes from hanging - Detach refresh work and bound fiber joins by the shard lock refresh interval --- .changeset/slow-spiders-refresh.md | 5 +++++ packages/effect/src/unstable/cluster/SqlRunnerStorage.ts | 4 ++++ 2 files changed, 9 insertions(+) create mode 100644 .changeset/slow-spiders-refresh.md diff --git a/.changeset/slow-spiders-refresh.md b/.changeset/slow-spiders-refresh.md new file mode 100644 index 00000000000..318ae8eb123 --- /dev/null +++ b/.changeset/slow-spiders-refresh.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Prevent SQL runner lock refreshes from hanging on unresponsive connections. diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index c5787475413..3c47b6b2d55 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -12,6 +12,7 @@ import * as Arr from "../../Array.ts" import * as Duration from "../../Duration.ts" import * as Effect from "../../Effect.ts" +import * as Fiber from "../../Fiber.ts" import * as Layer from "../../Layer.ts" import * as Scope from "../../Scope.ts" import * as SqlClient from "../sql/SqlClient.ts" @@ -574,6 +575,9 @@ export const make = Effect.fnUntraced(function*(options: { shardIds.length > 0 ? Effect.andThen(refreshShards(address, shardIds)) : Effect.as([]), + Effect.forkDetach({ startImmediately: true }), + Effect.flatMap(Fiber.join), + Effect.timeout(config.shardLockRefreshInterval), PersistenceError.refail, withTracerDisabled ), From ab082ca8c348c78e4e8f6378a7fec79cd7c21bdd Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 22 Jul 2026 09:47:57 +1200 Subject: [PATCH 03/10] Reduce SqlRunnerStorage test clock adjustment --- packages/platform-node/test/cluster/SqlRunnerStorage.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts index 8e5d6920c3a..740bc633c03 100644 --- a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts @@ -46,7 +46,7 @@ describe("SqlRunnerStorage", () => { Effect.exit, Effect.forkChild({ startImmediately: true }) ) - yield* TestClock.adjust(1001) + yield* TestClock.adjust(101) const result = fiber.pollUnsafe() partitioned.current = false yield* Fiber.interrupt(fiber) From eaaaef95d9e37ba4667e58544f348b25fe0ee72e Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 22 Jul 2026 13:40:52 +1200 Subject: [PATCH 04/10] Recover sharding safely from shard lock storage failures - Bound SQL shard-lock operations and rebuild stalled connections - Interrupt entities when lock ownership is uncertain - Reacquire assigned shards with fresh entities after storage recovers --- .changeset/slow-spiders-refresh.md | 6 +- .../effect/src/unstable/cluster/Sharding.ts | 166 ++++++++-- .../src/unstable/cluster/ShardingConfig.ts | 6 +- .../src/unstable/cluster/SqlRunnerStorage.ts | 288 ++++++++++-------- .../cluster/internal/entityManager.ts | 40 ++- .../unstable/cluster/internal/shardLock.ts | 9 + packages/effect/test/cluster/Sharding.test.ts | 227 +++++++++++++- .../test/cluster/SqlRunnerStorage.test.ts | 126 +++++++- 8 files changed, 702 insertions(+), 166 deletions(-) create mode 100644 packages/effect/src/unstable/cluster/internal/shardLock.ts diff --git a/.changeset/slow-spiders-refresh.md b/.changeset/slow-spiders-refresh.md index 318ae8eb123..b4b15c41dca 100644 --- a/.changeset/slow-spiders-refresh.md +++ b/.changeset/slow-spiders-refresh.md @@ -2,4 +2,8 @@ "effect": patch --- -Prevent SQL runner lock refreshes from hanging on unresponsive connections. +Bound SQL shard-lock operations so unresponsive connections cannot leave lock +fibers running indefinitely. When shard ownership becomes uncertain, runners +immediately interrupt affected entities and stop acquiring shards until storage +recovers. After recovery, assigned shards are reacquired through the normal lock +path and start fresh entity instances. diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index a202047cf87..2c6f82bca38 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -55,6 +55,7 @@ import { EntityReaper } from "./internal/entityReaper.ts" import { hashString } from "./internal/hash.ts" import { internalInterruptors } from "./internal/interruptors.ts" import { ResourceMap } from "./internal/resourceMap.ts" +import { effectiveInterval } from "./internal/shardLock.ts" import * as Message from "./Message.ts" import * as MessageStorage from "./MessageStorage.ts" import * as Reply from "./Reply.ts" @@ -217,6 +218,7 @@ interface EntityManagerState { const make = Effect.gen(function*() { const config = yield* ShardingConfig + const shardLockInterval = effectiveInterval(config) const shardGroups = shardGroupConfig(config) const getRunnerAddress = () => Option.getOrUndefined(config.runnerAddress) const clock = yield* Clock @@ -240,6 +242,8 @@ const make = Effect.gen(function*() { const shardAssignments = MutableHashMap.empty() const selfShards = MutableHashSet.empty() + let shardLocksHealthy = true + const shardLocksHealthyLatch = Latch.makeUnsafe(true) // the active shards are the ones that we have acquired the lock for const acquiredShards = MutableHashSet.empty() @@ -277,6 +281,9 @@ const make = Effect.gen(function*() { // allow them to move to another runner. const releasingShards = MutableHashSet.empty() + // Shards whose entities must be force interrupted and locks fully released + // before normal reacquisition. + const forceReleasingShards = MutableHashSet.empty() const initialRunnerAddress = getRunnerAddress() if (initialRunnerAddress) { const selfAddress = initialRunnerAddress @@ -286,16 +293,20 @@ const make = Effect.gen(function*() { }) const releaseShardsMap = yield* FiberMap.make() + let forcedShardReleaseRunning = false const releaseShard = Effect.fnUntraced( function*(shardId: ShardId) { const fibers = Arr.empty>() + const force = MutableHashSet.has(forceReleasingShards, shardId) for (const state of entityManagers.values()) { if (state.status === "closed") continue - fibers.push(yield* Effect.forkScoped(state.manager.interruptShard(shardId))) + fibers.push(yield* Effect.forkScoped(state.manager.interruptShard(shardId, { force }))) } yield* Fiber.joinAll(fibers) + yield* shardLocksHealthyLatch.await yield* runnerStorage.release(selfAddress, shardId) MutableHashSet.remove(releasingShards, shardId) + MutableHashSet.remove(forceReleasingShards, shardId) yield* storage.unregisterShardReplyHandlers(shardId) }, Effect.sandbox, @@ -308,15 +319,65 @@ const make = Effect.gen(function*() { fiber: "releaseShard", runner: selfAddress, shardId - }) + }), + // Effect.eventually retries immediately, so space failures to + // avoid hot-looping while storage is unavailable. + Effect.andThen(Effect.sleep(50)) ) ), Effect.eventually, FiberMap.run(releaseShardsMap, shardId, { onlyIfMissing: true }) ) ) + const releaseForcedShards = Effect.suspend(() => { + if (forcedShardReleaseRunning || MutableHashSet.size(forceReleasingShards) === 0) { + return Effect.void + } + forcedShardReleaseRunning = true + const shardIds = [...forceReleasingShards] + return Effect.gen(function*() { + const fibers = Arr.empty>() + for (const shardId of shardIds) { + for (const state of entityManagers.values()) { + if (state.status === "closed") continue + fibers.push(yield* Effect.forkScoped(state.manager.interruptShard(shardId, { force: true }))) + } + } + yield* Fiber.joinAll(fibers) + yield* shardLocksHealthyLatch.await + yield* runnerStorage.releaseAll(selfAddress) + for (const shardId of shardIds) { + MutableHashSet.remove(releasingShards, shardId) + MutableHashSet.remove(forceReleasingShards, shardId) + yield* storage.unregisterShardReplyHandlers(shardId) + } + }).pipe( + Effect.sandbox, + Effect.tapError((cause) => + Effect.logDebug(`Could not release forced shards, retrying`, cause).pipe( + Effect.annotateLogs({ + module: "effect/cluster/Sharding", + fiber: "releaseForcedShards", + runner: selfAddress + }), + // Effect.eventually retries immediately, so space failures to + // avoid hot-looping while storage is unavailable. + Effect.andThen(Effect.sleep(50)) + ) + ), + Effect.eventually, + Effect.ensuring(Effect.sync(() => { + forcedShardReleaseRunning = false + activeShardsLatch.openUnsafe() + })), + Effect.forkIn(shardingScope), + Effect.asVoid + ) + }) const releaseShards = Effect.gen(function*() { + yield* releaseForcedShards for (const shardId of releasingShards) { + if (MutableHashSet.has(forceReleasingShards, shardId)) continue if (FiberMap.hasUnsafe(releaseShardsMap, shardId)) continue yield* releaseShard(shardId) } @@ -341,6 +402,10 @@ const make = Effect.gen(function*() { yield* releaseShards } + if (!shardLocksHealthy) { + continue + } + // if a shard has been assigned to this runner, we acquire it const unacquiredShards = MutableHashSet.empty() for (const shardId of selfShards) { @@ -353,7 +418,7 @@ const make = Effect.gen(function*() { } const oacquired = yield* runnerStorage.acquire(selfAddress, unacquiredShards).pipe( - Effect.timeoutOption(config.shardLockRefreshInterval) + Effect.timeoutOption(shardLockInterval) ) if (Option.isNone(oacquired)) { activeShardsLatch.openUnsafe() @@ -363,10 +428,15 @@ const make = Effect.gen(function*() { const acquired = oacquired.value yield* storage.resetShards(acquired).pipe( Effect.ignore, - Effect.timeoutOption(config.shardLockRefreshInterval) + Effect.timeoutOption(shardLockInterval) ) for (const shardId of acquired) { - if (MutableHashSet.has(releasingShards, shardId) || !MutableHashSet.has(selfShards, shardId)) { + if ( + !shardLocksHealthy || + MutableHashSet.has(releasingShards, shardId) || + !MutableHashSet.has(selfShards, shardId) + ) { + MutableHashSet.add(releasingShards, shardId) continue } MutableHashSet.add(acquiredShards, shardId) @@ -392,8 +462,63 @@ const make = Effect.gen(function*() { Effect.forkIn(shardingScope) ) - // refresh the shard locks every `shardLockRefreshInterval` - yield* Effect.suspend(() => + const markShardLocksUnhealthy = (cause: Cause.Cause) => + Effect.suspend(() => { + if (!shardLocksHealthy) return Effect.void + + shardLocksHealthy = false + shardLocksHealthyLatch.closeUnsafe() + const affectedShards = MutableHashSet.empty() + for (const shardId of acquiredShards) { + MutableHashSet.add(affectedShards, shardId) + } + for (const shardId of releasingShards) { + MutableHashSet.add(affectedShards, shardId) + } + + MutableHashSet.clear(selfShards) + MutableHashSet.clear(acquiredShards) + for (const shardId of affectedShards) { + MutableHashSet.add(releasingShards, shardId) + MutableHashSet.add(forceReleasingShards, shardId) + } + ClusterMetrics.shards.updateUnsafe(BigInt(0), Context.empty()) + activeShardsLatch.openUnsafe() + + return Effect.gen(function*() { + yield* Effect.logError("Shard lock storage is unhealthy", cause) + yield* Effect.forkIn(syncSingletons, shardingScope, { startImmediately: true }) + + for (const shardId of affectedShards) { + for (const state of entityManagers.values()) { + if (state.status === "closed") continue + yield* Effect.forkIn( + state.manager.interruptShard(shardId, { force: true }), + shardingScope, + { startImmediately: true } + ) + } + } + activeShardsLatch.openUnsafe() + }) + }) + + const markShardLocksHealthy = Effect.suspend(() => { + if (shardLocksHealthy) return Effect.void + + shardLocksHealthy = true + shardLocksHealthyLatch.openUnsafe() + MutableHashSet.clear(selfShards) + MutableHashMap.forEach(shardAssignments, (runner, shardId) => { + if (isLocalRunner(runner)) { + MutableHashSet.add(selfShards, shardId) + } + }) + activeShardsLatch.openUnsafe() + return Effect.logInfo("Shard lock storage has recovered") + }) + + const refreshShardLocks = Effect.suspend(() => runnerStorage.refresh(selfAddress, [ ...acquiredShards, ...releasingShards @@ -421,12 +546,20 @@ const make = Effect.gen(function*() { times: 5, schedule: Schedule.spaced(50) }), - Effect.catchCause((cause) => - Effect.logError("Could not refresh shard locks", cause).pipe( - Effect.andThen(clearSelfShards) - ) - ), - Effect.repeat(Schedule.fixed(config.shardLockRefreshInterval)), + Effect.timeout(shardLockInterval), + Effect.catchCause(markShardLocksUnhealthy) + ) + + const probeShardLocks = runnerStorage.refresh(selfAddress, []).pipe( + Effect.timeout(shardLockInterval), + Effect.andThen(markShardLocksHealthy), + Effect.catchCause(() => Effect.void) + ) + + // Refresh shard locks at the lease-safe interval, or probe storage while + // lock ownership is uncertain. + yield* Effect.suspend(() => shardLocksHealthy ? refreshShardLocks : probeShardLocks).pipe( + Effect.repeat(Schedule.fixed(shardLockInterval)), Effect.forever, Effect.forkIn(shardingScope) ) @@ -439,11 +572,6 @@ const make = Effect.gen(function*() { ) } - const clearSelfShards = Effect.sync(() => { - MutableHashSet.clear(selfShards) - activeShardsLatch.openUnsafe() - }) - // --- Storage inbox --- // // Responsible for reading unprocessed messages from storage and sending them @@ -977,7 +1105,7 @@ const make = Effect.gen(function*() { if (newAssignments) { const runner = newAssignments[i] MutableHashMap.set(shardAssignments, shard, runner) - if (isLocalRunner(runner)) { + if (shardLocksHealthy && isLocalRunner(runner)) { MutableHashSet.add(selfShards, shard) } } else { diff --git a/packages/effect/src/unstable/cluster/ShardingConfig.ts b/packages/effect/src/unstable/cluster/ShardingConfig.ts index 5b5c18ae113..f0c582474cb 100644 --- a/packages/effect/src/unstable/cluster/ShardingConfig.ts +++ b/packages/effect/src/unstable/cluster/ShardingConfig.ts @@ -70,7 +70,11 @@ export class ShardingConfig extends Context.Service `${prefix}_${name}` + // Keep all PostgreSQL and MySQL shard-lock operations on a rebuildable + // reserved connection, including when advisory locks are disabled. const acquireLockConn = sql.onDialectOrElse({ pg: () => Effect.fnUntraced(function*(scope: Scope.Scope) { const conn = yield* Effect.orDie(sql.reserve).pipe( Scope.provide(scope) ) - const pid = (yield* conn.executeValues("SELECT pg_backend_pid()", []))[0][0] as number - yield* Scope.addFinalizerExit(scope, () => Effect.orDie(conn.executeRaw("SELECT pg_advisory_unlock_all()", []))) + const pid = disableAdvisoryLocks + ? 0 + : (yield* conn.executeValues("SELECT pg_backend_pid()", []))[0][0] as number + if (!disableAdvisoryLocks) { + yield* Scope.addFinalizerExit(scope, () => + conn.executeRaw("SELECT pg_advisory_unlock_all()", []).pipe( + Effect.timeout(lockOperationInterval), + Effect.interruptible, + Effect.ignoreCause + )) + } return [conn, pid] as const }, Effect.orDie), mysql: () => @@ -80,6 +95,7 @@ export const make = Effect.fnUntraced(function*(options: { const conn = yield* Effect.orDie(sql.reserve).pipe( Scope.provide(scope) ) + if (disableAdvisoryLocks) return [conn, 0] as const // we need to get the connection id using IS_USED_LOCK to properly // support vitess let pid: number | undefined = undefined @@ -92,12 +108,46 @@ export const make = Effect.fnUntraced(function*(options: { if (taken[0] === null) continue pid = taken[1] } - yield* Scope.addFinalizerExit(scope, () => Effect.orDie(conn.executeRaw("SELECT RELEASE_ALL_LOCKS()", []))) + if (!disableAdvisoryLocks) { + yield* Scope.addFinalizerExit(scope, () => + conn.executeRaw("SELECT RELEASE_ALL_LOCKS()", []).pipe( + Effect.timeout(lockOperationInterval), + Effect.interruptible, + Effect.ignoreCause + )) + } return [conn, pid] as const }, Effect.orDie), orElse: () => undefined }) - const lockConn = acquireLockConn && (yield* ResourceRef.from(yield* Effect.scope, acquireLockConn)) + const lockConn = acquireLockConn && (yield* ResourceRef.from(layerScope, acquireLockConn)) + let lockConnRebuilding = false + let lockConnRebuildNeeded = true + const rebuildLockConn = () => { + if ( + !lockConn || + lockConnRebuilding || + !lockConnRebuildNeeded || + lockConn.state.current._tag === "Closed" + ) return Effect.void + lockConnRebuilding = true + return lockConn.rebuildUnsafe().pipe( + Effect.timeout(lockOperationInterval), + Effect.exit, + Effect.tap((exit) => + Effect.sync(() => { + if (Exit.isSuccess(exit) && lockConn.state.current._tag === "Acquired") { + lockConnRebuildNeeded = false + } + }) + ), + Effect.ensuring(Effect.sync(() => { + lockConnRebuilding = false + })), + Effect.forkIn(layerScope, { startImmediately: true }), + Effect.asVoid + ) + } const runnersTable = table("runners") const runnersTableSql = sql(runnersTable) @@ -277,7 +327,7 @@ export const make = Effect.fnUntraced(function*(options: { const [query, params] = effect.compile() return lockConn.await.pipe( Effect.flatMap(([conn]) => conn.executeRaw(query, params)), - Effect.onError(() => lockConn.rebuildUnsafe()) + Effect.onError(rebuildLockConn) ) } const execWithLockConnUnprepared = ( @@ -287,7 +337,7 @@ export const make = Effect.fnUntraced(function*(options: { const [query, params] = effect.compile() return lockConn.await.pipe( Effect.flatMap(([conn]) => conn.executeUnprepared(query, params, undefined)), - Effect.onError(() => lockConn.rebuildUnsafe()) + Effect.onError(rebuildLockConn) ) } const execWithLockConnValues = ( @@ -297,7 +347,7 @@ export const make = Effect.fnUntraced(function*(options: { const [query, params] = effect.compile() return lockConn.await.pipe( Effect.flatMap(([conn]) => conn.executeValues(query, params)), - Effect.onError(() => lockConn.rebuildUnsafe()) + Effect.onError(rebuildLockConn) ) } @@ -315,6 +365,7 @@ export const make = Effect.fnUntraced(function*(options: { WHERE ${locksTableSql}.address = ${address} OR ${locksTableSql}.acquired_at < ${lockExpiresAt} `.pipe( + execWithLockConn, Effect.andThen(acquiredLocks(address, shardIds)) ) } @@ -343,7 +394,7 @@ export const make = Effect.fnUntraced(function*(options: { } } return acquiredShardIds - }, Effect.onError(() => lockConn!.rebuildUnsafe())) + }, Effect.onError(rebuildLockConn)) }, mysql: () => { @@ -357,7 +408,8 @@ export const make = Effect.fnUntraced(function*(options: { ON DUPLICATE KEY UPDATE address = IF(address = VALUES(address) OR acquired_at < ${lockExpiresAt}, VALUES(address), address), acquired_at = IF(address = VALUES(address) OR acquired_at < ${lockExpiresAt}, VALUES(acquired_at), acquired_at) -`.unprepared.pipe( +`.pipe( + execWithLockConnUnprepared, Effect.andThen(acquiredLocks(address, shardIds)) ) } @@ -386,7 +438,7 @@ export const make = Effect.fnUntraced(function*(options: { } } return acquiredShardIds - }, Effect.onError(() => lockConn!.rebuildUnsafe())) + }, Effect.onError(rebuildLockConn)) }, mssql: () => (address: string, shardIds: ReadonlyArray) => { @@ -478,7 +530,8 @@ export const make = Effect.fnUntraced(function*(options: { WHERE address = ${address} AND acquired_at >= ${lockExpiresAt} AND shard_id IN ${stringLiteralArr(shardIds)} - `.values.pipe( + `.pipe( + execWithLockConnValues, Effect.map((rows) => rows.map((row) => row[0] as string)) ) @@ -534,6 +587,89 @@ export const make = Effect.fnUntraced(function*(options: { `.pipe(execWithLockConnValues, Effect.map((rows) => rows.map((row) => row[0] as string))) }) + const withLockOperationDeadline = (operation: Effect.Effect) => + Effect.gen(function*() { + const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) + return yield* Fiber.join(fiber).pipe( + Effect.timeout(lockOperationInterval), + Effect.ensuring( + Fiber.interrupt(fiber).pipe( + Effect.forkIn(layerScope, { startImmediately: true }), + Effect.asVoid + ) + ), + Effect.tap(() => + Effect.sync(() => { + lockConnRebuildNeeded = true + }) + ), + Effect.onError(rebuildLockConn) + ) + }) + + const releaseShard = sql.onDialectOrElse({ + pg: () => { + if (disableAdvisoryLocks) { + return (address: string, shardId: string) => + sql`DELETE FROM ${locksTableSql} WHERE address = ${address} AND shard_id = ${shardId}`.pipe(execWithLockConn) + } + return Effect.fnUntraced( + function*(_address, shardId) { + const lockNum = lockNumbers.get(shardId)! + for (let i = 0; i < 5; i++) { + const [conn] = yield* lockConn!.await + yield* conn.executeRaw(`SELECT pg_advisory_unlock(${lockNum})`, []) + const takenLocks = yield* conn.executeValues( + `SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = pg_backend_pid() AND objid = ${lockNum}`, + [] + ) + if (takenLocks.length === 0) return + } + const [conn] = yield* lockConn!.await + yield* conn.executeRaw(`SELECT pg_advisory_unlock_all()`, []) + }, + Effect.onError(rebuildLockConn), + Effect.asVoid + ) + }, + mysql: () => { + if (disableAdvisoryLocks) { + return (address: string, shardId: string) => + sql`DELETE FROM ${locksTableSql} WHERE address = ${address} AND shard_id = ${shardId}`.pipe(execWithLockConn) + } + return Effect.fnUntraced( + function*(_address, shardId) { + const lockName = lockNames.get(shardId)! + while (true) { + const [conn, pid] = yield* lockConn!.await + yield* conn.executeRaw(`SELECT RELEASE_LOCK('${lockName}')`, []) + const takenLocks = yield* conn.executeValues( + `SELECT IS_USED_LOCK('${lockName}')`, + [] + ) + if (takenLocks.length === 0 || takenLocks[0][0] !== pid) return + } + }, + Effect.onError(rebuildLockConn), + Effect.asVoid + ) + }, + orElse: () => (address: string, shardId: string) => + sql`DELETE FROM ${locksTableSql} WHERE address = ${address} AND shard_id = ${shardId}` + }) + + const releaseAllShards = sql.onDialectOrElse({ + pg: () => (address: string) => + disableAdvisoryLocks + ? sql`DELETE FROM ${locksTableSql} WHERE address = ${address}`.pipe(execWithLockConn) + : sql`SELECT pg_advisory_unlock_all()`.pipe(execWithLockConn, Effect.asVoid), + mysql: () => (address: string) => + disableAdvisoryLocks + ? sql`DELETE FROM ${locksTableSql} WHERE address = ${address}`.pipe(execWithLockConn) + : sql`SELECT RELEASE_ALL_LOCKS()`.pipe(execWithLockConn, Effect.asVoid), + orElse: () => (address: string) => sql`DELETE FROM ${locksTableSql} WHERE address = ${address}` + }) + return RunnerStorage.makeEncoded({ getRunners: sql`SELECT runner, healthy FROM ${runnersTableSql} WHERE last_heartbeat > ${lockExpiresAt}`.values.pipe( PersistenceError.refail, @@ -564,123 +700,37 @@ export const make = Effect.fnUntraced(function*(options: { ), acquire: (address, shardIds) => - acquireLock(address, shardIds).pipe( + withLockOperationDeadline(acquireLock(address, shardIds)).pipe( PersistenceError.refail, withTracerDisabled ), refresh: (address, shardIds) => - sql`UPDATE ${runnersTableSql} SET last_heartbeat = ${sqlNow} WHERE address = ${address}`.pipe( - execWithLockConn, - shardIds.length > 0 ? - Effect.andThen(refreshShards(address, shardIds)) : - Effect.as([]), - Effect.forkDetach({ startImmediately: true }), - Effect.flatMap(Fiber.join), - Effect.timeout(config.shardLockRefreshInterval), + withLockOperationDeadline( + sql`UPDATE ${runnersTableSql} SET last_heartbeat = ${sqlNow} WHERE address = ${address}`.pipe( + execWithLockConn, + shardIds.length > 0 ? + Effect.andThen(refreshShards(address, shardIds)) : + Effect.as([]) + ) + ).pipe( PersistenceError.refail, withTracerDisabled ), - release: sql.onDialectOrElse({ - pg: () => { - if (disableAdvisoryLocks) { - return (address: string, shardId: string) => - sql`DELETE FROM ${locksTableSql} WHERE address = ${address} AND shard_id = ${shardId}`.pipe( - PersistenceError.refail, - withTracerDisabled - ) - } - return Effect.fnUntraced( - function*(_address, shardId) { - const lockNum = lockNumbers.get(shardId)! - for (let i = 0; i < 5; i++) { - const [conn] = yield* lockConn!.await - yield* conn.executeRaw(`SELECT pg_advisory_unlock(${lockNum})`, []) - const takenLocks = yield* conn.executeValues( - `SELECT 1 FROM pg_locks WHERE locktype = 'advisory' AND granted = true AND pid = pg_backend_pid() AND objid = ${lockNum}`, - [] - ) - if (takenLocks.length === 0) return - } - const [conn] = yield* lockConn!.await - yield* conn.executeRaw(`SELECT pg_advisory_unlock_all()`, []) - }, - Effect.onError(() => lockConn!.rebuildUnsafe()), - Effect.asVoid, - PersistenceError.refail, - withTracerDisabled - ) - }, - mysql: () => { - if (disableAdvisoryLocks) { - return (address: string, shardId: string) => - sql`DELETE FROM ${locksTableSql} WHERE address = ${address} AND shard_id = ${shardId}`.pipe( - PersistenceError.refail, - withTracerDisabled - ) - } - return Effect.fnUntraced( - function*(_address, shardId) { - const lockName = lockNames.get(shardId)! - while (true) { - const [conn, pid] = yield* lockConn!.await - yield* conn.executeRaw(`SELECT RELEASE_LOCK('${lockName}')`, []) - const takenLocks = yield* conn.executeValues( - `SELECT IS_USED_LOCK('${lockName}')`, - [] - ) - if (takenLocks.length === 0 || takenLocks[0][0] !== pid) return - } - }, - Effect.onError(() => lockConn!.rebuildUnsafe()), - Effect.asVoid, - PersistenceError.refail, - withTracerDisabled - ) - }, - orElse: () => (address, shardId) => - sql`DELETE FROM ${locksTableSql} WHERE address = ${address} AND shard_id = ${shardId}`.pipe( - PersistenceError.refail, - withTracerDisabled - ) - }), - - releaseAll: sql.onDialectOrElse({ - pg: () => (address) => { - if (disableAdvisoryLocks) { - return sql`DELETE FROM ${locksTableSql} WHERE address = ${address}`.pipe( - PersistenceError.refail, - withTracerDisabled - ) - } - return sql`SELECT pg_advisory_unlock_all()`.pipe( - execWithLockConn, - Effect.asVoid, - PersistenceError.refail, - withTracerDisabled - ) - }, - mysql: () => (address) => { - if (disableAdvisoryLocks) { - return sql`DELETE FROM ${locksTableSql} WHERE address = ${address}`.pipe( - PersistenceError.refail, - withTracerDisabled - ) - } - return sql`SELECT RELEASE_ALL_LOCKS()`.pipe( - execWithLockConn, - Effect.asVoid, - PersistenceError.refail, - withTracerDisabled - ) - }, - orElse: () => (address) => - sql`DELETE FROM ${locksTableSql} WHERE address = ${address}`.pipe( - PersistenceError.refail, - withTracerDisabled - ) - }) + release: (address, shardId) => + withLockOperationDeadline(releaseShard(address, shardId)).pipe( + Effect.asVoid, + PersistenceError.refail, + withTracerDisabled + ), + + releaseAll: (address) => + withLockOperationDeadline(releaseAllShards(address)).pipe( + Effect.asVoid, + PersistenceError.refail, + withTracerDisabled + ) }) }, withTracerDisabled) diff --git a/packages/effect/src/unstable/cluster/internal/entityManager.ts b/packages/effect/src/unstable/cluster/internal/entityManager.ts index ad655d89ad1..16ba9723ba4 100644 --- a/packages/effect/src/unstable/cluster/internal/entityManager.ts +++ b/packages/effect/src/unstable/cluster/internal/entityManager.ts @@ -56,7 +56,9 @@ export interface EntityManager { }) => boolean readonly clearProcessed: () => void - readonly interruptShard: (shardId: ShardId) => Effect.Effect + readonly interruptShard: (shardId: ShardId, options?: { + readonly force?: boolean + }) => Effect.Effect readonly activeEntityCount: Effect.Effect } @@ -117,7 +119,10 @@ export const make = Effect.fnUntraced(function*< entityRpcs.set(KeepAliveRpc._tag, KeepAliveRpc as any) const activeServers = new Map() - const serverCloseLatches = new Map() + const serverCloseLatches = new Map() const processedRequestIds = new Set() const entities: ResourceMap< @@ -132,12 +137,17 @@ export const make = Effect.fnUntraced(function*< const scope = yield* Effect.scope const endLatch = Latch.makeUnsafe() const keepAliveLatch = Latch.makeUnsafe() + const closeLatches = { + closed: Latch.makeUnsafe(), + force: Latch.makeUnsafe() + } + serverCloseLatches.set(address, closeLatches) // on shutdown, reset the storage for the entity yield* Scope.addFinalizerExit( scope, () => { - serverCloseLatches.get(address)?.openUnsafe() + serverCloseLatches.get(address)?.closed.openUnsafe() serverCloseLatches.delete(address) return Effect.void } @@ -373,11 +383,14 @@ export const make = Effect.fnUntraced(function*< scope, Effect.withFiber((fiber) => { activeServers.delete(address.entityId) - serverCloseLatches.set(address, Latch.makeUnsafe()) internalInterruptors.add(fiber.id) - return state.write(0, { _tag: "Eof" }).pipe( - Effect.andThen(Effect.interruptible(endLatch.await)), - Effect.timeoutOption(config.entityTerminationTimeout) + return Effect.raceFirst( + state.write(0, { _tag: "Eof" }).pipe( + Effect.andThen(endLatch.await), + Effect.timeoutOption(config.entityTerminationTimeout), + Effect.interruptible + ), + Effect.interruptible(closeLatches.force.await) ) }) ) @@ -537,17 +550,24 @@ export const make = Effect.fnUntraced(function*< const runFork = Effect.runForkWith(context) return identity({ - interruptShard: (shardId: ShardId) => + interruptShard: (shardId: ShardId, options) => Effect.suspend(function loop(): Effect.Effect { const fibers = Arr.empty>() + if (options?.force === true) { + serverCloseLatches.forEach((latches, address) => { + if (shardId[Equal.symbol](address.shardId)) { + latches.force.openUnsafe() + } + }) + } activeServers.forEach((state) => { if (shardId[Equal.symbol](state.address.shardId)) { fibers.push(runFork(entities.removeIgnore(state.address))) } }) - serverCloseLatches.forEach((latch, address) => { + serverCloseLatches.forEach((latches, address) => { if (shardId[Equal.symbol](address.shardId)) { - fibers.push(runFork(latch.await)) + fibers.push(runFork(latches.closed.await)) } }) if (fibers.length === 0) return Effect.void diff --git a/packages/effect/src/unstable/cluster/internal/shardLock.ts b/packages/effect/src/unstable/cluster/internal/shardLock.ts new file mode 100644 index 00000000000..7e37eed5e2a --- /dev/null +++ b/packages/effect/src/unstable/cluster/internal/shardLock.ts @@ -0,0 +1,9 @@ +import * as Duration from "../../../Duration.ts" +import type { ShardingConfig } from "../ShardingConfig.ts" + +/** @internal */ +export const effectiveInterval = (config: ShardingConfig["Service"]): Duration.Duration => + Duration.min( + Duration.fromInputUnsafe(config.shardLockRefreshInterval), + Duration.divideUnsafe(Duration.fromInputUnsafe(config.shardLockExpiration), 3) + ) diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 2a526bbc692..1765fbade97 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -1,12 +1,17 @@ import { assert, describe, expect, it } from "@effect/vitest" -import { Array, Cause, Clock, Effect, Exit, Fiber, Layer, MutableRef, Option, Queue, Stream } from "effect" +import { Array, Cause, Clock, Context, Effect, Exit, Fiber, Layer, MutableRef, Option, Queue, Stream } from "effect" import { TestClock } from "effect/testing" import { + ClusterMetrics, + EntityId, + MachineId, MessageStorage, + Runner, RunnerAddress, RunnerHealth, Runners, RunnerStorage, + type ShardId, Sharding, ShardingConfig, Snowflake @@ -583,6 +588,226 @@ describe.concurrent("Sharding", () => { })) }) +describe("Sharding shard lock failover", () => { + it.effect("interrupts entities and reacquires shards after lock storage recovers", () => + Effect.gen(function*() { + const storageState: FailoverStorageState = { + blackholed: false, + assignSelf: true, + runner: undefined, + acquireCalls: [], + refreshCalls: [], + releaseCalls: [] + } + const runnerStorage = Layer.effect( + RunnerStorage.RunnerStorage, + Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) + ) + const config = ShardingConfig.layer({ + runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), + shardsPerGroup: 1, + shardLockExpiration: 300, + shardLockRefreshInterval: 1000, + entityTerminationTimeout: 30_000, + entityMessagePollInterval: 10, + refreshAssignmentsInterval: 10, + sendRetryInterval: 10 + }) + const layer = TestEntityNoState.pipe( + Layer.provideMerge(Sharding.layer), + Layer.provide(runnerStorage), + Layer.provide(RunnerHealth.layerNoop), + Layer.provideMerge(TestEntityState.layer), + Layer.provide(Runners.layerNoop), + Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), + Layer.provide(config) + ) + + yield* Effect.gen(function*() { + const sharding = yield* Sharding.Sharding + const entityState = yield* TestEntityState + const makeClient = yield* TestEntity.client + const client = makeClient("1") + const shardId = sharding.getShardId(EntityId.make("1"), "default") + + while (!sharding.hasShardId(shardId)) { + yield* TestClock.adjust(10) + } + while (!storageState.refreshCalls.some((call) => call.shards.length > 0)) { + yield* TestClock.adjust(100) + } + + const entityFiber = yield* client.NeverVolatile().pipe( + Effect.forkChild({ startImmediately: true }) + ) + yield* TestClock.adjust(1) + assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 1) + + const acquireCount = storageState.acquireCalls.length + const partitionedAt = yield* Clock.currentTimeMillis + storageState.blackholed = true + + yield* TestClock.adjust(201) + + assert.isFalse(sharding.hasShardId(shardId)) + assert.isFalse(yield* sharding.isShutdown) + const entityExit = entityFiber.pollUnsafe() + assert(entityExit && Exit.hasInterrupts(entityExit)) + assert.strictEqual(ClusterMetrics.shards.valueUnsafe(Context.empty()).value, BigInt(0)) + + const failedRefreshes = storageState.refreshCalls.filter((call) => + call.at >= partitionedAt && call.shards.length > 0 + ) + assert.isAtMost(failedRefreshes.length, 2) + + yield* TestClock.adjust(1000) + assert.strictEqual(storageState.acquireCalls.length, acquireCount) + assert(storageState.refreshCalls.some((call) => call.at >= partitionedAt && call.shards.length === 0)) + assert( + storageState.refreshCalls + .filter((call) => call.at >= partitionedAt + 200) + .every((call) => call.shards.length === 0) + ) + + storageState.blackholed = false + yield* TestClock.adjust(101) + while (!sharding.hasShardId(shardId)) { + yield* TestClock.adjust(10) + } + + assert.isAbove(storageState.acquireCalls.length, acquireCount) + assert(storageState.acquireCalls.at(-1)!.some((shard) => shard.id === shardId.id)) + expect(yield* client.GetUserVolatile({ id: 2 })).toEqual(new User({ id: 2, name: "User 2" })) + assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 2) + }).pipe(Effect.provide(layer), Effect.scoped) + })) + + it.effect("keeps the graceful timeout for normal shard reassignment", () => + Effect.gen(function*() { + const storageState: FailoverStorageState = { + blackholed: false, + assignSelf: true, + runner: undefined, + acquireCalls: [], + refreshCalls: [], + releaseCalls: [] + } + const runnerStorage = Layer.effect( + RunnerStorage.RunnerStorage, + Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) + ) + const config = ShardingConfig.layer({ + runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), + shardsPerGroup: 1, + shardLockExpiration: 3000, + shardLockRefreshInterval: 100, + entityTerminationTimeout: 1000, + entityMessagePollInterval: 10, + refreshAssignmentsInterval: 10, + sendRetryInterval: 10 + }) + const layer = TestEntityNoState.pipe( + Layer.provideMerge(Sharding.layer), + Layer.provide(runnerStorage), + Layer.provide(RunnerHealth.layerNoop), + Layer.provideMerge(TestEntityState.layer), + Layer.provide(Runners.layerNoop), + Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), + Layer.provide(config) + ) + + yield* Effect.gen(function*() { + const sharding = yield* Sharding.Sharding + const makeClient = yield* TestEntity.client + const client = makeClient("1") + const shardId = sharding.getShardId(EntityId.make("1"), "default") + + while (!sharding.hasShardId(shardId)) { + yield* TestClock.adjust(10) + } + const entityFiber = yield* client.NeverVolatile().pipe( + Effect.forkChild({ startImmediately: true }) + ) + yield* TestClock.adjust(1) + + storageState.assignSelf = false + while (sharding.hasShardId(shardId)) { + yield* TestClock.adjust(10) + } + while ((yield* sharding.activeEntityCount) > 0) { + yield* TestClock.adjust(1) + } + + assert.isUndefined(entityFiber.pollUnsafe()) + assert.strictEqual(storageState.releaseCalls.length, 0) + yield* TestClock.adjust(900) + assert.isUndefined(entityFiber.pollUnsafe()) + assert.strictEqual(storageState.releaseCalls.length, 0) + + for (let i = 0; i < 20 && entityFiber.pollUnsafe() === undefined; i++) { + yield* TestClock.adjust(10) + } + const entityExit = entityFiber.pollUnsafe() + assert(entityExit && Exit.hasInterrupts(entityExit)) + assert.strictEqual(storageState.releaseCalls.length, 1) + }).pipe(Effect.provide(layer), Effect.scoped) + })) +}) + +interface FailoverStorageState { + blackholed: boolean + assignSelf: boolean + runner: Runner.Runner | undefined + readonly acquireCalls: Array> + readonly refreshCalls: Array<{ + readonly at: number + readonly shards: Array + }> + readonly releaseCalls: Array +} + +const makeFailoverStorage = (state: FailoverStorageState, clock: Clock.Clock) => + RunnerStorage.RunnerStorage.of({ + getRunners: Effect.sync(() => { + if (!state.runner) return [] + if (state.assignSelf) return [[state.runner, true]] + return [[state.runner, false], [otherRunner, true]] + }), + register: (runner) => + Effect.sync(() => { + state.runner = runner + return MachineId.make(1) + }), + unregister: () => Effect.void, + setRunnerHealth: () => Effect.void, + acquire: (_address, shardIds) => + Effect.sync(() => { + const shards = globalThis.Array.from(shardIds) + state.acquireCalls.push(shards) + return shards + }), + refresh: (_address, shardIds) => + Effect.suspend(() => { + const shards = globalThis.Array.from(shardIds) + state.refreshCalls.push({ + at: clock.currentTimeMillisUnsafe(), + shards + }) + return state.blackholed ? Effect.never : Effect.succeed(shards) + }), + release: (_address, shardId) => + Effect.sync(() => { + state.releaseCalls.push(shardId) + }), + releaseAll: () => Effect.void + }) + +const otherRunner = Runner.make({ + address: RunnerAddress.make("localhost", 5678), + groups: ["default"], + weight: 1 +}) + const TestShardingConfig = ShardingConfig.layer({ entityMailboxCapacity: 10, entityTerminationTimeout: 0, diff --git a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts index 740bc633c03..9d4e78ae9c6 100644 --- a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts @@ -1,7 +1,7 @@ import { NodeFileSystem } from "@effect/platform-node" import { SqliteClient } from "@effect/sql-sqlite-node" import { assert, describe, expect, it } from "@effect/vitest" -import { Effect, Fiber, FileSystem, Layer } from "effect" +import { Duration, Effect, Exit, FileSystem, Layer, Schedule } from "effect" import { TestClock } from "effect/testing" import { Runner, @@ -18,12 +18,16 @@ import { PgContainer } from "../fixtures/pg-utils.ts" const StorageLive = SqlRunnerStorage.layer describe("SqlRunnerStorage", () => { - it.effect("refresh does not hang when the reserved connection stops responding", () => { - const partitioned = { current: false } + it.effect("bounds shard lock operations and rebuilds an unresponsive reserved connection", () => { + const partitioned: PartitionState = { + current: false, + activeQueries: 0, + maxActiveQueries: 0, + interruptedQueries: 0 + } const layer = StorageLive.pipe( - Layer.provideMerge(blackholeReservedConnection(partitioned)), + Layer.provideMerge(blackholeReservedConnection(partitioned, true)), Layer.provide(ShardingConfig.layer({ - shardLockDisableAdvisory: true, shardLockExpiration: 1000, shardLockRefreshInterval: 100 })) @@ -42,16 +46,83 @@ describe("SqlRunnerStorage", () => { yield* storage.acquire(runnerAddress1, shards) partitioned.current = true - const fiber = yield* storage.refresh(runnerAddress1, shards).pipe( - Effect.exit, - Effect.forkChild({ startImmediately: true }) - ) - yield* TestClock.adjust(101) - const result = fiber.pollUnsafe() + const expectDeadline = Effect.fnUntraced(function*(operation: Effect.Effect) { + const [elapsed, exit] = yield* operation.pipe( + Effect.exit, + Effect.timed, + TestClock.withLive + ) + assert(Exit.isFailure(exit)) + assert.isAtLeast(Duration.toMillis(elapsed), 90) + assert.isBelow(Duration.toMillis(elapsed), 1000) + yield* Effect.sleep(20).pipe(TestClock.withLive) + }) + + yield* expectDeadline(storage.refresh(runnerAddress1, shards)) + yield* expectDeadline(storage.refresh(runnerAddress1, shards)) + yield* expectDeadline(storage.refresh(runnerAddress1, shards)) + + assert.isAtLeast(partitioned.interruptedQueries, 1) + assert.isAtMost(partitioned.maxActiveQueries, 1) + + partitioned.current = false + expect(yield* storage.refresh(runnerAddress1, shards).pipe(TestClock.withLive)).toEqual(shards) + + partitioned.current = true + yield* expectDeadline(storage.acquire(runnerAddress1, [ShardId.make("default", 2)])) + partitioned.current = false + yield* storage.refresh(runnerAddress1, shards).pipe(TestClock.withLive) + + partitioned.current = true + yield* expectDeadline(storage.release(runnerAddress1, shards[0])) partitioned.current = false - yield* Fiber.interrupt(fiber) + yield* storage.refresh(runnerAddress1, shards).pipe(TestClock.withLive) + yield* storage.release(runnerAddress1, shards[0]).pipe(TestClock.withLive) + + assert.strictEqual(partitioned.activeQueries, 0) + }).pipe(Effect.provide(layer)) + }, 60_000) + + it.effect("recovers when a blackholed query cannot resume after the partition clears", () => { + const partitioned: PartitionState = { + current: false, + activeQueries: 0, + maxActiveQueries: 0, + interruptedQueries: 0 + } + const layer = StorageLive.pipe( + Layer.provideMerge(blackholeReservedConnection(partitioned, false)), + Layer.provide(ShardingConfig.layer({ + shardLockDisableAdvisory: true, + shardLockExpiration: 1000, + shardLockRefreshInterval: 100 + })) + ) + + return Effect.gen(function*() { + const storage = yield* RunnerStorage.RunnerStorage + const runner = Runner.make({ + address: runnerAddress1, + groups: ["default"], + weight: 1 + }) + const shards = [ShardId.make("default", 1)] + + yield* storage.register(runner, true) + yield* storage.acquire(runnerAddress1, shards) + partitioned.current = true + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + yield* Effect.sleep(150).pipe(TestClock.withLive) - assert.isDefined(result, "refresh should complete before the shard lock expires") + partitioned.current = false + expect( + yield* storage.refresh(runnerAddress1, shards).pipe( + Effect.retry({ times: 5, schedule: Schedule.spaced(20) }), + TestClock.withLive + ) + ).toEqual(shards) + assert.isAtLeast(partitioned.interruptedQueries, 1) + assert.isAtMost(partitioned.maxActiveQueries, 1) }).pipe(Effect.provide(layer)) }, 60_000) ;([ @@ -130,14 +201,39 @@ describe("SqlRunnerStorage", () => { const runnerAddress1 = RunnerAddress.make("localhost", 1234) -const blackholeReservedConnection = (partitioned: { readonly current: boolean }) => +interface PartitionState { + current: boolean + activeQueries: number + maxActiveQueries: number + interruptedQueries: number +} + +const blackholeReservedConnection = (partitioned: PartitionState, resumePending: boolean) => Layer.effect( SqlClient.SqlClient, Effect.gen(function*() { const sql = yield* SqlClient.SqlClient const wrapConnection = (connection: SqlConnection.Connection): SqlConnection.Connection => { const execute = (effect: Effect.Effect) => - Effect.suspend(() => partitioned.current ? Effect.never : effect) + Effect.suspend(() => { + partitioned.activeQueries++ + partitioned.maxActiveQueries = Math.max(partitioned.maxActiveQueries, partitioned.activeQueries) + return Effect.suspend(function waitForConnection(): Effect.Effect { + if (!partitioned.current) return effect + return resumePending + ? Effect.andThen(Effect.sleep(5), waitForConnection) + : Effect.never + }).pipe( + Effect.onExit((exit) => + Effect.sync(() => { + partitioned.activeQueries-- + if (Exit.hasInterrupts(exit)) { + partitioned.interruptedQueries++ + } + }) + ) + ) + }) return { ...connection, execute: (...args) => execute(connection.execute(...args)), From 3f54e51af51b35dbbed3f9fb9f73ddd5d9748ca0 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 22 Jul 2026 13:55:11 +1200 Subject: [PATCH 05/10] Fix lock deadline tracing and sharding test assertion - Define the lock deadline helper with Effect.fnUntraced - Replace the remaining Vitest expectation with an Effect assertion --- .../src/unstable/cluster/SqlRunnerStorage.ts | 37 +++++++++---------- packages/effect/test/cluster/Sharding.test.ts | 2 +- 2 files changed, 19 insertions(+), 20 deletions(-) diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index 69df1b18297..6afd32c2313 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -587,25 +587,24 @@ export const make = Effect.fnUntraced(function*(options: { `.pipe(execWithLockConnValues, Effect.map((rows) => rows.map((row) => row[0] as string))) }) - const withLockOperationDeadline = (operation: Effect.Effect) => - Effect.gen(function*() { - const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) - return yield* Fiber.join(fiber).pipe( - Effect.timeout(lockOperationInterval), - Effect.ensuring( - Fiber.interrupt(fiber).pipe( - Effect.forkIn(layerScope, { startImmediately: true }), - Effect.asVoid - ) - ), - Effect.tap(() => - Effect.sync(() => { - lockConnRebuildNeeded = true - }) - ), - Effect.onError(rebuildLockConn) - ) - }) + const withLockOperationDeadline = Effect.fnUntraced(function*(operation: Effect.Effect) { + const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) + return yield* Fiber.join(fiber).pipe( + Effect.timeout(lockOperationInterval), + Effect.ensuring( + Fiber.interrupt(fiber).pipe( + Effect.forkIn(layerScope, { startImmediately: true }), + Effect.asVoid + ) + ), + Effect.tap(() => + Effect.sync(() => { + lockConnRebuildNeeded = true + }) + ), + Effect.onError(rebuildLockConn) + ) + }) const releaseShard = sql.onDialectOrElse({ pg: () => { diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 1765fbade97..31c1f606423 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -677,7 +677,7 @@ describe("Sharding shard lock failover", () => { assert.isAbove(storageState.acquireCalls.length, acquireCount) assert(storageState.acquireCalls.at(-1)!.some((shard) => shard.id === shardId.id)) - expect(yield* client.GetUserVolatile({ id: 2 })).toEqual(new User({ id: 2, name: "User 2" })) + assert.deepStrictEqual(yield* client.GetUserVolatile({ id: 2 }), new User({ id: 2, name: "User 2" })) assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 2) }).pipe(Effect.provide(layer), Effect.scoped) })) From 0b5703cecad3951d267b9a80e47c4642f419cc8d Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Wed, 22 Jul 2026 14:27:43 +1200 Subject: [PATCH 06/10] Document SQL lock operation timeout behavior - Clarify that the deadline applies while joining cleanup fibers --- packages/effect/src/unstable/cluster/SqlRunnerStorage.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index 6afd32c2313..4be35fb0203 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -588,6 +588,8 @@ export const make = Effect.fnUntraced(function*(options: { }) const withLockOperationDeadline = Effect.fnUntraced(function*(operation: Effect.Effect) { + // Effect.timeout waits for the timed-out effect to finish interrupting. + // Timeout the join so unresponsive SQL cleanup cannot extend the deadline. const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) return yield* Fiber.join(fiber).pipe( Effect.timeout(lockOperationInterval), From 72ee21be4585f902dddf2b6015c002f1fd38cde8 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 24 Jul 2026 19:04:54 +1200 Subject: [PATCH 07/10] Harden shard release and lock operation cleanup - Reuse latch state and retry logic across shard release paths - Avoid interrupting completed SQL lock operations --- .../effect/src/unstable/cluster/Sharding.ts | 109 ++++++++---------- .../src/unstable/cluster/SqlRunnerStorage.ts | 6 +- 2 files changed, 48 insertions(+), 67 deletions(-) diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 2c6f82bca38..7a06c355d66 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -242,7 +242,7 @@ const make = Effect.gen(function*() { const shardAssignments = MutableHashMap.empty() const selfShards = MutableHashSet.empty() - let shardLocksHealthy = true + // open while shard lock storage is healthy const shardLocksHealthyLatch = Latch.makeUnsafe(true) // the active shards are the ones that we have acquired the lock for @@ -294,38 +294,59 @@ const make = Effect.gen(function*() { const releaseShardsMap = yield* FiberMap.make() let forcedShardReleaseRunning = false - const releaseShard = Effect.fnUntraced( - function*(shardId: ShardId) { - const fibers = Arr.empty>() - const force = MutableHashSet.has(forceReleasingShards, shardId) + // Interrupt the shards' entities, wait for lock health, run the storage + // release, then clear the shards' bookkeeping. + const runShardRelease = Effect.fnUntraced(function*( + shardIds: ReadonlyArray, + force: boolean, + release: Effect.Effect + ) { + const fibers = Arr.empty>() + for (const shardId of shardIds) { for (const state of entityManagers.values()) { if (state.status === "closed") continue fibers.push(yield* Effect.forkScoped(state.manager.interruptShard(shardId, { force }))) } - yield* Fiber.joinAll(fibers) - yield* shardLocksHealthyLatch.await - yield* runnerStorage.release(selfAddress, shardId) + } + yield* Fiber.joinAll(fibers) + yield* shardLocksHealthyLatch.await + yield* release + for (const shardId of shardIds) { MutableHashSet.remove(releasingShards, shardId) MutableHashSet.remove(forceReleasingShards, shardId) yield* storage.unregisterShardReplyHandlers(shardId) - }, - Effect.sandbox, - (effect, shardId) => + } + }) + const retryShardRelease = + (annotations: { readonly fiber: string; readonly shardId?: ShardId }) => + (effect: Effect.Effect) => effect.pipe( + Effect.sandbox, Effect.tapError((cause) => - Effect.logDebug(`Could not release shard, retrying`, cause).pipe( + Effect.logDebug(`Could not release shards, retrying`, cause).pipe( Effect.annotateLogs({ module: "effect/cluster/Sharding", - fiber: "releaseShard", runner: selfAddress, - shardId + ...annotations }), // Effect.eventually retries immediately, so space failures to // avoid hot-looping while storage is unavailable. Effect.andThen(Effect.sleep(50)) ) ), - Effect.eventually, + Effect.eventually + ) + const releaseShard = Effect.fnUntraced( + function*(shardId: ShardId) { + yield* runShardRelease( + [shardId], + MutableHashSet.has(forceReleasingShards, shardId), + runnerStorage.release(selfAddress, shardId) + ) + }, + (effect, shardId) => + effect.pipe( + retryShardRelease({ fiber: "releaseShard", shardId }), FiberMap.run(releaseShardsMap, shardId, { onlyIfMissing: true }) ) ) @@ -335,37 +356,8 @@ const make = Effect.gen(function*() { } forcedShardReleaseRunning = true const shardIds = [...forceReleasingShards] - return Effect.gen(function*() { - const fibers = Arr.empty>() - for (const shardId of shardIds) { - for (const state of entityManagers.values()) { - if (state.status === "closed") continue - fibers.push(yield* Effect.forkScoped(state.manager.interruptShard(shardId, { force: true }))) - } - } - yield* Fiber.joinAll(fibers) - yield* shardLocksHealthyLatch.await - yield* runnerStorage.releaseAll(selfAddress) - for (const shardId of shardIds) { - MutableHashSet.remove(releasingShards, shardId) - MutableHashSet.remove(forceReleasingShards, shardId) - yield* storage.unregisterShardReplyHandlers(shardId) - } - }).pipe( - Effect.sandbox, - Effect.tapError((cause) => - Effect.logDebug(`Could not release forced shards, retrying`, cause).pipe( - Effect.annotateLogs({ - module: "effect/cluster/Sharding", - fiber: "releaseForcedShards", - runner: selfAddress - }), - // Effect.eventually retries immediately, so space failures to - // avoid hot-looping while storage is unavailable. - Effect.andThen(Effect.sleep(50)) - ) - ), - Effect.eventually, + return runShardRelease(shardIds, true, runnerStorage.releaseAll(selfAddress)).pipe( + retryShardRelease({ fiber: "releaseForcedShards" }), Effect.ensuring(Effect.sync(() => { forcedShardReleaseRunning = false activeShardsLatch.openUnsafe() @@ -402,7 +394,7 @@ const make = Effect.gen(function*() { yield* releaseShards } - if (!shardLocksHealthy) { + if (!shardLocksHealthyLatch.isOpen()) { continue } @@ -432,7 +424,7 @@ const make = Effect.gen(function*() { ) for (const shardId of acquired) { if ( - !shardLocksHealthy || + !shardLocksHealthyLatch.isOpen() || MutableHashSet.has(releasingShards, shardId) || !MutableHashSet.has(selfShards, shardId) ) { @@ -464,18 +456,9 @@ const make = Effect.gen(function*() { const markShardLocksUnhealthy = (cause: Cause.Cause) => Effect.suspend(() => { - if (!shardLocksHealthy) return Effect.void - - shardLocksHealthy = false - shardLocksHealthyLatch.closeUnsafe() - const affectedShards = MutableHashSet.empty() - for (const shardId of acquiredShards) { - MutableHashSet.add(affectedShards, shardId) - } - for (const shardId of releasingShards) { - MutableHashSet.add(affectedShards, shardId) - } + if (!shardLocksHealthyLatch.closeUnsafe()) return Effect.void + const affectedShards = MutableHashSet.fromIterable([...acquiredShards, ...releasingShards]) MutableHashSet.clear(selfShards) MutableHashSet.clear(acquiredShards) for (const shardId of affectedShards) { @@ -504,10 +487,8 @@ const make = Effect.gen(function*() { }) const markShardLocksHealthy = Effect.suspend(() => { - if (shardLocksHealthy) return Effect.void + if (!shardLocksHealthyLatch.openUnsafe()) return Effect.void - shardLocksHealthy = true - shardLocksHealthyLatch.openUnsafe() MutableHashSet.clear(selfShards) MutableHashMap.forEach(shardAssignments, (runner, shardId) => { if (isLocalRunner(runner)) { @@ -558,7 +539,7 @@ const make = Effect.gen(function*() { // Refresh shard locks at the lease-safe interval, or probe storage while // lock ownership is uncertain. - yield* Effect.suspend(() => shardLocksHealthy ? refreshShardLocks : probeShardLocks).pipe( + yield* Effect.suspend(() => shardLocksHealthyLatch.isOpen() ? refreshShardLocks : probeShardLocks).pipe( Effect.repeat(Schedule.fixed(shardLockInterval)), Effect.forever, Effect.forkIn(shardingScope) @@ -1105,7 +1086,7 @@ const make = Effect.gen(function*() { if (newAssignments) { const runner = newAssignments[i] MutableHashMap.set(shardAssignments, shard, runner) - if (shardLocksHealthy && isLocalRunner(runner)) { + if (shardLocksHealthyLatch.isOpen() && isLocalRunner(runner)) { MutableHashSet.add(selfShards, shard) } } else { diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index 4be35fb0203..b0cda4b0ecd 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -593,12 +593,12 @@ export const make = Effect.fnUntraced(function*(options: { const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) return yield* Fiber.join(fiber).pipe( Effect.timeout(lockOperationInterval), - Effect.ensuring( - Fiber.interrupt(fiber).pipe( + Effect.ensuring(Effect.suspend(() => + fiber.pollUnsafe() !== undefined ? Effect.void : Fiber.interrupt(fiber).pipe( Effect.forkIn(layerScope, { startImmediately: true }), Effect.asVoid ) - ), + )), Effect.tap(() => Effect.sync(() => { lockConnRebuildNeeded = true From a7f22c2f76d872a4e7de0cbee9ddcf437d4faa8b Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sat, 25 Jul 2026 17:03:27 +1200 Subject: [PATCH 08/10] Wait for forced lock release before reacquiring shards - Skip shard acquisition while a forced releaseAll is pending - Cover reacquisition during a pending forced release Co-Authored-By: Claude Opus 5 --- .changeset/slow-spiders-refresh.md | 3 +- .../effect/src/unstable/cluster/Sharding.ts | 17 +- packages/effect/test/cluster/Sharding.test.ts | 151 +++++++++++++++--- 3 files changed, 146 insertions(+), 25 deletions(-) diff --git a/.changeset/slow-spiders-refresh.md b/.changeset/slow-spiders-refresh.md index b4b15c41dca..cf066cc787f 100644 --- a/.changeset/slow-spiders-refresh.md +++ b/.changeset/slow-spiders-refresh.md @@ -6,4 +6,5 @@ Bound SQL shard-lock operations so unresponsive connections cannot leave lock fibers running indefinitely. When shard ownership becomes uncertain, runners immediately interrupt affected entities and stop acquiring shards until storage recovers. After recovery, assigned shards are reacquired through the normal lock -path and start fresh entity instances. +path and start fresh entity instances, once the forced release of the previously +held locks has completed. diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index 7a06c355d66..f6ecd0907f2 100644 --- a/packages/effect/src/unstable/cluster/Sharding.ts +++ b/packages/effect/src/unstable/cluster/Sharding.ts @@ -350,6 +350,11 @@ const make = Effect.gen(function*() { FiberMap.run(releaseShardsMap, shardId, { onlyIfMissing: true }) ) ) + // The forced release ends with `runnerStorage.releaseAll`, which drops + // every lock held by this runner. Shards must not be reacquired through the + // normal path while it is pending, otherwise the bulk release wipes a lock + // the runner already considers acquired. + const forcedShardReleasePending = () => forcedShardReleaseRunning || MutableHashSet.size(forceReleasingShards) > 0 const releaseForcedShards = Effect.suspend(() => { if (forcedShardReleaseRunning || MutableHashSet.size(forceReleasingShards) === 0) { return Effect.void @@ -389,7 +394,7 @@ const make = Effect.gen(function*() { MutableHashSet.add(releasingShards, shardId) } - if (MutableHashSet.size(releasingShards) > 0) { + if (MutableHashSet.size(releasingShards) > 0 || MutableHashSet.size(forceReleasingShards) > 0) { yield* Effect.forkIn(syncSingletons, shardingScope) yield* releaseShards } @@ -398,6 +403,12 @@ const make = Effect.gen(function*() { continue } + // Wait for the pending bulk release before reacquiring, so it cannot + // drop a lock acquired here. `releaseForcedShards` reopens the latch. + if (forcedShardReleasePending()) { + continue + } + // if a shard has been assigned to this runner, we acquire it const unacquiredShards = MutableHashSet.empty() for (const shardId of selfShards) { @@ -422,9 +433,13 @@ const make = Effect.gen(function*() { Effect.ignore, Effect.timeoutOption(shardLockInterval) ) + // A forced release can start while `acquire` is in flight, so re-check + // it here as well as before acquiring. + const forcedReleasePending = forcedShardReleasePending() for (const shardId of acquired) { if ( !shardLocksHealthyLatch.isOpen() || + forcedReleasePending || MutableHashSet.has(releasingShards, shardId) || !MutableHashSet.has(selfShards, shardId) ) { diff --git a/packages/effect/test/cluster/Sharding.test.ts b/packages/effect/test/cluster/Sharding.test.ts index 31c1f606423..e177e099a97 100644 --- a/packages/effect/test/cluster/Sharding.test.ts +++ b/packages/effect/test/cluster/Sharding.test.ts @@ -11,7 +11,7 @@ import { RunnerHealth, Runners, RunnerStorage, - type ShardId, + ShardId, Sharding, ShardingConfig, Snowflake @@ -591,14 +591,7 @@ describe.concurrent("Sharding", () => { describe("Sharding shard lock failover", () => { it.effect("interrupts entities and reacquires shards after lock storage recovers", () => Effect.gen(function*() { - const storageState: FailoverStorageState = { - blackholed: false, - assignSelf: true, - runner: undefined, - acquireCalls: [], - refreshCalls: [], - releaseCalls: [] - } + const storageState = makeFailoverStorageState() const runnerStorage = Layer.effect( RunnerStorage.RunnerStorage, Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) @@ -676,7 +669,7 @@ describe("Sharding shard lock failover", () => { } assert.isAbove(storageState.acquireCalls.length, acquireCount) - assert(storageState.acquireCalls.at(-1)!.some((shard) => shard.id === shardId.id)) + assert(storageState.acquireCalls.at(-1)!.shards.some((shard) => shard.id === shardId.id)) assert.deepStrictEqual(yield* client.GetUserVolatile({ id: 2 }), new User({ id: 2, name: "User 2" })) assert.strictEqual(Queue.sizeUnsafe(entityState.envelopes), 2) }).pipe(Effect.provide(layer), Effect.scoped) @@ -684,14 +677,7 @@ describe("Sharding shard lock failover", () => { it.effect("keeps the graceful timeout for normal shard reassignment", () => Effect.gen(function*() { - const storageState: FailoverStorageState = { - blackholed: false, - assignSelf: true, - runner: undefined, - acquireCalls: [], - refreshCalls: [], - releaseCalls: [] - } + const storageState = makeFailoverStorageState() const runnerStorage = Layer.effect( RunnerStorage.RunnerStorage, Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) @@ -752,26 +738,132 @@ describe("Sharding shard lock failover", () => { assert.strictEqual(storageState.releaseCalls.length, 1) }).pipe(Effect.provide(layer), Effect.scoped) })) + + it.effect("does not acquire shards while a forced release is pending", () => + Effect.gen(function*() { + const shardsPerGroup = 4 + const storageState = makeFailoverStorageState({ + otherRunnerHealthy: true, + releaseAllDuration: 500 + }) + const runnerStorage = Layer.effect( + RunnerStorage.RunnerStorage, + Effect.map(Clock.Clock, (clock) => makeFailoverStorage(storageState, clock)) + ) + const config = ShardingConfig.layer({ + runnerAddress: Option.some(RunnerAddress.make("localhost", 1234)), + shardsPerGroup, + shardLockExpiration: 300, + shardLockRefreshInterval: 1000, + entityTerminationTimeout: 0, + entityMessagePollInterval: 10, + refreshAssignmentsInterval: 10, + sendRetryInterval: 10 + }) + const layer = TestEntityNoState.pipe( + Layer.provideMerge(Sharding.layer), + Layer.provide(runnerStorage), + Layer.provide(RunnerHealth.layerNoop), + Layer.provideMerge(TestEntityState.layer), + Layer.provide(Runners.layerNoop), + Layer.provide([MessageStorage.layerMemory, Snowflake.layerGenerator]), + Layer.provide(config) + ) + + yield* Effect.gen(function*() { + const sharding = yield* Sharding.Sharding + const allShards = Array.makeBy(shardsPerGroup, (i) => ShardId.make("default", i + 1)) + const ownedCount = () => allShards.filter((shardId) => sharding.hasShardId(shardId)).length + + // the other runner holds part of the ring, so this runner starts with a + // strict subset of the shards + while (ownedCount() === 0) { + yield* TestClock.adjust(10) + } + assert.isBelow(ownedCount(), shardsPerGroup) + while (!storageState.refreshCalls.some((call) => call.shards.length > 0)) { + yield* TestClock.adjust(100) + } + + const acquiresBeforeOutage = storageState.acquireCalls.length + storageState.blackholed = true + yield* TestClock.adjust(201) + assert.strictEqual(ownedCount(), 0) + + // the other runner's shards are reassigned to this runner during the + // outage, so they are not part of the forced release set + storageState.otherRunnerHealthy = false + yield* TestClock.adjust(100) + assert.strictEqual(storageState.releaseAllCalls.length, 0) + + // recovery runs the forced release, which stays in flight for + // `releaseAllDuration` + storageState.blackholed = false + while (storageState.releaseAllCalls.length === 0) { + yield* TestClock.adjust(10) + } + while (!storageState.releaseAllCalls[0].completed) { + yield* TestClock.adjust(10) + } + // keep shutdown from blocking on the finalizer release + storageState.releaseAllDuration = 0 + + while (ownedCount() < shardsPerGroup) { + yield* TestClock.adjust(10) + } + + // `releaseAll` drops every lock held by this runner, so nothing may be + // acquired before it has completed + assert.strictEqual(storageState.releaseAllCalls.length, 1) + assert( + storageState.acquireCalls + .slice(acquiresBeforeOutage) + .every((call) => call.completedReleaseAlls > 0) + ) + }).pipe(Effect.provide(layer), Effect.scoped) + })) }) interface FailoverStorageState { blackholed: boolean assignSelf: boolean + otherRunnerHealthy: boolean + /** Test clock duration `releaseAll` stays in flight for. */ + releaseAllDuration: number runner: Runner.Runner | undefined - readonly acquireCalls: Array> + readonly acquireCalls: Array<{ + readonly shards: Array + readonly completedReleaseAlls: number + }> readonly refreshCalls: Array<{ readonly at: number readonly shards: Array }> readonly releaseCalls: Array + readonly releaseAllCalls: Array<{ completed: boolean }> } +const makeFailoverStorageState = ( + overrides?: Partial +): FailoverStorageState => ({ + blackholed: false, + assignSelf: true, + otherRunnerHealthy: false, + releaseAllDuration: 0, + runner: undefined, + acquireCalls: [], + refreshCalls: [], + releaseCalls: [], + releaseAllCalls: [], + ...overrides +}) + const makeFailoverStorage = (state: FailoverStorageState, clock: Clock.Clock) => RunnerStorage.RunnerStorage.of({ getRunners: Effect.sync(() => { if (!state.runner) return [] - if (state.assignSelf) return [[state.runner, true]] - return [[state.runner, false], [otherRunner, true]] + if (!state.assignSelf) return [[state.runner, false], [otherRunner, true]] + return state.otherRunnerHealthy ? [[state.runner, true], [otherRunner, true]] : [[state.runner, true]] }), register: (runner) => Effect.sync(() => { @@ -783,7 +875,10 @@ const makeFailoverStorage = (state: FailoverStorageState, clock: Clock.Clock) => acquire: (_address, shardIds) => Effect.sync(() => { const shards = globalThis.Array.from(shardIds) - state.acquireCalls.push(shards) + state.acquireCalls.push({ + shards, + completedReleaseAlls: state.releaseAllCalls.filter((call) => call.completed).length + }) return shards }), refresh: (_address, shardIds) => @@ -799,7 +894,17 @@ const makeFailoverStorage = (state: FailoverStorageState, clock: Clock.Clock) => Effect.sync(() => { state.releaseCalls.push(shardId) }), - releaseAll: () => Effect.void + releaseAll: () => + Effect.suspend(() => { + const call = { completed: false } + state.releaseAllCalls.push(call) + return Effect.andThen( + Effect.sleep(state.releaseAllDuration), + Effect.sync(() => { + call.completed = true + }) + ) + }) }) const otherRunner = Runner.make({ From 139c693db24277c4b0cc830a84933374adcde04f Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sat, 25 Jul 2026 18:19:36 +1200 Subject: [PATCH 09/10] Keep rebuilding unresponsive lock connections Rebuilds were only re-armed by a successful lock operation, so once a rebuild succeeded the runner could not rebuild again while every operation kept failing - exactly the unresponsive-connection case. Track a connection generation instead: a failing operation schedules a rebuild unless the connection it ran on has already been replaced. Co-Authored-By: Claude Opus 5 --- .changeset/slow-spiders-refresh.md | 4 +- .../src/unstable/cluster/SqlRunnerStorage.ts | 40 ++++--- .../test/cluster/SqlRunnerStorage.test.ts | 109 +++++++++++++++--- 3 files changed, 119 insertions(+), 34 deletions(-) diff --git a/.changeset/slow-spiders-refresh.md b/.changeset/slow-spiders-refresh.md index cf066cc787f..6106e0fa3c3 100644 --- a/.changeset/slow-spiders-refresh.md +++ b/.changeset/slow-spiders-refresh.md @@ -7,4 +7,6 @@ fibers running indefinitely. When shard ownership becomes uncertain, runners immediately interrupt affected entities and stop acquiring shards until storage recovers. After recovery, assigned shards are reacquired through the normal lock path and start fresh entity instances, once the forced release of the previously -held locks has completed. +held locks has completed. Every failing lock operation keeps scheduling a +rebuild of the reserved connection, so a replacement connection that is also +unresponsive is rebuilt again instead of wedging the runner. diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index b0cda4b0ecd..e07aa9ae2eb 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -122,12 +122,15 @@ export const make = Effect.fnUntraced(function*(options: { }) const lockConn = acquireLockConn && (yield* ResourceRef.from(layerScope, acquireLockConn)) let lockConnRebuilding = false - let lockConnRebuildNeeded = true - const rebuildLockConn = () => { + // Incremented every time the reserved connection is replaced, so failures + // from operations that ran on an already replaced connection do not trigger + // another rebuild. + let lockConnGeneration = 0 + const rebuildLockConn = (generation: number) => { if ( !lockConn || lockConnRebuilding || - !lockConnRebuildNeeded || + generation !== lockConnGeneration || lockConn.state.current._tag === "Closed" ) return Effect.void lockConnRebuilding = true @@ -137,7 +140,7 @@ export const make = Effect.fnUntraced(function*(options: { Effect.tap((exit) => Effect.sync(() => { if (Exit.isSuccess(exit) && lockConn.state.current._tag === "Acquired") { - lockConnRebuildNeeded = false + lockConnGeneration++ } }) ), @@ -148,6 +151,14 @@ export const make = Effect.fnUntraced(function*(options: { Effect.asVoid ) } + // Rebuild the reserved connection when `effect` fails on it. Failures keep + // scheduling rebuilds, so a rebuilt connection that is also unresponsive is + // replaced again. + const onErrorRebuildLockConn = (effect: Effect.Effect): Effect.Effect => + Effect.suspend(() => { + const generation = lockConnGeneration + return Effect.onError(effect, () => rebuildLockConn(generation)) + }) const runnersTable = table("runners") const runnersTableSql = sql(runnersTable) @@ -327,7 +338,7 @@ export const make = Effect.fnUntraced(function*(options: { const [query, params] = effect.compile() return lockConn.await.pipe( Effect.flatMap(([conn]) => conn.executeRaw(query, params)), - Effect.onError(rebuildLockConn) + onErrorRebuildLockConn ) } const execWithLockConnUnprepared = ( @@ -337,7 +348,7 @@ export const make = Effect.fnUntraced(function*(options: { const [query, params] = effect.compile() return lockConn.await.pipe( Effect.flatMap(([conn]) => conn.executeUnprepared(query, params, undefined)), - Effect.onError(rebuildLockConn) + onErrorRebuildLockConn ) } const execWithLockConnValues = ( @@ -347,7 +358,7 @@ export const make = Effect.fnUntraced(function*(options: { const [query, params] = effect.compile() return lockConn.await.pipe( Effect.flatMap(([conn]) => conn.executeValues(query, params)), - Effect.onError(rebuildLockConn) + onErrorRebuildLockConn ) } @@ -394,7 +405,7 @@ export const make = Effect.fnUntraced(function*(options: { } } return acquiredShardIds - }, Effect.onError(rebuildLockConn)) + }, onErrorRebuildLockConn) }, mysql: () => { @@ -438,7 +449,7 @@ export const make = Effect.fnUntraced(function*(options: { } } return acquiredShardIds - }, Effect.onError(rebuildLockConn)) + }, onErrorRebuildLockConn) }, mssql: () => (address: string, shardIds: ReadonlyArray) => { @@ -599,12 +610,7 @@ export const make = Effect.fnUntraced(function*(options: { Effect.asVoid ) )), - Effect.tap(() => - Effect.sync(() => { - lockConnRebuildNeeded = true - }) - ), - Effect.onError(rebuildLockConn) + onErrorRebuildLockConn ) }) @@ -629,7 +635,7 @@ export const make = Effect.fnUntraced(function*(options: { const [conn] = yield* lockConn!.await yield* conn.executeRaw(`SELECT pg_advisory_unlock_all()`, []) }, - Effect.onError(rebuildLockConn), + onErrorRebuildLockConn, Effect.asVoid ) }, @@ -651,7 +657,7 @@ export const make = Effect.fnUntraced(function*(options: { if (takenLocks.length === 0 || takenLocks[0][0] !== pid) return } }, - Effect.onError(rebuildLockConn), + onErrorRebuildLockConn, Effect.asVoid ) }, diff --git a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts index 9d4e78ae9c6..fe3001589c4 100644 --- a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts @@ -11,7 +11,7 @@ import { ShardingConfig, SqlRunnerStorage } from "effect/unstable/cluster" -import { SqlClient, type SqlConnection } from "effect/unstable/sql" +import { SqlClient, type SqlConnection, SqlError } from "effect/unstable/sql" import { MysqlContainer } from "../fixtures/mysql2-utils.ts" import { PgContainer } from "../fixtures/pg-utils.ts" @@ -19,12 +19,7 @@ const StorageLive = SqlRunnerStorage.layer describe("SqlRunnerStorage", () => { it.effect("bounds shard lock operations and rebuilds an unresponsive reserved connection", () => { - const partitioned: PartitionState = { - current: false, - activeQueries: 0, - maxActiveQueries: 0, - interruptedQueries: 0 - } + const partitioned = makePartitionState() const layer = StorageLive.pipe( Layer.provideMerge(blackholeReservedConnection(partitioned, true)), Layer.provide(ShardingConfig.layer({ @@ -84,12 +79,7 @@ describe("SqlRunnerStorage", () => { }, 60_000) it.effect("recovers when a blackholed query cannot resume after the partition clears", () => { - const partitioned: PartitionState = { - current: false, - activeQueries: 0, - maxActiveQueries: 0, - interruptedQueries: 0 - } + const partitioned = makePartitionState() const layer = StorageLive.pipe( Layer.provideMerge(blackholeReservedConnection(partitioned, false)), Layer.provide(ShardingConfig.layer({ @@ -125,6 +115,51 @@ describe("SqlRunnerStorage", () => { assert.isAtMost(partitioned.maxActiveQueries, 1) }).pipe(Effect.provide(layer)) }, 60_000) + + it.effect("rebuilds the reserved connection again when a rebuilt connection stops responding", () => { + const partitioned = makePartitionState() + const layer = StorageLive.pipe( + Layer.provideMerge(blackholeReservedConnection(partitioned, false)), + Layer.provide(ShardingConfig.layer({ + shardLockExpiration: 1000, + shardLockRefreshInterval: 100 + })) + ) + + return Effect.gen(function*() { + const storage = yield* RunnerStorage.RunnerStorage + const runner = Runner.make({ + address: runnerAddress1, + groups: ["default"], + weight: 1 + }) + const shards = [ShardId.make("default", 1)] + + yield* storage.register(runner, true) + yield* storage.acquire(runnerAddress1, shards) + + // a failing lock operation rebuilds the reserved connection + partitioned.failNextQueries = 1 + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + yield* waitUntil(() => partitioned.usableConnections === 2) + + // the rebuilt connection then wedges, without any lock operation + // succeeding in between - a further rebuild still has to be attempted + const reserved = partitioned.reservedConnections + partitioned.current = true + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + partitioned.current = false + yield* waitUntil(() => partitioned.reservedConnections > reserved) + + expect( + yield* storage.refresh(runnerAddress1, shards).pipe( + Effect.retry({ times: 5, schedule: Schedule.spaced(20) }), + TestClock.withLive + ) + ).toEqual(shards) + }).pipe(Effect.provide(layer)) + }, 60_000) ;([ ["pg", Layer.orDie(PgContainer.layerClient)], ["mysql", Layer.orDie(MysqlContainer.layerClient)], @@ -206,16 +241,51 @@ interface PartitionState { activeQueries: number maxActiveQueries: number interruptedQueries: number + failNextQueries: number + reservedConnections: number + usableConnections: number } +const makePartitionState = (): PartitionState => ({ + current: false, + activeQueries: 0, + maxActiveQueries: 0, + interruptedQueries: 0, + failNextQueries: 0, + reservedConnections: 0, + usableConnections: 0 +}) + +const waitUntil = Effect.fnUntraced( + function*(predicate: () => boolean) { + while (!predicate()) { + yield* Effect.sleep(20) + } + }, + Effect.timeoutOrElse({ + duration: 10_000, + orElse: () => Effect.die("timed out waiting for condition") + }), + TestClock.withLive +) + const blackholeReservedConnection = (partitioned: PartitionState, resumePending: boolean) => Layer.effect( SqlClient.SqlClient, Effect.gen(function*() { const sql = yield* SqlClient.SqlClient const wrapConnection = (connection: SqlConnection.Connection): SqlConnection.Connection => { - const execute = (effect: Effect.Effect) => - Effect.suspend(() => { + let usable = false + const execute = (effect: Effect.Effect) => + Effect.suspend((): Effect.Effect => { + if (partitioned.failNextQueries > 0) { + partitioned.failNextQueries-- + return Effect.fail( + new SqlError.SqlError({ + reason: new SqlError.ConnectionError({ cause: new Error("connection lost") }) + }) + ) + } partitioned.activeQueries++ partitioned.maxActiveQueries = Math.max(partitioned.maxActiveQueries, partitioned.activeQueries) return Effect.suspend(function waitForConnection(): Effect.Effect { @@ -230,6 +300,10 @@ const blackholeReservedConnection = (partitioned: PartitionState, resumePending: if (Exit.hasInterrupts(exit)) { partitioned.interruptedQueries++ } + if (Exit.isSuccess(exit) && !usable) { + usable = true + partitioned.usableConnections++ + } }) ) ) @@ -247,7 +321,10 @@ const blackholeReservedConnection = (partitioned: PartitionState, resumePending: client = new Proxy(sql, { get(target, property, receiver) { if (property === "reserve") { - return Effect.map(target.reserve, wrapConnection) + return Effect.map(target.reserve, (connection) => { + partitioned.reservedConnections++ + return wrapConnection(connection) + }) } if (property === "withoutTransforms") { return () => client From f542e4cd3f0e59a31b7910ab13f66778ea60172b Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sat, 25 Jul 2026 19:13:36 +1200 Subject: [PATCH 10/10] Bound lock connection rebuilds Rebuilding the reserved lock connection wrapped `rebuildUnsafe` in a plain `Effect.timeout`, which waits for the timed-out effect to finish interrupting. The rebuild starts by closing the previous scope, so a driver that stalls releasing the unresponsive connection kept the rebuild fiber alive past its deadline, left `lockConnRebuilding` set forever, and disabled every later rebuild. Reuse the fork-and-timeout-the-join pattern already used for lock operations, so the stalled release finishes detached and the flag is always reset. Co-Authored-By: Claude Opus 5 --- .changeset/slow-spiders-refresh.md | 4 +- .../src/unstable/cluster/SqlRunnerStorage.ts | 43 ++++++----- .../test/cluster/SqlRunnerStorage.test.ts | 75 ++++++++++++++++++- 3 files changed, 100 insertions(+), 22 deletions(-) diff --git a/.changeset/slow-spiders-refresh.md b/.changeset/slow-spiders-refresh.md index 6106e0fa3c3..ef55e25e2f1 100644 --- a/.changeset/slow-spiders-refresh.md +++ b/.changeset/slow-spiders-refresh.md @@ -9,4 +9,6 @@ recovers. After recovery, assigned shards are reacquired through the normal lock path and start fresh entity instances, once the forced release of the previously held locks has completed. Every failing lock operation keeps scheduling a rebuild of the reserved connection, so a replacement connection that is also -unresponsive is rebuilt again instead of wedging the runner. +unresponsive is rebuilt again instead of wedging the runner. Rebuilds are +themselves bounded, so a connection release that never completes cannot block +every later rebuild. diff --git a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts index e07aa9ae2eb..51d21df5a37 100644 --- a/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts +++ b/packages/effect/src/unstable/cluster/SqlRunnerStorage.ts @@ -121,6 +121,25 @@ export const make = Effect.fnUntraced(function*(options: { orElse: () => undefined }) const lockConn = acquireLockConn && (yield* ResourceRef.from(layerScope, acquireLockConn)) + + // `Effect.timeout` waits for the timed-out effect to finish interrupting, so + // an operation stuck in an uninterruptible region (such as a scope finalizer + // releasing an unresponsive connection) can outlive its deadline. Fork the + // operation and timeout the join instead, leaving stalled cleanup to finish + // detached in the layer scope. + const withDeadline = Effect.fnUntraced(function*(operation: Effect.Effect) { + const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) + return yield* Fiber.join(fiber).pipe( + Effect.timeout(lockOperationInterval), + Effect.ensuring(Effect.suspend(() => + fiber.pollUnsafe() !== undefined ? Effect.void : Fiber.interrupt(fiber).pipe( + Effect.forkIn(layerScope, { startImmediately: true }), + Effect.asVoid + ) + )) + ) + }) + let lockConnRebuilding = false // Incremented every time the reserved connection is replaced, so failures // from operations that ran on an already replaced connection do not trigger @@ -134,8 +153,11 @@ export const make = Effect.fnUntraced(function*(options: { lockConn.state.current._tag === "Closed" ) return Effect.void lockConnRebuilding = true - return lockConn.rebuildUnsafe().pipe( - Effect.timeout(lockOperationInterval), + // The rebuild starts by closing the previous scope, releasing the + // unresponsive connection back to the pool. Bound it with `withDeadline` + // so a release that never completes cannot leave `lockConnRebuilding` set + // forever, which would disable every subsequent rebuild. + return withDeadline(lockConn.rebuildUnsafe()).pipe( Effect.exit, Effect.tap((exit) => Effect.sync(() => { @@ -598,21 +620,8 @@ export const make = Effect.fnUntraced(function*(options: { `.pipe(execWithLockConnValues, Effect.map((rows) => rows.map((row) => row[0] as string))) }) - const withLockOperationDeadline = Effect.fnUntraced(function*(operation: Effect.Effect) { - // Effect.timeout waits for the timed-out effect to finish interrupting. - // Timeout the join so unresponsive SQL cleanup cannot extend the deadline. - const fiber = yield* Effect.forkIn(operation, layerScope, { startImmediately: true }) - return yield* Fiber.join(fiber).pipe( - Effect.timeout(lockOperationInterval), - Effect.ensuring(Effect.suspend(() => - fiber.pollUnsafe() !== undefined ? Effect.void : Fiber.interrupt(fiber).pipe( - Effect.forkIn(layerScope, { startImmediately: true }), - Effect.asVoid - ) - )), - onErrorRebuildLockConn - ) - }) + const withLockOperationDeadline = (operation: Effect.Effect) => + onErrorRebuildLockConn(withDeadline(operation)) const releaseShard = sql.onDialectOrElse({ pg: () => { diff --git a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts index fe3001589c4..6e6f7c9da13 100644 --- a/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts +++ b/packages/platform-node/test/cluster/SqlRunnerStorage.test.ts @@ -160,6 +160,57 @@ describe("SqlRunnerStorage", () => { ).toEqual(shards) }).pipe(Effect.provide(layer)) }, 60_000) + + it.effect("rebuilds the reserved connection when releasing the previous one hangs", () => { + const partitioned = makePartitionState() + const layer = StorageLive.pipe( + Layer.provideMerge(blackholeReservedConnection(partitioned, false)), + Layer.provide(ShardingConfig.layer({ + shardLockDisableAdvisory: true, + shardLockExpiration: 1000, + shardLockRefreshInterval: 100 + })) + ) + + return Effect.gen(function*() { + const storage = yield* RunnerStorage.RunnerStorage + const runner = Runner.make({ + address: runnerAddress1, + groups: ["default"], + weight: 1 + }) + const shards = [ShardId.make("default", 1)] + + yield* storage.register(runner, true) + yield* storage.acquire(runnerAddress1, shards) + + // the connection wedges and the driver never releases it back to the + // pool, so the rebuild stalls closing the previous scope + partitioned.current = true + partitioned.blockRelease = true + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + yield* Effect.sleep(150).pipe(TestClock.withLive) + + // the stalled release must not disable further rebuilds + partitioned.current = false + const reserved = partitioned.reservedConnections + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + yield* waitUntil(() => partitioned.reservedConnections > reserved) + + expect( + yield* storage.refresh(runnerAddress1, shards).pipe( + Effect.retry({ times: 5, schedule: Schedule.spaced(20) }), + TestClock.withLive + ) + ).toEqual(shards) + }).pipe( + // let the stalled release finish so the layer can be torn down + Effect.ensuring(Effect.sync(() => { + partitioned.blockRelease = false + })), + Effect.provide(layer) + ) + }, 60_000) ;([ ["pg", Layer.orDie(PgContainer.layerClient)], ["mysql", Layer.orDie(MysqlContainer.layerClient)], @@ -238,6 +289,7 @@ const runnerAddress1 = RunnerAddress.make("localhost", 1234) interface PartitionState { current: boolean + blockRelease: boolean activeQueries: number maxActiveQueries: number interruptedQueries: number @@ -248,6 +300,7 @@ interface PartitionState { const makePartitionState = (): PartitionState => ({ current: false, + blockRelease: false, activeQueries: 0, maxActiveQueries: 0, interruptedQueries: 0, @@ -321,10 +374,24 @@ const blackholeReservedConnection = (partitioned: PartitionState, resumePending: client = new Proxy(sql, { get(target, property, receiver) { if (property === "reserve") { - return Effect.map(target.reserve, (connection) => { - partitioned.reservedConnections++ - return wrapConnection(connection) - }) + return Effect.andThen( + // simulates a driver that stalls uninterruptibly while tearing + // down a reserved connection, so closing its scope cannot be + // interrupted + Effect.addFinalizer(() => + Effect.uninterruptible(Effect.suspend(function waitForRelease(): Effect.Effect { + if (!partitioned.blockRelease) return Effect.void + return Effect.andThen( + Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 5))), + waitForRelease + ) + })) + ), + Effect.map(target.reserve, (connection) => { + partitioned.reservedConnections++ + return wrapConnection(connection) + }) + ) } if (property === "withoutTransforms") { return () => client