diff --git a/.changeset/slow-spiders-refresh.md b/.changeset/slow-spiders-refresh.md new file mode 100644 index 00000000000..ef55e25e2f1 --- /dev/null +++ b/.changeset/slow-spiders-refresh.md @@ -0,0 +1,14 @@ +--- +"effect": patch +--- + +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, 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. Rebuilds are +themselves bounded, so a connection release that never completes cannot block +every later rebuild. diff --git a/packages/effect/src/unstable/cluster/Sharding.ts b/packages/effect/src/unstable/cluster/Sharding.ts index a202047cf87..f6ecd0907f2 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() + // 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 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,37 +293,88 @@ const make = Effect.gen(function*() { }) const releaseShardsMap = yield* FiberMap.make() - const releaseShard = Effect.fnUntraced( - function*(shardId: ShardId) { - const fibers = Arr.empty>() + let forcedShardReleaseRunning = false + // 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))) + fibers.push(yield* Effect.forkScoped(state.manager.interruptShard(shardId, { force }))) } - yield* Fiber.joinAll(fibers) - 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 }) ) ) + // 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 + } + forcedShardReleaseRunning = true + const shardIds = [...forceReleasingShards] + return runShardRelease(shardIds, true, runnerStorage.releaseAll(selfAddress)).pipe( + retryShardRelease({ fiber: "releaseForcedShards" }), + 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) } @@ -336,11 +394,21 @@ 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 } + if (!shardLocksHealthyLatch.isOpen()) { + 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) { @@ -353,7 +421,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 +431,19 @@ const make = Effect.gen(function*() { const acquired = oacquired.value yield* storage.resetShards(acquired).pipe( Effect.ignore, - Effect.timeoutOption(config.shardLockRefreshInterval) + 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 (MutableHashSet.has(releasingShards, shardId) || !MutableHashSet.has(selfShards, shardId)) { + if ( + !shardLocksHealthyLatch.isOpen() || + forcedReleasePending || + MutableHashSet.has(releasingShards, shardId) || + !MutableHashSet.has(selfShards, shardId) + ) { + MutableHashSet.add(releasingShards, shardId) continue } MutableHashSet.add(acquiredShards, shardId) @@ -392,8 +469,52 @@ 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 (!shardLocksHealthyLatch.closeUnsafe()) return Effect.void + + const affectedShards = MutableHashSet.fromIterable([...acquiredShards, ...releasingShards]) + 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 (!shardLocksHealthyLatch.openUnsafe()) return Effect.void + + 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 +542,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(() => shardLocksHealthyLatch.isOpen() ? refreshShardLocks : probeShardLocks).pipe( + Effect.repeat(Schedule.fixed(shardLockInterval)), Effect.forever, Effect.forkIn(shardingScope) ) @@ -439,11 +568,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 +1101,7 @@ const make = Effect.gen(function*() { if (newAssignments) { const runner = newAssignments[i] MutableHashMap.set(shardAssignments, shard, runner) - if (isLocalRunner(runner)) { + if (shardLocksHealthyLatch.isOpen() && 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: () => @@ -79,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 @@ -91,12 +108,79 @@ 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)) + + // `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 + // another rebuild. + let lockConnGeneration = 0 + const rebuildLockConn = (generation: number) => { + if ( + !lockConn || + lockConnRebuilding || + generation !== lockConnGeneration || + lockConn.state.current._tag === "Closed" + ) return Effect.void + lockConnRebuilding = true + // 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(() => { + if (Exit.isSuccess(exit) && lockConn.state.current._tag === "Acquired") { + lockConnGeneration++ + } + }) + ), + Effect.ensuring(Effect.sync(() => { + lockConnRebuilding = false + })), + Effect.forkIn(layerScope, { startImmediately: true }), + 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) @@ -276,7 +360,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()) + onErrorRebuildLockConn ) } const execWithLockConnUnprepared = ( @@ -286,7 +370,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()) + onErrorRebuildLockConn ) } const execWithLockConnValues = ( @@ -296,7 +380,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()) + onErrorRebuildLockConn ) } @@ -314,6 +398,7 @@ export const make = Effect.fnUntraced(function*(options: { WHERE ${locksTableSql}.address = ${address} OR ${locksTableSql}.acquired_at < ${lockExpiresAt} `.pipe( + execWithLockConn, Effect.andThen(acquiredLocks(address, shardIds)) ) } @@ -342,7 +427,7 @@ export const make = Effect.fnUntraced(function*(options: { } } return acquiredShardIds - }, Effect.onError(() => lockConn!.rebuildUnsafe())) + }, onErrorRebuildLockConn) }, mysql: () => { @@ -356,7 +441,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)) ) } @@ -385,7 +471,7 @@ export const make = Effect.fnUntraced(function*(options: { } } return acquiredShardIds - }, Effect.onError(() => lockConn!.rebuildUnsafe())) + }, onErrorRebuildLockConn) }, mssql: () => (address: string, shardIds: ReadonlyArray) => { @@ -477,7 +563,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)) ) @@ -533,6 +620,72 @@ export const make = Effect.fnUntraced(function*(options: { `.pipe(execWithLockConnValues, Effect.map((rows) => rows.map((row) => row[0] as string))) }) + const withLockOperationDeadline = (operation: Effect.Effect) => + onErrorRebuildLockConn(withDeadline(operation)) + + 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()`, []) + }, + onErrorRebuildLockConn, + 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 + } + }, + onErrorRebuildLockConn, + 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, @@ -563,120 +716,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([]), + 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..e177e099a97 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, + ShardId, Sharding, ShardingConfig, Snowflake @@ -583,6 +588,331 @@ describe.concurrent("Sharding", () => { })) }) +describe("Sharding shard lock failover", () => { + it.effect("interrupts entities and reacquires shards after lock storage recovers", () => + Effect.gen(function*() { + const storageState = makeFailoverStorageState() + 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)!.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) + })) + + it.effect("keeps the graceful timeout for normal shard reassignment", () => + Effect.gen(function*() { + const storageState = makeFailoverStorageState() + 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) + })) + + 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 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, false], [otherRunner, true]] + return state.otherRunnerHealthy ? [[state.runner, true], [otherRunner, true]] : [[state.runner, 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, + completedReleaseAlls: state.releaseAllCalls.filter((call) => call.completed).length + }) + 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.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({ + 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 913899875e5..6e6f7c9da13 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 { Duration, Effect, Exit, FileSystem, Layer, Schedule } from "effect" +import { TestClock } from "effect/testing" import { Runner, RunnerAddress, @@ -10,12 +11,206 @@ import { ShardingConfig, SqlRunnerStorage } from "effect/unstable/cluster" +import { SqlClient, type SqlConnection, SqlError } 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("bounds shard lock operations and rebuilds an unresponsive reserved connection", () => { + const partitioned = makePartitionState() + const layer = StorageLive.pipe( + Layer.provideMerge(blackholeReservedConnection(partitioned, true)), + 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) + partitioned.current = true + + 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* 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 = 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) + partitioned.current = true + yield* storage.refresh(runnerAddress1, shards).pipe(Effect.exit, TestClock.withLive) + yield* Effect.sleep(150).pipe(TestClock.withLive) + + 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) + + 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) + + 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)], @@ -92,6 +287,122 @@ describe("SqlRunnerStorage", () => { const runnerAddress1 = RunnerAddress.make("localhost", 1234) +interface PartitionState { + current: boolean + blockRelease: boolean + activeQueries: number + maxActiveQueries: number + interruptedQueries: number + failNextQueries: number + reservedConnections: number + usableConnections: number +} + +const makePartitionState = (): PartitionState => ({ + current: false, + blockRelease: 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 => { + 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 { + 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++ + } + if (Exit.isSuccess(exit) && !usable) { + usable = true + partitioned.usableConnections++ + } + }) + ) + ) + }) + 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.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 + } + 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()