keyed-mutex.ts 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445
  1. export * as KeyedMutex from "./keyed-mutex"
  2. import { Effect, Semaphore } from "effect"
  3. export interface KeyedMutex<in Key> {
  4. readonly size: Effect.Effect<number>
  5. readonly withLock: (key: Key) => <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
  6. }
  7. /**
  8. * Creates an in-memory mutex with one lock per key. Entries are removed when no
  9. * holder or waiter remains.
  10. *
  11. * same key -> queue
  12. * different key -> run independently
  13. *
  14. * `users` counts holders and waiters so an entry is not removed while a waiter
  15. * will reuse it.
  16. */
  17. export const makeUnsafe = <Key>(): KeyedMutex<Key> => {
  18. const locks = new Map<Key, { readonly semaphore: Semaphore.Semaphore; users: number }>()
  19. const withLock =
  20. (key: Key) =>
  21. <A, E, R>(effect: Effect.Effect<A, E, R>) =>
  22. Effect.suspend(() => {
  23. const current = locks.get(key)
  24. const entry = current ?? { semaphore: Semaphore.makeUnsafe(1), users: 0 }
  25. if (!current) locks.set(key, entry)
  26. entry.users++
  27. return entry.semaphore.withPermit(effect).pipe(
  28. Effect.ensuring(
  29. Effect.sync(() => {
  30. entry.users--
  31. if (entry.users === 0) locks.delete(key)
  32. }),
  33. ),
  34. )
  35. })
  36. return { size: Effect.sync(() => locks.size), withLock }
  37. }
  38. /** Creates an in-memory keyed mutex inside an Effect workflow. */
  39. export const make = <Key>(): Effect.Effect<KeyedMutex<Key>> => Effect.sync(makeUnsafe<Key>)