Open History to see the last commit
1import * as Effect from "effect/Effect";
2import * as Ref from "effect/Ref";
3import * as Semaphore from "effect/Semaphore";
4
5export interface KeyedLock<Key> {
6 readonly withLock: <A, E, R>(key: Key, effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
7 readonly activeKeys: Effect.Effect<ReadonlyArray<Key>>;
8}
9
10interface LockEntry {
11 readonly semaphore: Semaphore.Semaphore;
12 readonly users: number;
13}
14
15/**
16 * Not reentrant: taking a key while already holding it deadlocks.
17 */
18export const make = Effect.fn("KeyedLock.make")(function* <Key>(): Effect.fn.Return<
19 KeyedLock<Key>
20> {
21 const locks = yield* Ref.make<ReadonlyMap<Key, LockEntry>>(new Map());
22
23 const acquire = (key: Key) =>
24 Effect.flatMap(Semaphore.make(1), (candidate) =>
25 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 release = (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 acquire(key),
51 (semaphore) => semaphore.withPermit(effect),
52 () => release(key),
53 ),
54 activeKeys: Effect.map(Ref.get(locks), (current) => Array.from(current.keys())),
55 } satisfies KeyedLock<Key>;
56});