authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-04 21:31:52-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-04 21:31:52-07:00
log37de6cbde65c7cf9ba90a2557c232e63b7e16988
treec51f721ea8412070c800408151542e7939271184
parent1beb0355d05bbe4a780b238e3138f0a8ed99df5f
signature Signed by PGP key B5690EEEBB952194

refactor(server): share one keyed lock that releases idle keys (#15577)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

15 files changed, 227 insertions(+), 233 deletions(-)

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.ts+3-7
......@@ -78,6 +78,7 @@ import * as McpProviderSession from "../../mcp/McpProviderSession.ts";
7878import { buildRuntimeInstructions } from "../../provider/RuntimeInstructions.ts";
7979import { t3OrchestrationSystemPrompt } from "../../provider/T3OrchestrationInstructions.ts";
8080import { SKILL_MENTION_PATTERN } from "@t3tools/shared/composerInlineTokens";
81import * as KeyedLock from "@t3tools/shared/KeyedLock";
8182import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3tools/shared/model";
8283import { causeErrorTag } from "@t3tools/shared/observability";
8384
......@@ -887,18 +888,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr
887888 const stagedReverts = new Set<string>();
888889 // Starting a turn and cutting the history take turns on a session: each
889890 // checks that the other is not running before its own requests yield.
890 const sessionGates = new Map<string, Semaphore.Semaphore>();
891 const sessionGates = yield* KeyedLock.make<string>();
891892 const exclusive =
892893 (providerThread: OrchestrationV2ProviderThread) =>
893894 <A, E, R>(effect: Effect.Effect<A, E, R>) => {
894895 const sessionId = providerThread.nativeThreadRef?.nativeId;
895896 if (sessionId == null) return effect;
896 let gate = sessionGates.get(sessionId);
897 if (gate === undefined) {
898 gate = Semaphore.makeUnsafe(1);
899 sessionGates.set(sessionId, gate);
900 }
901 return gate.withPermit(effect);
897 return sessionGates.withLock(sessionId, effect);
902898 };
903899 const emit = (event: ProviderAdapter.ProviderAdapterV2Event) =>
904900 Queue.offer(events, event).pipe(Effect.asVoid);
apps/server/src/orchestration-v2/CheckpointService.ts+3-24
......@@ -14,10 +14,9 @@ import * as Context from "effect/Context";
1414import * as DateTime from "effect/DateTime";
1515import * as Effect from "effect/Effect";
1616import * as Encoding from "effect/Encoding";
17import * as KeyedLock from "@t3tools/shared/KeyedLock";
1718import * as Layer from "effect/Layer";
18import * as Ref from "effect/Ref";
1919import * as Schema from "effect/Schema";
20import * as Semaphore from "effect/Semaphore";
2120
2221import { parseTurnDiffFilesFromNumstat } from "../checkpointing/Diffs.ts";
2322import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
......@@ -249,29 +248,9 @@ export const layer: Layer.Layer<
249248 Effect.gen(function* () {
250249 const checkpointStore = yield* CheckpointStore.CheckpointStore;
251250 const idAllocator = yield* IdAllocator.IdAllocatorV2;
252 const workspaceSemaphores = yield* Ref.make(new Map<string, Semaphore.Semaphore>());
253
254 const getWorkspaceSemaphore = (cwd: string) =>
255 Effect.gen(function* () {
256 const existing = (yield* Ref.get(workspaceSemaphores)).get(cwd);
257 if (existing !== undefined) {
258 return existing;
259 }
260
261 const created = yield* Semaphore.make(1);
262 return yield* Ref.modify(workspaceSemaphores, (current) => {
263 const concurrent = current.get(cwd);
264 if (concurrent !== undefined) {
265 return [concurrent, current];
266 }
267 const updated = new Map(current);
268 updated.set(cwd, created);
269 return [created, updated];
270 });
271 });
272
251 const workspaceLocks = yield* KeyedLock.make<string>();
273252 const withWorkspaceLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) =>
274 Effect.flatMap(getWorkspaceSemaphore(cwd), (semaphore) => semaphore.withPermits(1)(effect));
253 workspaceLocks.withLock(cwd, effect);
275254
276255 const isGitCheckpointable = (cwd: string) =>
277256 checkpointStore.isGitRepository(cwd).pipe(Effect.orElseSucceed(() => false));
apps/server/src/orchestration-v2/KeyedSerialExecutor.test.ts deleted-67
......@@ -1,67 +0,0 @@
1import { describe, expect, it } from "@effect/vitest";
2
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";
6import * as Queue from "effect/Queue";
7import * as Ref from "effect/Ref";
8
9import { makeKeyedSerialExecutor } from "./KeyedSerialExecutor.ts";
10
11describe("makeKeyedSerialExecutor", () => {
12 it.effect("allows unrelated keys to run concurrently", () =>
13 Effect.gen(function* () {
14 const executor = yield* makeKeyedSerialExecutor<string>();
15 const arrivals = yield* Queue.unbounded<string>();
16 const release = yield* Deferred.make<void>();
17 const run = (key: string) =>
18 executor.withLock(
19 key,
20 Queue.offer(arrivals, key).pipe(Effect.andThen(Deferred.await(release))),
21 );
22
23 const fibers = yield* Effect.forEach(["a", "b"], run, {
24 concurrency: "unbounded",
25 discard: false,
26 }).pipe(Effect.forkChild);
27 const observed = [yield* Queue.take(arrivals), yield* Queue.take(arrivals)];
28 yield* Deferred.succeed(release, undefined);
29 yield* Fiber.join(fibers);
30 expect(new Set(observed)).toEqual(new Set(["a", "b"]));
31 }).pipe(Effect.timeout("1 second")),
32 );
33
34 it.effect("serializes work for the same key", () =>
35 Effect.gen(function* () {
36 const executor = yield* makeKeyedSerialExecutor<string>();
37 const firstEntered = yield* Deferred.make<void>();
38 const releaseFirst = yield* Deferred.make<void>();
39 const events = yield* Ref.make<Array<string>>([]);
40
41 const first = yield* executor
42 .withLock(
43 "thread",
44 Ref.update(events, (current) => [...current, "first:start"]).pipe(
45 Effect.andThen(Deferred.succeed(firstEntered, undefined)),
46 Effect.andThen(Deferred.await(releaseFirst)),
47 Effect.andThen(Ref.update(events, (current) => [...current, "first:end"])),
48 ),
49 )
50 .pipe(Effect.forkChild);
51 yield* Deferred.await(firstEntered);
52 const second = yield* executor
53 .withLock(
54 "thread",
55 Ref.update(events, (current) => [...current, "second:start"]),
56 )
57 .pipe(Effect.forkChild);
58
59 yield* Effect.yieldNow;
60 expect(yield* Ref.get(events)).toEqual(["first:start"]);
61 yield* Deferred.succeed(releaseFirst, undefined);
62 yield* Fiber.join(first);
63 yield* Fiber.join(second);
64 expect(yield* Ref.get(events)).toEqual(["first:start", "first:end", "second:start"]);
65 }).pipe(Effect.timeout("1 second")),
66 );
67});
apps/server/src/orchestration-v2/KeyedSerialExecutor.ts deleted-55
......@@ -1,55 +0,0 @@
1import * as Effect from "effect/Effect";
2import * as Ref from "effect/Ref";
3import * as Semaphore from "effect/Semaphore";
4
5interface LockEntry {
6 readonly semaphore: Semaphore.Semaphore;
7 readonly users: number;
8}
9
10export interface KeyedSerialExecutor<Key> {
11 readonly withLock: <A, E, R>(key: Key, effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
12}
13
14/**
15 * Serializes work that targets the same domain identity without coupling
16 * unrelated identities to a process-wide mutex.
17 */
18export const makeKeyedSerialExecutor = <Key>(): Effect.Effect<KeyedSerialExecutor<Key>> =>
19 Effect.gen(function* () {
20 const locks = yield* Ref.make(new Map<Key, LockEntry>());
21
22 const acquireLock = (key: Key) =>
23 Effect.gen(function* () {
24 const candidate = yield* Semaphore.make(1);
25 return yield* Ref.modify(locks, (current) => {
26 const existing = current.get(key);
27 const semaphore = existing?.semaphore ?? candidate;
28 const next = new Map(current);
29 next.set(key, { semaphore, users: (existing?.users ?? 0) + 1 });
30 return [semaphore, next] as const;
31 });
32 });
33
34 const releaseLock = (key: Key) =>
35 Ref.update(locks, (current) => {
36 const existing = current.get(key);
37 if (existing === undefined) return current;
38 const next = new Map(current);
39 if (existing.users === 1) {
40 next.delete(key);
41 } else {
42 next.set(key, { ...existing, users: existing.users - 1 });
43 }
44 return next;
45 });
46
47 return {
48 withLock: (key, effect) =>
49 Effect.acquireUseRelease(
50 acquireLock(key),
51 (semaphore) => semaphore.withPermit(effect),
52 () => releaseLock(key),
53 ),
54 } satisfies KeyedSerialExecutor<Key>;
55 });
apps/server/src/orchestration-v2/ProviderSessionManager.ts+4-4
......@@ -1,3 +1,4 @@
1import * as KeyedLock from "@t3tools/shared/KeyedLock";
12import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
23import {
34 ModelSelection,
......@@ -36,7 +37,6 @@ import * as ServerSettings from "../serverSettings.ts";
3637import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
3738import * as EventSink from "./EventSink.ts";
3839import * as IdAllocator from "./IdAllocator.ts";
39import { makeKeyedSerialExecutor } from "./KeyedSerialExecutor.ts";
4040import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
4141import {
4242 ProviderAdapterEventStreamError,
......@@ -386,9 +386,9 @@ export const layerWithOptions = (
386386 // cannot drop cleanup for threads only the earlier session served.
387387 const releaseRecordRetries = yield* FiberSet.make();
388388 const nextSubscriberId = yield* Ref.make(0);
389 const sessionOpen = yield* makeKeyedSerialExecutor<ProviderSessionId>();
389 const sessionOpen = yield* KeyedLock.make<ProviderSessionId>();
390390 // Orders a thread's attach against a detach unloading it on the same session.
391 const threadAttachment = yield* makeKeyedSerialExecutor<string>();
391 const threadAttachment = yield* KeyedLock.make<string>();
392392 const threadAttachmentKey = (input: {
393393 readonly providerSessionId: ProviderSessionId;
394394 readonly threadId: ThreadId;
......@@ -427,7 +427,7 @@ export const layerWithOptions = (
427427 };
428428 const isMcpCredentialReserved = (threadId: ThreadId, mcpCredentialId: string) =>
429429 (mcpCredentialReservations.get(mcpReservationKey(threadId, mcpCredentialId)) ?? 0) > 0;
430 const mcpPrepareLock = yield* makeKeyedSerialExecutor<ThreadId>();
430 const mcpPrepareLock = yield* KeyedLock.make<ThreadId>();
431431 /**
432432 * Resolves (or mints) the thread's MCP credential and returns it with a
433433 * reservation held; the caller must drop the reservation exactly once.
apps/server/src/orchestration-v2/ThreadCommandExecutor.ts+3-4
......@@ -1,13 +1,12 @@
11import type { ThreadId } from "@t3tools/contracts";
2import * as KeyedLock from "@t3tools/shared/KeyedLock";
23import * as Context from "effect/Context";
34import * as Layer from "effect/Layer";
45
5import { makeKeyedSerialExecutor, type KeyedSerialExecutor } from "./KeyedSerialExecutor.ts";
6
76/** Shared by thread commands and project deletion so both plan against current thread state. */
87export class ThreadCommandExecutor extends Context.Service<
98 ThreadCommandExecutor,
10 KeyedSerialExecutor<ThreadId>
9 KeyedLock.KeyedLock<ThreadId>
1110>()("t3/orchestration-v2/ThreadCommandExecutor") {}
1211
13export const layer = Layer.effect(ThreadCommandExecutor, makeKeyedSerialExecutor<ThreadId>());
12export const layer = Layer.effect(ThreadCommandExecutor, KeyedLock.make<ThreadId>());
apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.ts+2-2
......@@ -21,6 +21,7 @@ import {
2121 ThreadPullRequestLink,
2222 TurnItemId,
2323} from "@t3tools/contracts";
24import * as KeyedLock from "@t3tools/shared/KeyedLock";
2425import * as Context from "effect/Context";
2526import * as DateTime from "effect/DateTime";
2627import * as Effect from "effect/Effect";
......@@ -30,7 +31,6 @@ import * as Schema from "effect/Schema";
3031import * as SqlClient from "effect/unstable/sql/SqlClient";
3132
3233import * as EventSink from "../EventSink.ts";
33import { makeKeyedSerialExecutor } from "../KeyedSerialExecutor.ts";
3434import { randomUuidV4 } from "../RandomUuid.ts";
3535
3636const IMPORT_EVENT_PREFIX = "migration:v1";
......@@ -347,7 +347,7 @@ function chunks<A>(items: ReadonlyArray<A>, size: number): Array<ReadonlyArray<A
347347const make = Effect.gen(function* () {
348348 const sql = yield* SqlClient.SqlClient;
349349 const eventSink = yield* EventSink.EventSinkV2;
350 const transcriptImports = yield* makeKeyedSerialExecutor<ThreadId>();
350 const transcriptImports = yield* KeyedLock.make<ThreadId>();
351351
352352 const listMessages = (threadId: ThreadId) =>
353353 sql<LegacyMessageRow>`
apps/server/src/project/ProjectService.ts+3-3
......@@ -8,6 +8,7 @@ import {
88 type ProjectSnapshot,
99 type ThreadId,
1010} from "@t3tools/contracts";
11import * as KeyedLock from "@t3tools/shared/KeyedLock";
1112import * as Context from "effect/Context";
1213import * as DateTime from "effect/DateTime";
1314import * as Effect from "effect/Effect";
......@@ -18,7 +19,6 @@ import * as Schema from "effect/Schema";
1819
1920import * as EventSink from "../orchestration-v2/EventSink.ts";
2021import * as IdAllocator from "../orchestration-v2/IdAllocator.ts";
21import { makeKeyedSerialExecutor } from "../orchestration-v2/KeyedSerialExecutor.ts";
2222import * as LegacyV1ThreadImporter from "../orchestration-v2/legacy/LegacyV1ThreadImporter.ts";
2323import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
2424import {
......@@ -155,8 +155,8 @@ export const make = Effect.gen(function* () {
155155 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
156156 // Commands for one project run in order. Commands that claim a workspace root
157157 // also hold that root, so two projects cannot both claim it.
158 const projectLocks = yield* makeKeyedSerialExecutor<ProjectId>();
159 const workspaceLocks = yield* makeKeyedSerialExecutor<string>();
158 const projectLocks = yield* KeyedLock.make<ProjectId>();
159 const workspaceLocks = yield* KeyedLock.make<string>();
160160
161161 const toProject = (
162162 row: ProjectStore.ProjectRow,
apps/server/src/provider/providerMaintenanceCommandCoordinator.ts+6-27
......@@ -1,6 +1,6 @@
1import * as KeyedLock from "@t3tools/shared/KeyedLock";
12import * as Effect from "effect/Effect";
23import * as Ref from "effect/Ref";
3import * as Semaphore from "effect/Semaphore";
44
55export interface ProviderMaintenanceCommandCoordinatorShape<E> {
66 readonly withCommandLock: <A, R>(input: {
......@@ -15,7 +15,7 @@ export const makeProviderMaintenanceCommandCoordinator = Effect.fn(
1515 "makeProviderMaintenanceCommandCoordinator",
1616)(function* <E>(input: { readonly makeAlreadyRunningError: (targetKey: string) => E }) {
1717 const runningTargetsRef = yield* Ref.make<ReadonlySet<string>>(new Set());
18 const locksRef = yield* Ref.make<ReadonlyMap<string, Semaphore.Semaphore>>(new Map());
18 const locks = yield* KeyedLock.make<string>();
1919
2020 const acquireTarget = Effect.fn("acquireTarget")(function* (targetKey: string) {
2121 return yield* Ref.modify(runningTargetsRef, (runningTargets) => {
......@@ -35,24 +35,6 @@ export const makeProviderMaintenanceCommandCoordinator = Effect.fn(
3535 return next;
3636 });
3737
38 const getLock = Effect.fn("getProviderMaintenanceCommandLock")(function* (lockKey: string) {
39 const existing = (yield* Ref.get(locksRef)).get(lockKey);
40 if (existing) {
41 return existing;
42 }
43
44 const lock = yield* Semaphore.make(1);
45 return yield* Ref.modify(locksRef, (locks) => {
46 const current = locks.get(lockKey);
47 if (current) {
48 return [current, locks] as const;
49 }
50 const next = new Map(locks);
51 next.set(lockKey, lock);
52 return [lock, next] as const;
53 });
54 });
55
5638 const withCommandLock: ProviderMaintenanceCommandCoordinatorShape<E>["withCommandLock"] = ({
5739 targetKey,
5840 lockKey,
......@@ -65,13 +47,10 @@ export const makeProviderMaintenanceCommandCoordinator = Effect.fn(
6547 return yield* Effect.fail(input.makeAlreadyRunningError(targetKey));
6648 }
6749
68 return yield* Effect.gen(function* () {
69 const lock = yield* getLock(lockKey);
70 if (onQueued) {
71 yield* onQueued;
72 }
73 return yield* lock.withPermits(1)(run);
74 }).pipe(Effect.ensuring(releaseTarget(targetKey)));
50 return yield* (onQueued ?? Effect.void).pipe(
51 Effect.andThen(locks.withLock(lockKey, run)),
52 Effect.ensuring(releaseTarget(targetKey)),
53 );
7554 });
7655
7756 return {
apps/server/src/terminal/Manager.ts+3-22
......@@ -40,6 +40,7 @@ import {
4040 ProviderInstanceId,
4141} from "@t3tools/contracts";
4242import { makeKeyedCoalescingWorker } from "@t3tools/shared/KeyedCoalescingWorker";
43import * as KeyedLock from "@t3tools/shared/KeyedLock";
4344import { HostProcessArchitecture, HostProcessPlatform } from "@t3tools/shared/hostProcess";
4445import { mergePathEntries } from "@t3tools/shared/shell";
4546
......@@ -58,7 +59,6 @@ import * as Option from "effect/Option";
5859import * as Path from "effect/Path";
5960import * as Schema from "effect/Schema";
6061import * as Scope from "effect/Scope";
61import * as Semaphore from "effect/Semaphore";
6262import * as SynchronizedRef from "effect/SynchronizedRef";
6363
6464import * as ServerConfig from "../config.ts";
......@@ -1569,7 +1569,7 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
15691569 sessions: new Map(),
15701570 killFibers: new Map(),
15711571 });
1572 const threadLocksRef = yield* SynchronizedRef.make(new Map<string, Semaphore.Semaphore>());
1572 const threadLocks = yield* KeyedLock.make<string>();
15731573 const terminalEventListeners = new Set<(event: TerminalEvent) => Effect.Effect<void>>();
15741574 const workerScope = yield* Scope.make("sequential");
15751575 yield* Effect.addFinalizer(() => Scope.close(workerScope, Exit.void));
......@@ -1598,29 +1598,10 @@ export const makeWithOptions = Effect.fn("TerminalManager.makeWithOptions")(func
15981598 f: (state: TerminalManagerState) => readonly [A, TerminalManagerState],
15991599 ) => SynchronizedRef.modify(managerStateRef, f);
16001600
1601 const getThreadSemaphore = (threadId: string) =>
1602 SynchronizedRef.modifyEffect(threadLocksRef, (current) => {
1603 const existing: Option.Option<Semaphore.Semaphore> = Option.fromNullishOr(
1604 current.get(threadId),
1605 );
1606 return Option.match(existing, {
1607 onNone: () =>
1608 Semaphore.make(1).pipe(
1609 Effect.map((semaphore) => {
1610 const next = new Map(current);
1611 next.set(threadId, semaphore);
1612 return [semaphore, next] as const;
1613 }),
1614 ),
1615 onSome: (semaphore) => Effect.succeed([semaphore, current] as const),
1616 });
1617 });
1618
16191601 const withThreadLock = <A, E, R>(
16201602 threadId: string,
16211603 effect: Effect.Effect<A, E, R>,
1622 ): Effect.Effect<A, E, R> =>
1623 Effect.flatMap(getThreadSemaphore(threadId), (semaphore) => semaphore.withPermit(effect));
1604 ): Effect.Effect<A, E, R> => threadLocks.withLock(threadId, effect);
16241605
16251606 const clearKillFiber = Effect.fn("terminal.clearKillFiber")(function* (
16261607 process: PtyAdapter.PtyProcess | null,
apps/server/src/vcs/VcsStatusBroadcaster.ts+4-10
......@@ -10,7 +10,6 @@ import * as PubSub from "effect/PubSub";
1010import * as Ref from "effect/Ref";
1111import * as Schedule from "effect/Schedule";
1212import * as Scope from "effect/Scope";
13import * as Semaphore from "effect/Semaphore";
1413import * as Stream from "effect/Stream";
1514import * as SynchronizedRef from "effect/SynchronizedRef";
1615import type {
......@@ -22,6 +21,7 @@ import type {
2221 VcsStatusStreamEvent,
2322} from "@t3tools/contracts";
2423import { mergeGitStatusParts } from "@t3tools/shared/git";
24import * as KeyedLock from "@t3tools/shared/KeyedLock";
2525import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
2626
2727import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
......@@ -234,15 +234,9 @@ export const make = Effect.gen(function* () {
234234 // One permit per cwd for remote reads that write the cache. Without it a
235235 // periodic poll that started before `gh pr create` can finish after the
236236 // turn-end refresh and overwrite the fresh PR with its stale `pr: null`.
237 const remoteWriteLocks = new Map<string, Semaphore.Semaphore>();
238 const withRemoteWriteLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) => {
239 let lock = remoteWriteLocks.get(cwd);
240 if (lock === undefined) {
241 lock = Semaphore.makeUnsafe(1);
242 remoteWriteLocks.set(cwd, lock);
243 }
244 return lock.withPermits(1)(effect);
245 };
237 const remoteWriteLocks = yield* KeyedLock.make<string>();
238 const withRemoteWriteLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) =>
239 remoteWriteLocks.withLock(cwd, effect);
246240 const pollersRef = yield* SynchronizedRef.make(new Map<string, ActiveRemotePoller>());
247241
248242 const getCachedStatus = Effect.fn("VcsStatusBroadcaster.getCachedStatus")(function* (
packages/shared/package.json+4
......@@ -95,6 +95,10 @@
9595 "types": "./src/KeyedCoalescingWorker.ts",
9696 "import": "./src/KeyedCoalescingWorker.ts"
9797 },
98 "./KeyedLock": {
99 "types": "./src/KeyedLock.ts",
100 "import": "./src/KeyedLock.ts"
101 },
98102 "./schemaJson": {
99103 "types": "./src/schemaJson.ts",
100104 "import": "./src/schemaJson.ts"
packages/shared/src/KeyedLock.test.ts created+127
......@@ -0,0 +1,127 @@
1import { assert, describe, it } from "@effect/vitest";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Exit from "effect/Exit";
5import * as Fiber from "effect/Fiber";
6import * as Queue from "effect/Queue";
7import * as Ref from "effect/Ref";
8import * as Scope from "effect/Scope";
9
10import * as KeyedLock from "./KeyedLock.ts";
11
12describe("KeyedLock", () => {
13 it.effect("runs one holder of a key at a time, waiters in the order they queued", () =>
14 Effect.gen(function* () {
15 const lock = yield* KeyedLock.make<string>();
16 const events = yield* Ref.make<ReadonlyArray<string>>([]);
17 const record = (event: string) => Ref.update(events, (current) => [...current, event]);
18 const firstEntered = yield* Deferred.make<void>();
19 const releaseFirst = yield* Deferred.make<void>();
20
21 const first = yield* lock
22 .withLock(
23 "key",
24 record("first:start").pipe(
25 Effect.andThen(Deferred.succeed(firstEntered, undefined)),
26 Effect.andThen(Deferred.await(releaseFirst)),
27 Effect.andThen(record("first:end")),
28 ),
29 )
30 .pipe(Effect.forkChild);
31 yield* Deferred.await(firstEntered);
32 const second = yield* lock.withLock("key", record("second")).pipe(Effect.forkChild);
33 const third = yield* lock.withLock("key", record("third")).pipe(Effect.forkChild);
34 yield* Effect.yieldNow;
35 assert.deepStrictEqual(yield* Ref.get(events), ["first:start"]);
36
37 yield* Deferred.succeed(releaseFirst, undefined);
38 yield* Fiber.joinAll([first, second, third]);
39 assert.deepStrictEqual(yield* Ref.get(events), [
40 "first:start",
41 "first:end",
42 "second",
43 "third",
44 ]);
45 }),
46 );
47
48 it.effect("runs holders of different keys concurrently", () =>
49 Effect.gen(function* () {
50 const lock = yield* KeyedLock.make<string>();
51 const entered = yield* Queue.unbounded<string>();
52 const release = yield* Deferred.make<void>();
53 const hold = (key: string) =>
54 lock.withLock(key, Queue.offer(entered, key).pipe(Effect.andThen(Deferred.await(release))));
55
56 const holders = yield* Effect.forEach(["a", "b"], hold, { concurrency: "unbounded" }).pipe(
57 Effect.forkChild,
58 );
59 // Both enter while the other still holds its key.
60 const both = new Set([yield* Queue.take(entered), yield* Queue.take(entered)]);
61 assert.deepStrictEqual(both, new Set(["a", "b"]));
62 yield* Deferred.succeed(release, undefined);
63 yield* Fiber.join(holders);
64 }),
65 );
66
67 it.effect("keeps no lock for a key once its holders and waiters are gone", () =>
68 Effect.gen(function* () {
69 const lock = yield* KeyedLock.make<string>();
70 const entered = yield* Deferred.make<void>();
71 const release = yield* Deferred.make<void>();
72
73 const holder = yield* lock
74 .withLock(
75 "key",
76 Deferred.succeed(entered, undefined).pipe(Effect.andThen(Deferred.await(release))),
77 )
78 .pipe(Effect.forkChild);
79 yield* Deferred.await(entered);
80 // A waiter that gives up and one that runs.
81 const abandoned = yield* lock.withLock("key", Effect.void).pipe(Effect.forkChild);
82 const waiting = yield* lock.withLock("key", Effect.void).pipe(Effect.forkChild);
83 yield* Effect.yieldNow;
84 assert.deepStrictEqual(yield* lock.activeKeys, ["key"]);
85
86 yield* Fiber.interrupt(abandoned);
87 // The holder and the other waiter still need the lock.
88 assert.deepStrictEqual(yield* lock.activeKeys, ["key"]);
89 yield* Deferred.succeed(release, undefined);
90 yield* Fiber.joinAll([holder, waiting]);
91 // A holder that fails releases its key too.
92 assert.strictEqual(
93 yield* lock.withLock("other", Effect.fail("boom")).pipe(Effect.flip),
94 "boom",
95 );
96
97 assert.deepStrictEqual(yield* lock.activeKeys, []);
98 }),
99 );
100
101 it.effect("keeps working after the scope that made it closes", () =>
102 Effect.gen(function* () {
103 // Services built per request hand their lock to work that outlives the build.
104 const lock = yield* Effect.scoped(KeyedLock.make<string>());
105 assert.strictEqual(yield* lock.withLock("key", Effect.succeed("ran")), "ran");
106 }),
107 );
108
109 it.effect("runs the effect in the caller's scope", () =>
110 Effect.gen(function* () {
111 const lock = yield* KeyedLock.make<string>();
112 const finalized = yield* Ref.make(false);
113 const callerScope = yield* Scope.make();
114
115 yield* lock
116 .withLock(
117 "key",
118 Effect.addFinalizer(() => Ref.set(finalized, true)),
119 )
120 .pipe(Scope.provide(callerScope));
121 // The finalizer belongs to the caller, so releasing the lock leaves it.
122 assert.isFalse(yield* Ref.get(finalized));
123 yield* Scope.close(callerScope, Exit.void);
124 assert.isTrue(yield* Ref.get(finalized));
125 }),
126 );
127});
packages/shared/src/KeyedLock.ts created+62
......@@ -0,0 +1,62 @@
1import * as Effect from "effect/Effect";
2import * as Ref from "effect/Ref";
3import * as Semaphore from "effect/Semaphore";
4
5export interface KeyedLock<Key> {
6 /** Runs `effect` once no other holder of `key` is running. Queued waiters go first-in, first-out. */
7 readonly withLock: <A, E, R>(key: Key, effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
8 /** The keys someone holds or waits on right now. */
9 readonly activeKeys: Effect.Effect<ReadonlyArray<Key>>;
10}
11
12interface LockEntry {
13 readonly semaphore: Semaphore.Semaphore;
14 readonly users: number;
15}
16
17/**
18 * Mutual exclusion per key. A key's lock exists only while someone holds or
19 * waits on it, so locks for keys that come and go never accumulate. Not
20 * reentrant: taking a key while already holding it deadlocks.
21 *
22 * Keys compare like `Map` keys (by value for strings and numbers). The lock is
23 * not tied to a scope, so it keeps working for as long as anyone references it.
24 */
25export const make = <Key>(): Effect.Effect<KeyedLock<Key>> =>
26 Effect.gen(function* () {
27 const locks = yield* Ref.make<ReadonlyMap<Key, LockEntry>>(new Map());
28
29 const acquire = (key: Key) =>
30 Effect.flatMap(Semaphore.make(1), (candidate) =>
31 Ref.modify(locks, (current) => {
32 const existing = current.get(key);
33 const semaphore = existing?.semaphore ?? candidate;
34 const next = new Map(current);
35 next.set(key, { semaphore, users: (existing?.users ?? 0) + 1 });
36 return [semaphore, next] as const;
37 }),
38 );
39
40 const release = (key: Key) =>
41 Ref.update(locks, (current) => {
42 const existing = current.get(key);
43 if (existing === undefined) return current;
44 const next = new Map(current);
45 if (existing.users === 1) {
46 next.delete(key);
47 } else {
48 next.set(key, { ...existing, users: existing.users - 1 });
49 }
50 return next;
51 });
52
53 return {
54 withLock: (key, effect) =>
55 Effect.acquireUseRelease(
56 acquire(key),
57 (semaphore) => semaphore.withPermit(effect),
58 () => release(key),
59 ),
60 activeKeys: Effect.map(Ref.get(locks), (current) => Array.from(current.keys())),
61 } satisfies KeyedLock<Key>;
62 });
packages/ssh/src/tunnel.ts+3-8
......@@ -7,6 +7,7 @@ import {
77 waitForHttpReady as waitForHttpReadyShared,
88} from "@t3tools/shared/httpReadiness";
99import { cliReleaseDownloadBaseUrl } from "@t3tools/shared/cliRelease";
10import * as KeyedLock from "@t3tools/shared/KeyedLock";
1011import * as NetService from "@t3tools/shared/Net";
1112import { extractJsonObject, fromLenientJson } from "@t3tools/shared/schemaJson";
1213import { satisfiesSemverRange } from "@t3tools/shared/semver";
......@@ -18,7 +19,6 @@ import * as Layer from "effect/Layer";
1819import * as Path from "effect/Path";
1920import * as Schema from "effect/Schema";
2021import * as Scope from "effect/Scope";
21import * as Semaphore from "effect/Semaphore";
2222import * as Stream from "effect/Stream";
2323import { HttpClient } from "effect/unstable/http";
2424import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
......@@ -1328,7 +1328,7 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
13281328): Effect.fn.Return<SshEnvironmentManagerShape, never, Scope.Scope> {
13291329 const managerScope = yield* Scope.Scope;
13301330 const tunnels = new Map<string, SshTunnelEntry>();
1331 const targetLocks = new Map<string, Semaphore.Semaphore>();
1331 const targetLocks = yield* KeyedLock.make<string>();
13321332 const authSecrets = new Map<string, string>();
13331333
13341334 // Keep one lock per target so reconnect cannot reuse a server while stop is pending.
......@@ -1336,12 +1336,7 @@ const makeSshEnvironmentManager = Effect.fn("ssh/tunnel.SshEnvironmentManager.ma
13361336 key: string,
13371337 effect: Effect.Effect<A, E, R>,
13381338 ): Effect.fn.Return<A, E, R> {
1339 let lock = targetLocks.get(key);
1340 if (lock === undefined) {
1341 lock = Semaphore.makeUnsafe(1);
1342 targetLocks.set(key, lock);
1343 }
1344 return yield* lock.withPermits(1)(effect);
1339 return yield* targetLocks.withLock(key, effect);
13451340 });
13461341
13471342 const closeTunnelEntry = Effect.fn("ssh/tunnel.closeTunnelEntry")(function* (