DurabilitySemaphore | Durability

Import DurabilitySemaphore from durability/experimental/semaphore and pin an exact version. This is the holder-side authority for a shared concurrency budget; the queue is a separate primitive.

How it uses your Durable Object

DurabilitySemaphore runs inside the holder object you define, one per semaphore key, and keeps the whole budget there:

ctx.storage.sql holds durability_semaphore_grants and durability_semaphore_leases. Every round applies reports and writes new grants and leases in one storage transaction, before any client hears back.
The object’s one alarm follows the earliest claim deadline of a grant with unclaimed permits, so unclaimed capacity comes back without another request. Your class’s alarm() must delegate to the scheduler. Lease expiry is checked when rounds and reports run, not on the alarm.
The live instance holds the requests waiting for a round, the round timer, and whatever your limit and groupLimit callbacks read. Eviction fails the waiting RPC calls; clients ask again.
RPC with client objects goes through DurabilitySemaphore.client(), which reaches the holder with getByName(key) and calls the method you choose. Clients keep their own copy of grants and leases and report cumulative counts.

1/9 · The limit lives in an object you own

The semaphore attaches to a DurabilityScheduler built on this.ctx, so it uses this object’s storage and alarm. limit() and groupLimit() are your callbacks; the semaphore evaluates them once per round.

Open source ↗

import { DurableObject } from 'cloudflare:workers';
import { timeout } from '@durability/transforms';
import { Durability, DurabilityScheduler, type DurableCall } from 'durability';
import {
  DurabilitySemaphore,
  DurabilitySemaphoreQueue,
  type AdmittedItem,
  type SemaphoreRequest,
} from 'durability/experimental/semaphore';

type Env = {
  JOB_SHARDS: DurableObjectNamespace<JobShard>;
  JOB_API: Fetcher;
  /** Running jobs allowed per account. */
  ACCOUNT_CONCURRENCY: string;
};

type Job = {
  id: string;
  accountId: string;
  /** Jobs of one workflow may carry their own ceiling under the account's. */
  workflowId: string;
  payload: unknown;
};

/**
 * Holds one account's limit; the account ID is the semaphore key. Leases let
 * shards admit bursts without a round trip, and `strict` keeps grants plus
 * lease footprint under the limit at all times.
 */
export class AccountLimit extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
  private readonly workflowLimits = new Map<string, number>();
  private readonly semaphore = new DurabilitySemaphore({
    scheduler: this.scheduler,
    limit: () => Number(this.env.ACCOUNT_CONCURRENCY),
    groupLimit: (workflowId) => this.workflowLimits.get(workflowId),
    leases: {
      provisioning: { mode: 'strict' },
      permitsPerLease: 16,
      ttlMs: 10 * 60_000,
    },
  });

  permits(message: SemaphoreRequest) {
    return this.semaphore.handle(message);
  }

  /** Administrative: cap one workflow. Takes effect at the next round. */
  limitWorkflow(workflowId: string, limit: number) {
    this.workflowLimits.set(workflowId, limit);
  }

  usage() {
    return this.semaphore.usage();
  }

  override alarm(info?: AlarmInvocationInfo) {
    return this.scheduler.alarm(info);
  }
}

export const limits = DurabilitySemaphore.client({
  target: AccountLimit,
  // Vite injects exportName from this application class and Wrangler's config.
  invoke: (stub, message) => stub.permits(message),
});

/**
 * Many shards, each queueing jobs for many accounts. Admitted jobs run as
 * durable operations; the permit returns when the operation reaches a
 * terminal state, success or failure, through the lifecycle hook or, if that
 * is lost, the queue's `finished` check.
 */
export class JobShard extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
  private readonly jobs = new Durability({
    scheduler: this.scheduler,
    handlers: {
      run: async ({ id, payload, signal }: DurableCall<Job>) => {
        const response = await this.env.JOB_API.fetch('https://job-api/run', {
          method: 'POST',
          headers: { 'Idempotency-Key': id },
          body: JSON.stringify(payload.payload),
          signal,
        });
        if (!response.ok) {
          throw new Error(`Job API responded ${response.status}`);
        }
      },
    },
    retries: { maxAttempts: 3 },
    onLifecycleEvent: (event): Promise<void> | undefined => {
      const finished =
        event.type === 'terminal' ||
        (event.type === 'attempt_settled' && event.outcome === 'completed');
      if (finished && 'id' in event) {
        return this.permits.release(event.id).then(() => undefined);
      }
      return undefined;
    },
  });
  private readonly permits = new DurabilitySemaphoreQueue<Job>({
    scheduler: this.scheduler,
    clientId: this.ctx.id.toString(),
    server: limits.with(timeout, 5_000),
    // Operation IDs equal item IDs, so a repeated admit() is a no-op.
    admit: (items: AdmittedItem<Job>[]): Promise<void> =>
      Promise.all(
        items.map((item) => this.jobs.run({ id: item.id, payload: item.body }))
      ).then(() => undefined),
    // The lifecycle hook above is best-effort; this returns any permit it missed.
    finished: async (ids) => {
      const results = await Promise.all(
        ids.map((id) => this.jobs.run.getResult(id))
      );
      return ids.filter((_, index) => results[index]?.status !== 'pending');
    },
  });

  submit(jobs: Job[]) {
    return this.permits.enqueue(
      jobs.map((job) => ({
        id: job.id,
        key: job.accountId,
        group: job.workflowId,
        body: job,
      }))
    );
  }

  status(id: string) {
    return this.permits.status(id);
  }

  override alarm(info?: AlarmInvocationInfo) {
    return this.scheduler.alarm(info);
  }
}

/** Spread one account's jobs over a few shards; any stable function works. */
const shardFor = (accountId: string, width = 4): string => {
  let hash = 0;
  for (const char of accountId) {
    hash = (hash * 31 + char.charCodeAt(0)) >>> 0;
  }
  return `${accountId}-${hash % width}`;
};

export default {
  async fetch(request, env): Promise<Response> {
    if (request.method !== 'POST') {
      return new Response('POST a job array', { status: 405 });
    }
    // Demo input only: authenticate and validate untrusted payloads in an application.
    const jobs = await request.json<Job[]>();
    const byShard = new Map<string, Job[]>();
    for (const job of jobs) {
      const shard = shardFor(job.accountId);
      byShard.set(shard, [...(byShard.get(shard) ?? []), job]);
    }
    const results = await Promise.all(
      [...byShard].map(([shard, shardJobs]) =>
        env.JOB_SHARDS.getByName(shard).submit(shardJobs)
      )
    );
    return Response.json(results, { status: 202 });
  },
} satisfies ExportedHandler<Env>;

/**
 * Experimental: this entry point may change or be removed in a minor
 * release. Pin an exact version of `durability` when depending on it.
 *
 * @packageDocumentation
 */
import {
  createTransformStub,
  type TransformContext,
  type TransformStub,
} from '@durability/transforms';
import {
  resolveNamespace,
  type DurableObjectTarget,
} from './durable-object-target.js';
import { SemaphoreProtocolError } from './errors.js';
import { SemaphoreClient } from './semaphore-client.js';
import {
  type PermitGrant,
  type PermitRequest,
  type PermitSync,
  type PermitSyncResult,
  type SemaphoreRequest,
  type SemaphoreResponse,
} from './semaphore-protocol.js';
import { DurabilitySemaphoreServer } from './semaphore-server.js';

export { SemaphoreProtocolError } from './errors.js';
export {
  durabilitySemaphoreMigrations,
  durabilitySemaphoreQueueMigrations,
} from './migrations.js';

export type {
  GrantReport,
  GrantResult,
  PermitGrant,
  PermitRequest,
  PermitSync,
  PermitSyncResult,
  SemaphoreBlockedReason,
  SemaphoreRequest,
  SemaphoreResponse,
} from './semaphore-protocol.js';
export type {
  SemaphoreAddress,
  SemaphoreClientConfig,
} from './semaphore-client.js';
export { SemaphoreClient } from './semaphore-client.js';
export type {
  DurabilitySemaphoreConfig,
  DurabilitySemaphoreEvent,
  SemaphoreUsage,
} from './semaphore-server.js';
export type {
  SemaphoreLeaseOptions,
  SemaphoreProvisioning,
} from './semaphore-leases.js';
export {
  DurabilitySemaphoreQueue,
  type AdmittedItem,
  type DurabilitySemaphoreQueueConfig,
  type DurabilitySemaphoreQueueEvent,
  type SemaphoreHolder,
  type SemaphoreInput,
  type SemaphoreItemStatus,
} from './semaphore-queue.js';

export type SemaphoreSession = {
  request(key: string, message: PermitRequest): Promise<PermitGrant>;
  sync(key: string, message: PermitSync): Promise<PermitSyncResult>;
  setContext(context: TransformContext): {
    invoke(method: string, args: unknown[]): Promise<unknown>;
  };
};

export type DurabilitySemaphoreClientConfig<
  Target extends Rpc.DurableObjectBranded,
> = DurableObjectTarget<Target> & {
  /** Chooses the application RPC method that forwards to `handle`; never hardcoded. */
  invoke(
    stub: DurableObjectStub<Target>,
    message: SemaphoreRequest
  ): Promise<SemaphoreResponse>;
  locationHint?: (key: string) => DurableObjectLocationHint | undefined;
};

/**
 * Inside a DO, holds one concurrency limit for many client Durable Objects.
 * Outside the DO, `client()` routes semaphore messages to the application-owned
 * holder class by key using ordinary @durability/transforms caller transforms.
 */
export class DurabilitySemaphore extends DurabilitySemaphoreServer {
  /**
   * Routes semaphore messages to the application-owned holder class by key.
   * Caller transforms such as `timeout` apply; transform context is not used.
   */
  static client<Target extends Rpc.DurableObjectBranded>(
    config: DurabilitySemaphoreClientConfig<Target>
  ): TransformStub<SemaphoreSession> {
    const client = new SemaphoreClient({
      invoke: (address, message) => {
        const namespace = resolveNamespace(
          config,
          (text) => new SemaphoreProtocolError(text)
        );
        const stub = namespace.getByName(
          address.key,
          address.locationHint === undefined
            ? undefined
            : { locationHint: address.locationHint }
        );
        return config.invoke(stub, message);
      },
      ...(config.locationHint === undefined
        ? {}
        : { locationHint: config.locationHint }),
    });

    const session: SemaphoreSession = {
      request: (key, message) => client.request(key, message),
      sync: (key, message) => client.sync(key, message),
      setContext: () => ({
        invoke: (method, args) => {
          const key = args[0] as string;
          if (method === 'request') {
            return client.request(key, args[1] as PermitRequest);
          }
          if (method === 'sync') {
            return client.sync(key, args[1] as PermitSync);
          }
          throw new SemaphoreProtocolError(
            `Unknown semaphore method "${method}"`
          );
        },
      }),
    };
    return createTransformStub(session);
  }
}

import { traceLifecycle, traceMethods } from './dev.js';
import { SemaphoreProtocolError, reportFailure } from './errors.js';
import {
  durabilitySemaphoreMigrations,
  migrateCapability,
  type DurabilityMigrationResult,
} from './migrations.js';
import { assertPositiveInteger } from './policy.js';
import {
  allocateLeases,
  applyLeaseReports,
  budgets,
  leaseFootprint,
  resolveLeasePolicy,
  type LeaseCandidate,
  type LeaseEvent,
  type LeasePolicy,
  type SemaphoreLeaseOptions,
} from './semaphore-leases.js';
import {
  engineFor,
  type Engine,
  type SchedulerAttachment,
} from './scheduler.js';
import {
  semaphoreRequestSchema,
  type GrantReport,
  type GrantResult,
  type LeaseGrant,
  type PermitGrant,
  type PermitRequest,
  type PermitSync,
  type PermitSyncResult,
  type SemaphoreBlockedReason,
  type SemaphoreRequest,
  type SemaphoreResponse,
} from './semaphore-protocol.js';
import { HolderStore, type GrantRow } from './semaphore-store.js';
import type { DurableTable } from './storage.js';

const grantsTable: DurableTable = {
  name: 'durability_semaphore_grants',
  key: 'grant_id',
};

export type DurabilitySemaphoreEvent =
  | {
      entityKind: 'semaphore';
      type: 'round';
      timestamp: number;
      /** Permits asked for and handed out in this round. */
      requested: number;
      granted: number;
      usage: SemaphoreUsage;
    }
  | {
      entityKind: 'semaphore';
      type: 'blocked';
      timestamp: number;
      group: string;
      reason: SemaphoreBlockedReason;
      /** Permits that stayed queued behind the limit. */
      count: number;
    }
  | LeaseEvent
  | {
      entityKind: 'semaphore';
      type: 'grant_reclaimed';
      timestamp: number;
      grantId: string;
      clientId: string;
      group: string;
      /** Unclaimed permits returned to the pool. */
      permits: number;
    };

export type DurabilitySemaphoreConfig = SchedulerAttachment & {
  /** Evaluated once per round. Lowering it never evicts running work. */
  limit: number | (() => number | Promise<number>);
  /**
   * Per-group ceiling; `undefined` means the group only shares the limit.
   * Throwing fails closed for that group with `limit_resolution_error`.
   */
  groupLimit?: (
    group: string
  ) => number | undefined | Promise<number | undefined>;
  /** Entitlement gate; `false` blocks the group with `not_permitted`. */
  canGrant?: (group: string) => boolean | Promise<boolean>;
  /** How long a round waits to coalesce requests. Defaults to 1 second. */
  roundWaitMs?: number;
  /** Minimum time between rounds. Defaults to 2 seconds. */
  roundPeriodMs?: number;
  /** Granted permits a client never claims are reclaimed after this. Defaults to 10 seconds. */
  grantTimeoutMs?: number;
  /** Omit to disable leases: clients then admit only through confirmed grants. */
  leases?: SemaphoreLeaseOptions;
  onLifecycleEvent?: (event: DurabilitySemaphoreEvent) => void | Promise<void>;
};

export type SemaphoreUsage = {
  limit: number;
  /**
   * Permits granted and not yet released, including unclaimed grants. May
   * exceed `limit` after it was lowered: running work is never evicted.
   */
  granted: number;
  /** Live lease footprint: lease permits not yet released. */
  leased: number;
  /** Permits grants may still receive, under the provisioning mode. */
  available: number;
  groups: ReadonlyMap<string, { granted: number; limit?: number }>;
};

type Pending = {
  request: PermitRequest;
  resolve: (grant: PermitGrant) => void;
  reject: (error: unknown) => void;
};

type GroupPolicy = { limit?: number; blocked?: SemaphoreBlockedReason };

type Allocation = {
  grants: PermitGrant['grants'];
  blocked: PermitGrant['blocked'];
  grantResults: GrantResult[];
  leases: LeaseGrant[];
};

type RoundResult = {
  allocations: Map<Pending, Allocation>;
  usage: SemaphoreUsage;
  requested: number;
  granted: number;
  reclaimed: DurabilitySemaphoreEvent[];
  leaseEvents: LeaseEvent[];
  /** Lease demand went unserved, or grants were blocked by lease footprint. */
  contended: boolean;
};

// setTimeout overflows past 2^31 - 1 ms.
const maxDelayMs = 2 ** 31 - 1;

/** Unclaimed permits are reclaimed once the claim window passes. */
const effectivePermits = (grant: GrantRow, now: number): number =>
  now >= grant.expires_at ? grant.claimed : grant.permits;

const outstanding = (grant: GrantRow, now: number): number =>
  Math.max(0, effectivePermits(grant, now) - grant.released);

/** Permits left queued per (group, reason) across one round's allocations. */
const blockedCounts = (
  allocations: ReadonlyMap<Pending, Allocation>
): Array<[string, SemaphoreBlockedReason, number]> => {
  const counts = new Map<string, number>();
  for (const [entry, allocation] of allocations) {
    const given = new Map<string, number>();
    for (const grant of allocation.grants) {
      given.set(grant.group, (given.get(grant.group) ?? 0) + grant.permits);
    }
    for (const { group, reason } of allocation.blocked) {
      const demanded = entry.request.demand.find((d) => d.group === group);
      const count = (demanded?.count ?? 0) - (given.get(group) ?? 0);
      const key = JSON.stringify([group, reason]);
      counts.set(key, (counts.get(key) ?? 0) + count);
    }
  }
  return [...counts].map(([key, count]) => {
    const [group, reason] = JSON.parse(key) as [string, SemaphoreBlockedReason];
    return [group, reason, count];
  });
};

/**
 * The holder half of {@link DurabilitySemaphore}, free of Workers-only imports
 * so it can be tested outside workerd.
 *
 * A composable DO capability holding one concurrency limit for many client
 * Durable Objects. Requests arriving within a round are allocated together in
 * arrival order; clients report claimed and released permits as cumulative
 * counts, so lost or repeated reports can never free capacity twice.
 */
export class DurabilitySemaphoreServer {
  readonly #engine: Engine;
  readonly #store: HolderStore;
  readonly #limit: () => number | Promise<number>;
  readonly #groupLimit: DurabilitySemaphoreConfig['groupLimit'];
  readonly #canGrant: DurabilitySemaphoreConfig['canGrant'];
  readonly #roundWaitMs: number;
  readonly #roundPeriodMs: number;
  readonly #grantTimeoutMs: number;
  readonly #leases: LeasePolicy | undefined;
  readonly #emit: (event: DurabilitySemaphoreEvent) => void;
  #pending: Pending[] = [];
  #roundTimer: ReturnType<typeof setTimeout> | undefined;
  #round: Promise<void> | undefined;
  #lastRoundAt = -Infinity;
  // From the last round; tells clients in later syncs to return idle lease headroom.
  #contended = false;

  constructor(config: DurabilitySemaphoreConfig) {
    const limit = config.limit;
    this.#limit = typeof limit === 'function' ? limit : () => limit;
    if (
      typeof limit === 'number' &&
      (!Number.isSafeInteger(limit) || limit < 0)
    ) {
      throw new RangeError('limit must be a non-negative safe integer');
    }
    this.#groupLimit = config.groupLimit;
    this.#canGrant = config.canGrant;
    this.#roundWaitMs = config.roundWaitMs ?? 1_000;
    this.#roundPeriodMs = config.roundPeriodMs ?? 2_000;
    this.#grantTimeoutMs = config.grantTimeoutMs ?? 10_000;
    this.#leases = config.leases && resolveLeasePolicy(config.leases);
    for (const [name, value] of [
      ['roundWaitMs', this.#roundWaitMs],
      ['roundPeriodMs', this.#roundPeriodMs],
      ['grantTimeoutMs', this.#grantTimeoutMs],
    ] as const) {
      assertPositiveInteger(name, value);
      if (value > maxDelayMs) {
        throw new RangeError(`${name} must be at most ${maxDelayMs}`);
      }
    }

    this.#engine = engineFor(config);
    this.#emit = traceLifecycle(
      this.#engine.emitter<DurabilitySemaphoreEvent>(config.onLifecycleEvent)
    );
    this.#engine.migrateSchema('semaphore', durabilitySemaphoreMigrations);
    this.#store = new HolderStore(this.#engine.sql);
    this.#engine.register({
      table: grantsTable,
      runDue: () => this.#reclaimExpired(),
      nextPendingAt: () => this.#store.nextPendingAt(),
    });
  }

  static migrate(
    context: Pick<DurableObjectState, 'storage'>,
    target?: string | null
  ): DurabilityMigrationResult {
    return migrateCapability(
      context.storage,
      'semaphore',
      durabilitySemaphoreMigrations,
      target
    );
  }

  alarm(info?: AlarmInvocationInfo): Promise<void> {
    return this.#engine.alarm(info);
  }

  /**
   * The single entry point the application exposes over RPC. A `request` is
   * answered when its round runs; the caller must keep awaiting it. Pending
   * requests live in memory only: if the object is evicted first, the call
   * fails and the client asks again.
   */
  handle(message: SemaphoreRequest): Promise<SemaphoreResponse> {
    const parsed = semaphoreRequestSchema.safeParse(message);
    if (!parsed.success) {
      return Promise.reject(
        new SemaphoreProtocolError('Malformed semaphore message')
      );
    }
    if (parsed.data.type === 'sync') {
      return this.#sync(parsed.data);
    }

    const request = parsed.data;
    return new Promise<PermitGrant>((resolve, reject) => {
      this.#pending.push({ request, resolve, reject });
      this.#scheduleRound();
    });
  }

  async usage(): Promise<SemaphoreUsage> {
    const now = Date.now();
    const limit = await this.#resolveLimit();
    const grants = this.#store.list();
    const leases = this.#store.listLeases();
    const groups = new Map<string, { granted: number; limit?: number }>();
    let granted = 0;
    for (const grant of grants) {
      const count = outstanding(grant, now);
      granted += count;
      const group = groups.get(grant.grp) ?? { granted: 0 };
      group.granted += count;
      groups.set(grant.grp, group);
    }
    let leased = 0;
    for (const lease of leases) {
      leased += leaseFootprint(lease);
    }
    await Promise.all(
      [...groups.entries()].map(async ([name, group]) => {
        const policy = await this.#groupPolicy(name);
        if (policy.limit !== undefined) {
          group.limit = policy.limit;
        }
      })
    );
    return {
      limit,
      granted,
      leased,
      available: this.#grantRoom(limit, granted, leased),
      groups,
    };
  }

  /** Permits grants may still receive, given the provisioning mode. */
  #grantRoom(limit: number, granted: number, leased: number): number {
    if (!this.#leases) {
      return Math.max(0, limit - granted);
    }
    const budget = budgets(this.#leases.provisioning, limit);
    return Math.max(0, budget.grants - granted - (budget.shared ? leased : 0));
  }

  // Rounds never overlap: a request arriving mid-round joins the next one,
  // which is scheduled once the current round settles.
  #scheduleRound(): void {
    if (this.#roundTimer !== undefined || this.#round !== undefined) {
      return;
    }
    const earliest = this.#lastRoundAt + this.#roundPeriodMs;
    const delay = Math.max(this.#roundWaitMs, earliest - Date.now());
    this.#roundTimer = setTimeout(() => {
      this.#roundTimer = undefined;
      const onError = (error: unknown) =>
        reportFailure('durability.semaphore.round_failed', {}, error);
      this.#round = this.#runRound()
        .catch(onError)
        .finally(() => {
          this.#round = undefined;
          if (this.#pending.length > 0) {
            this.#scheduleRound();
          }
        });
      this.#engine.waitUntil(this.#round, onError);
    }, delay);
  }

  async #runRound(): Promise<void> {
    const pending = this.#pending;
    this.#pending = [];
    const now = Date.now();
    this.#lastRoundAt = now;

    let round: RoundResult;
    try {
      const limit = await this.#resolveLimit();
      // Held leases are re-checked every round, not only when their group is
      // requested, so a group that became limited or blocked is retired.
      const leaseGroups = this.#leases
        ? this.#store.listLeases().map((lease) => lease.grp)
        : [];
      const groups = new Set([
        ...pending.flatMap(({ request }) => request.demand.map((d) => d.group)),
        ...leaseGroups,
      ]);
      const policies = new Map(
        await Promise.all(
          [...groups].map(
            async (group) => [group, await this.#groupPolicy(group)] as const
          )
        )
      );
      round = await this.#engine.transaction(() =>
        this.#allocate(pending, limit, policies, now)
      );
    } catch (error) {
      for (const entry of pending) {
        entry.reject(error);
      }
      throw error;
    }

    const nextRequestAt = now + this.#roundPeriodMs;
    this.#contended = round.contended;
    for (const [entry, allocation] of round.allocations) {
      const { leases, ...rest } = allocation;
      entry.resolve({
        type: 'grant',
        v: 1,
        ...rest,
        nextRequestAt,
        ...(entry.request.leases === undefined
          ? {}
          : { leases, ...(round.contended ? { contended: true } : {}) }),
      });
    }
    for (const event of [...round.reclaimed, ...round.leaseEvents]) {
      this.#emit(event);
    }
    this.#emit({
      entityKind: 'semaphore',
      type: 'round',
      timestamp: Date.now(),
      requested: round.requested,
      granted: round.granted,
      usage: round.usage,
    });
    for (const [group, reason, count] of blockedCounts(round.allocations)) {
      this.#emit({
        entityKind: 'semaphore',
        type: 'blocked',
        timestamp: Date.now(),
        group,
        reason,
        count,
      });
    }
  }

  /** One round: apply reports, reclaim, then hand out permits in arrival order. */
  #allocate(
    pending: Pending[],
    limit: number,
    policies: ReadonlyMap<string, GroupPolicy>,
    now: number
  ): RoundResult {
    const store = this.#store;
    const allocations = new Map<Pending, Allocation>();
    let unknownDebt = 0;
    const drawn = new Map<string, number>();
    for (const entry of pending) {
      const { clientId } = entry.request;
      const grantResults = this.#applyReports(
        store,
        clientId,
        entry.request.grants,
        now
      );
      const leases: LeaseGrant[] = [];
      if (this.#leases && entry.request.leases) {
        const outcome = applyLeaseReports(
          store,
          this.#leases,
          clientId,
          entry.request.leases,
          now
        );
        leases.push(...outcome.echoes);
        unknownDebt += outcome.unknownDebt;
        for (const [leaseId, count] of outcome.drawn) {
          drawn.set(leaseId, count);
        }
      }
      allocations.set(entry, { grants: [], blocked: [], grantResults, leases });
    }

    const used = new Map<string, number>();
    let total = 0;
    const reclaimed = this.#reclaim(store, now);
    for (const grant of reclaimed.grants) {
      const count = outstanding(grant, now);
      total += count;
      used.set(grant.grp, (used.get(grant.grp) ?? 0) + count);
    }
    // In strict mode live lease footprint is reserved from the same pool as grants.
    const budget = this.#leases
      ? budgets(this.#leases.provisioning, limit)
      : { grants: limit, leases: 0, shared: false };
    let leased = 0;
    if (this.#leases && budget.shared) {
      for (const lease of store.listLeases()) {
        leased += leaseFootprint(lease);
      }
    }
    let remaining = budget.grants - total - (budget.shared ? leased : 0);

    const demands = pending
      .flatMap((entry) =>
        entry.request.demand.map((demand) => ({ entry, ...demand }))
      )
      .sort(
        (left, right) =>
          left.oldestAt - right.oldestAt ||
          left.entry.request.clientId.localeCompare(
            right.entry.request.clientId
          )
      );
    let requested = 0;
    let granted = 0;
    for (const demand of demands) {
      requested += demand.count;
      const allocation = allocations.get(demand.entry)!;
      const policy = policies.get(demand.group) ?? {};
      if (policy.blocked) {
        allocation.blocked.push({
          group: demand.group,
          reason: policy.blocked,
        });
        continue;
      }

      const groupUsed = used.get(demand.group) ?? 0;
      const groupRemaining =
        policy.limit === undefined ? Infinity : policy.limit - groupUsed;
      const give = Math.max(
        0,
        Math.min(demand.count, groupRemaining, remaining)
      );
      if (give > 0) {
        const grant: GrantRow = {
          grant_id: crypto.randomUUID(),
          client_id: demand.entry.request.clientId,
          grp: demand.group,
          permits: give,
          claimed: 0,
          released: 0,
          created_at: now,
          expires_at: now + this.#grantTimeoutMs,
        };
        store.insert(grant);
        allocation.grants.push({
          grantId: grant.grant_id,
          group: demand.group,
          permits: give,
        });
        remaining -= give;
        used.set(demand.group, groupUsed + give);
        granted += give;
      }
      if (give < demand.count) {
        // The group gate is checked first so attribution names the tighter one.
        allocation.blocked.push({
          group: demand.group,
          reason: groupRemaining - give <= 0 ? 'group_limit' : 'limit',
        });
      }
    }
    let leaseEvents: LeaseEvent[] = [];
    let contended = false;
    if (this.#leases) {
      const candidates: LeaseCandidate[] = pending.flatMap((entry) =>
        entry.request.leases === undefined
          ? []
          : entry.request.demand.map((demand) => ({
              clientId: entry.request.clientId,
              group: demand.group,
              oldestAt: demand.oldestAt,
            }))
      );
      // The allocator nets out live footprint itself; hand it gross capacity.
      const leaseBudget = budget.shared
        ? Math.max(0, limit - total - granted)
        : budget.leases;
      const outcome = allocateLeases(store, this.#leases, {
        budget: leaseBudget,
        unknownDebt,
        drawn,
        candidates,
        policies,
        groupUsed: used,
        now,
      });
      leased = outcome.footprint;
      leaseEvents = outcome.events;
      // Returned headroom helps grants only when they share the limit with leases.
      contended =
        outcome.unserved ||
        (budget.shared &&
          [...allocations.values()].some((allocation) =>
            allocation.blocked.some(({ reason }) => reason === 'limit')
          ));
      for (const [entry, allocation] of allocations) {
        // One echo per lease; the allocator's view (e.g. retired) supersedes the report's.
        const byLease = new Map(
          [
            ...allocation.leases,
            ...(outcome.grants.get(entry.request.clientId) ?? []),
          ].map((lease) => [lease.leaseId, lease] as const)
        );
        allocation.leases = [...byLease.values()];
      }
      if (budget.shared) {
        remaining = limit - total - granted - leased;
      }
    }

    const groups = new Map(
      [...used].map(([group, count]) => {
        const policy = policies.get(group);
        return [
          group,
          policy?.limit === undefined
            ? { granted: count }
            : { granted: count, limit: policy.limit },
        ];
      })
    );
    return {
      allocations,
      requested,
      granted,
      reclaimed: reclaimed.events,
      leaseEvents,
      contended,
      usage: {
        limit,
        granted: total + granted,
        leased,
        available: Math.max(0, remaining),
        groups,
      },
    };
  }

  async #sync(message: PermitSync): Promise<PermitSyncResult> {
    const now = Date.now();
    const result = await this.#engine.transaction(() => {
      const store = this.#store;
      const grantResults = this.#applyReports(
        store,
        message.clientId,
        message.grants,
        now
      );
      const leaseOutcome =
        this.#leases && message.leases
          ? applyLeaseReports(
              store,
              this.#leases,
              message.clientId,
              message.leases,
              now
            )
          : undefined;
      return { grantResults, leaseEchoes: leaseOutcome?.echoes };
    });
    return {
      type: 'synced',
      v: 1,
      grantResults: result.grantResults,
      ...(result.leaseEchoes === undefined
        ? {}
        : {
            leases: result.leaseEchoes,
            ...(this.#contended ? { contended: true } : {}),
          }),
    };
  }

  /**
   * Applies cumulative claim/release reports. Counts only move forward and
   * are clamped to what the grant can honor, so replays are harmless.
   */
  #applyReports(
    store: HolderStore,
    clientId: string,
    reports: GrantReport[],
    now: number
  ): GrantResult[] {
    const results: GrantResult[] = [];
    for (const report of reports) {
      const grant = store.get(report.grantId);
      if (!grant || grant.client_id !== clientId) {
        results.push({
          grantId: report.grantId,
          claimed: 0,
          status: 'not_found',
        });
        continue;
      }

      const expired = now >= grant.expires_at;
      const permits = effectivePermits(grant, now);
      const claimed = Math.max(
        grant.claimed,
        Math.min(report.claimed, permits)
      );
      const released = Math.max(
        grant.released,
        Math.min(report.released, claimed)
      );
      const updated: GrantRow = { ...grant, permits, claimed, released };
      if (released >= permits) {
        store.remove(grant.grant_id);
      } else {
        store.update(updated);
      }
      results.push({
        grantId: grant.grant_id,
        claimed,
        status: expired && report.claimed > claimed ? 'expired' : 'ok',
      });
    }
    return results;
  }

  /**
   * Returns every grant after reclaiming unclaimed permits past their window.
   * Events are returned, not emitted: the caller emits them after commit.
   */
  #reclaim(
    store: HolderStore,
    now: number
  ): { grants: GrantRow[]; events: DurabilitySemaphoreEvent[] } {
    const kept: GrantRow[] = [];
    const events: DurabilitySemaphoreEvent[] = [];
    for (const grant of store.list()) {
      if (now < grant.expires_at || grant.claimed >= grant.permits) {
        kept.push(grant);
        continue;
      }

      const updated: GrantRow = { ...grant, permits: grant.claimed };
      if (updated.released >= updated.permits) {
        store.remove(grant.grant_id);
      } else {
        store.update(updated);
        kept.push(updated);
      }
      events.push({
        entityKind: 'semaphore',
        type: 'grant_reclaimed',
        timestamp: now,
        grantId: grant.grant_id,
        clientId: grant.client_id,
        group: grant.grp,
        permits: grant.permits - grant.claimed,
      });
    }
    return { grants: kept, events };
  }

  async #reclaimExpired(): Promise<void> {
    const events = await this.#engine.transaction(
      () => this.#reclaim(this.#store, Date.now()).events
    );
    for (const event of events) {
      this.#emit(event);
    }
  }

  async #resolveLimit(): Promise<number> {
    const limit = await this.#limit();
    if (!Number.isSafeInteger(limit) || limit < 0) {
      throw new RangeError('limit must resolve to a non-negative safe integer');
    }
    return limit;
  }

  // Resolution failures block the group rather than granting unlimited permits.
  async #groupPolicy(group: string): Promise<GroupPolicy> {
    try {
      if (this.#canGrant && !(await this.#canGrant(group))) {
        return { blocked: 'not_permitted' };
      }
      const limit = await this.#groupLimit?.(group);
      if (limit === undefined) {
        return {};
      }
      if (!Number.isSafeInteger(limit) || limit < 0) {
        throw new RangeError(
          `groupLimit for "${group}" must be a non-negative safe integer`
        );
      }
      return { limit };
    } catch (error) {
      reportFailure(
        'durability.semaphore.limit_resolution_failed',
        { group },
        error
      );
      return { blocked: 'limit_resolution_error' };
    }
  }
}

traceMethods(DurabilitySemaphoreServer.prototype, 'DurabilitySemaphore');

import { z } from 'zod';

export const grantRowSchema = z.object({
  grant_id: z.string(),
  client_id: z.string(),
  grp: z.string(),
  permits: z.number().int().nonnegative(),
  claimed: z.number().int().nonnegative(),
  released: z.number().int().nonnegative(),
  created_at: z.number().int().nonnegative(),
  /** Unclaimed permits are reclaimed once this passes. */
  expires_at: z.number().int().nonnegative(),
});

export type GrantRow = z.infer<typeof grantRowSchema>;

/** Permits a client may draw locally without a round trip; held by the holder. */
export const leaseRowSchema = z.object({
  lease_id: z.string(),
  client_id: z.string(),
  grp: z.string(),
  permits: z.number().int().nonnegative(),
  /** Drawn permits the client has returned, as a high-water mark. */
  released: z.number().int().nonnegative(),
  created_at: z.number().int().nonnegative(),
  /** Refreshed by every report; the lease is swept once this passes. */
  expires_at: z.number().int().nonnegative(),
  /** 1 once no more permits may be drawn; footprint stays until released. */
  retired: z.number().int().min(0).max(1),
});

export type LeaseRow = z.infer<typeof leaseRowSchema>;

const grantColumns = grantRowSchema.keyof().options;
const leaseColumns = leaseRowSchema.keyof().options;

/**
 * Permits a semaphore has handed out. Grants and leases are small and bounded
 * by live clients, so aggregates are computed from `list()`.
 */
export class HolderStore {
  constructor(private readonly sql: SqlStorage) {}

  list(): GrantRow[] {
    return grantRowSchema
      .array()
      .parse(
        this.sql
          .exec('SELECT * FROM durability_semaphore_grants ORDER BY created_at')
          .toArray()
      );
  }

  get(grantId: string): GrantRow | undefined {
    return grantRowSchema
      .optional()
      .parse(
        this.sql
          .exec(
            'SELECT * FROM durability_semaphore_grants WHERE grant_id = ? LIMIT 1',
            grantId
          )
          .toArray()[0]
      );
  }

  nextPendingAt(): number | undefined {
    return grantRowSchema
      .pick({ expires_at: true })
      .optional()
      .parse(
        this.sql
          .exec(
            `SELECT expires_at FROM durability_semaphore_grants
             WHERE claimed < permits
             ORDER BY expires_at ASC
             LIMIT 1`
          )
          .toArray()[0]
      )?.expires_at;
  }

  insert(row: GrantRow): boolean {
    return (
      this.sql
        .exec(
          `INSERT INTO durability_semaphore_grants (${grantColumns.join(', ')})
           VALUES (${grantColumns.map(() => '?').join(', ')})
           ON CONFLICT(grant_id) DO NOTHING
           RETURNING grant_id`,
          ...grantColumns.map((column) => row[column])
        )
        .toArray().length > 0
    );
  }

  update(row: GrantRow): void {
    this.sql.exec(
      `UPDATE durability_semaphore_grants
       SET permits = ?, claimed = ?, released = ?, expires_at = ?
       WHERE grant_id = ?`,
      row.permits,
      row.claimed,
      row.released,
      row.expires_at,
      row.grant_id
    );
  }

  remove(grantId: string): void {
    this.sql.exec(
      'DELETE FROM durability_semaphore_grants WHERE grant_id = ?',
      grantId
    );
  }

  listLeases(): LeaseRow[] {
    return leaseRowSchema
      .array()
      .parse(
        this.sql
          .exec('SELECT * FROM durability_semaphore_leases ORDER BY created_at')
          .toArray()
      );
  }

  getLease(leaseId: string): LeaseRow | undefined {
    return leaseRowSchema
      .optional()
      .parse(
        this.sql
          .exec(
            'SELECT * FROM durability_semaphore_leases WHERE lease_id = ? LIMIT 1',
            leaseId
          )
          .toArray()[0]
      );
  }

  insertLease(row: LeaseRow): boolean {
    return (
      this.sql
        .exec(
          `INSERT INTO durability_semaphore_leases (${leaseColumns.join(', ')})
           VALUES (${leaseColumns.map(() => '?').join(', ')})
           ON CONFLICT(lease_id) DO NOTHING
           RETURNING lease_id`,
          ...leaseColumns.map((column) => row[column])
        )
        .toArray().length > 0
    );
  }

  updateLease(row: LeaseRow): void {
    this.sql.exec(
      `UPDATE durability_semaphore_leases
       SET permits = ?, released = ?, expires_at = ?, retired = ?
       WHERE lease_id = ?`,
      row.permits,
      row.released,
      row.expires_at,
      row.retired,
      row.lease_id
    );
  }

  removeLease(leaseId: string): void {
    this.sql.exec(
      'DELETE FROM durability_semaphore_leases WHERE lease_id = ?',
      leaseId
    );
  }
}

import { assertPositiveInteger } from './policy.js';
import type {
  LeaseGrant,
  LeaseReport,
  SemaphoreBlockedReason,
} from './semaphore-protocol.js';
import type { HolderStore, LeaseRow } from './semaphore-store.js';

/**
 * Where lease capacity comes from.
 *
 * - `over`: leases draw from a separate budget above the limit, so the
 *   ceiling is `limit + extra` (the original gatekeeper behaviour).
 * - `under`: a slice of the limit is reserved for leases; grants may use at
 *   most `limit - reserve`, and always at least one permit. Nothing ever
 *   exceeds the limit and leases have guaranteed headroom.
 * - `strict`: grants and live lease footprint share the limit, grants first.
 */
export type SemaphoreProvisioning =
  | { mode: 'strict' }
  | { mode: 'under'; reserve: number | { ratio: number } }
  | { mode: 'over'; extra: number | { ratio: number } };

export type SemaphoreLeaseOptions = {
  provisioning: SemaphoreProvisioning;
  /** Permits per lease. Defaults to 64. */
  permitsPerLease?: number;
  /** A lease not reported for this long is swept. Defaults to 30 minutes. */
  ttlMs?: number;
  /** Groups that may receive new leases in one round, oldest demand first. Defaults to 100. */
  maxGroups?: number;
  /** Clients per group that may receive new leases in one round. Defaults to 100. */
  maxClientsPerGroup?: number;
  /**
   * Groups with a `groupLimit`: `exclude` them from leases (exact steady
   * state only, as gatekeeper does) or lease `withinGroupLimit`. Defaults to
   * `exclude`.
   */
  limitedGroups?: 'exclude' | 'withinGroupLimit';
};

export type LeasePolicy = Required<SemaphoreLeaseOptions>;

export type LeaseEvent = {
  entityKind: 'semaphore';
  type: 'lease_granted' | 'lease_retired';
  timestamp: number;
  leaseId: string;
  clientId: string;
  group: string;
  permits: number;
};

type GroupPolicy = { limit?: number; blocked?: SemaphoreBlockedReason };

const maxDelayMs = 2 ** 31 - 1;

export const resolveLeasePolicy = (
  options: SemaphoreLeaseOptions
): LeasePolicy => {
  const policy: LeasePolicy = {
    provisioning: options.provisioning,
    permitsPerLease: options.permitsPerLease ?? 64,
    ttlMs: options.ttlMs ?? 30 * 60_000,
    maxGroups: options.maxGroups ?? 100,
    maxClientsPerGroup: options.maxClientsPerGroup ?? 100,
    limitedGroups: options.limitedGroups ?? 'exclude',
  };
  for (const name of [
    'permitsPerLease',
    'ttlMs',
    'maxGroups',
    'maxClientsPerGroup',
  ] as const) {
    assertPositiveInteger(`leases.${name}`, policy[name]);
  }
  if (policy.ttlMs > maxDelayMs) {
    throw new RangeError(`leases.ttlMs must be at most ${maxDelayMs}`);
  }
  const amount =
    policy.provisioning.mode === 'under'
      ? policy.provisioning.reserve
      : policy.provisioning.mode === 'over'
        ? policy.provisioning.extra
        : undefined;
  if (amount !== undefined) {
    const value = typeof amount === 'number' ? amount : amount.ratio;
    if (!Number.isFinite(value) || value < 0) {
      throw new RangeError('lease provisioning amounts must be non-negative');
    }
  }
  return policy;
};

const scaled = (amount: number | { ratio: number }, limit: number): number =>
  typeof amount === 'number' ? amount : Math.floor(limit * amount.ratio);

/** Permits grants may use, and permits leases may hold, for one limit. */
export const budgets = (
  provisioning: SemaphoreProvisioning,
  limit: number
): { grants: number; leases: number; shared: boolean } => {
  if (provisioning.mode === 'over') {
    return {
      grants: limit,
      leases: scaled(provisioning.extra, limit),
      shared: false,
    };
  }
  if (provisioning.mode === 'under') {
    // The reserve never takes the last permit: groups excluded from leases and
    // clients without lease support would otherwise never be admitted.
    const reserve = Math.min(
      Math.max(0, limit - 1),
      scaled(provisioning.reserve, limit)
    );
    return { grants: limit - reserve, leases: reserve, shared: false };
  }
  return { grants: limit, leases: limit, shared: true };
};

export const leaseFootprint = (lease: LeaseRow): number =>
  Math.max(0, lease.permits - lease.released);

/** A retired lease is echoed as expired so the client stops drawing at once. */
const toLeaseGrant = (lease: LeaseRow, now: number): LeaseGrant => ({
  leaseId: lease.lease_id,
  group: lease.grp,
  permits: lease.permits,
  expiresAt: lease.retired ? now : lease.expires_at,
});

export const leaseEligible = (
  policy: LeasePolicy,
  group: GroupPolicy
): boolean =>
  group.blocked === undefined &&
  (group.limit === undefined || policy.limitedGroups === 'withinGroupLimit');

export type LeaseReportOutcome = {
  /** Echoed grants: refreshed live leases and retired (`expiresAt = now`) ones. */
  echoes: LeaseGrant[];
  /** Live footprint of leases the holder no longer knows, charged this round. */
  unknownDebt: number;
  /** Reported `drawn` per lease, so undrawn headroom can be seen. */
  drawn: Map<string, number>;
};

/**
 * Applies one client's lease reports: heartbeats live leases, advances the
 * release high-water mark, retires leases the holder forgot or that expired,
 * and charges their reported live footprint so a forgotten lease cannot
 * over-admit for the rest of this round.
 */
export const applyLeaseReports = (
  store: HolderStore,
  policy: LeasePolicy,
  clientId: string,
  reports: LeaseReport[],
  now: number
): LeaseReportOutcome => {
  const outcome: LeaseReportOutcome = {
    echoes: [],
    unknownDebt: 0,
    drawn: new Map(),
  };
  for (const report of reports) {
    const lease = store.getLease(report.leaseId);
    if (!lease || lease.client_id !== clientId) {
      outcome.echoes.push({
        leaseId: report.leaseId,
        group: lease?.grp ?? '',
        permits: lease?.permits ?? 0,
        expiresAt: now,
      });
      // Client-reported usage is untrusted: clamp before it can affect budget.
      const live = Math.max(0, report.drawn - Math.max(0, report.released));
      outcome.unknownDebt += Math.min(live, policy.permitsPerLease);
      continue;
    }

    const drawn = Math.min(Math.max(0, report.drawn), lease.permits);
    outcome.drawn.set(lease.lease_id, drawn);
    const released = Math.max(
      lease.released,
      Math.min(Math.max(0, report.released), drawn)
    );
    // An expired lease is retired, not forgotten: its permits may be running.
    // Only the client's own retired report shrinks a lease to what it drew:
    // until then it may still be drawing, whoever asked it to stop.
    const retired =
      lease.retired === 1 || report.retired === true || lease.expires_at <= now;
    const permits =
      report.retired === true ? Math.max(released, drawn) : lease.permits;
    if (released >= permits) {
      store.removeLease(lease.lease_id);
      // Echoed as settled (permits = released) so the client can drop the lease.
      outcome.echoes.push({
        ...toLeaseGrant(lease, now),
        permits: released,
        expiresAt: now,
      });
      continue;
    }
    const refreshed: LeaseRow = {
      ...lease,
      permits,
      released,
      retired: retired ? 1 : 0,
      expires_at: retired ? lease.expires_at : now + policy.ttlMs,
    };
    store.updateLease(refreshed);
    outcome.echoes.push(toLeaseGrant(refreshed, now));
  }
  return outcome;
};

export type LeaseCandidate = {
  clientId: string;
  group: string;
  oldestAt: number;
};

export type LeaseAllocation = {
  /** New and retired leases per client. */
  grants: Map<string, LeaseGrant[]>;
  footprint: number;
  events: LeaseEvent[];
  /** Some eligible candidate received no lease this round. */
  unserved: boolean;
};

/**
 * One round's lease pass: sweeps expired leases, retires leases of groups no
 * longer eligible, computes the live footprint, and shares the lease budget
 * equally among candidates that do not already hold usable headroom.
 */
export const allocateLeases = (
  store: HolderStore,
  policy: LeasePolicy,
  input: {
    budget: number;
    unknownDebt: number;
    drawn: ReadonlyMap<string, number>;
    candidates: LeaseCandidate[];
    policies: ReadonlyMap<string, GroupPolicy>;
    /** Group permits already used by grants, for `withinGroupLimit`. */
    groupUsed: ReadonlyMap<string, number>;
    now: number;
  }
): LeaseAllocation => {
  const { now } = input;
  const grants = new Map<string, LeaseGrant[]>();
  const events: LeaseEvent[] = [];
  const echo = (clientId: string, grant: LeaseGrant): void => {
    grants.set(clientId, [...(grants.get(clientId) ?? []), grant]);
  };

  let footprint = 0;
  const groupFootprint = new Map<string, number>();
  const usable = new Set<string>();
  for (const lease of store.listLeases()) {
    const group = input.policies.get(lease.grp) ?? {};
    let current = lease;
    // An unreported lease is retired, not swept: its client may be cut off
    // with work still running, so its footprint stays until it reports, as a
    // claimed grant permit stays until it is released.
    if (
      !lease.retired &&
      (lease.expires_at <= now || !leaseEligible(policy, group))
    ) {
      // Stop draws now; the full footprint stays until the client confirms.
      current = { ...lease, retired: 1 };
      store.updateLease(current);
      echo(lease.client_id, toLeaseGrant(current, now));
      events.push({
        entityKind: 'semaphore',
        type: 'lease_retired',
        timestamp: now,
        leaseId: lease.lease_id,
        clientId: lease.client_id,
        group: lease.grp,
        permits:
          lease.permits - (input.drawn.get(lease.lease_id) ?? lease.permits),
      });
    }

    const live = leaseFootprint(current);
    footprint += live;
    groupFootprint.set(lease.grp, (groupFootprint.get(lease.grp) ?? 0) + live);
    // An unreported lease is assumed fully drawn, as gatekeeper does.
    if (
      !current.retired &&
      (input.drawn.get(lease.lease_id) ?? lease.permits) < lease.permits
    ) {
      usable.add(JSON.stringify([lease.client_id, lease.grp]));
    }
  }

  const byGroup = new Map<string, LeaseCandidate[]>();
  for (const candidate of input.candidates) {
    if (
      !leaseEligible(policy, input.policies.get(candidate.group) ?? {}) ||
      usable.has(JSON.stringify([candidate.clientId, candidate.group]))
    ) {
      continue;
    }
    const existing = byGroup.get(candidate.group) ?? [];
    const prior = existing.find((c) => c.clientId === candidate.clientId);
    if (prior) {
      prior.oldestAt = Math.min(prior.oldestAt, candidate.oldestAt);
    } else {
      existing.push({ ...candidate });
    }
    byGroup.set(candidate.group, existing);
  }
  const chosen = [...byGroup.entries()]
    .map(([group, clients]) => ({
      group,
      clients: clients
        .sort(
          (a, b) =>
            a.oldestAt - b.oldestAt || a.clientId.localeCompare(b.clientId)
        )
        .slice(0, policy.maxClientsPerGroup),
      oldestAt: Math.min(...clients.map((c) => c.oldestAt)),
    }))
    .sort((a, b) => a.oldestAt - b.oldestAt || a.group.localeCompare(b.group))
    .slice(0, policy.maxGroups)
    .flatMap(({ clients }) => clients);

  // A lowered budget retires leases with undrawn headroom, newest first, until
  // the headroom their clients reported covers the excess. Their footprint
  // stays until each client confirms what it actually drew.
  let excess = footprint + input.unknownDebt - input.budget;
  if (excess > 0) {
    const live = store
      .listLeases()
      .filter((lease) => lease.expires_at > now && !lease.retired)
      .sort((a, b) => b.created_at - a.created_at);
    for (const lease of live) {
      if (excess <= 0) {
        break;
      }
      const drawn = input.drawn.get(lease.lease_id);
      if (drawn === undefined || drawn >= lease.permits) {
        continue;
      }
      const retired: LeaseRow = { ...lease, retired: 1 };
      store.updateLease(retired);
      excess -= lease.permits - drawn;
      echo(lease.client_id, toLeaseGrant(retired, now));
      events.push({
        entityKind: 'semaphore',
        type: 'lease_retired',
        timestamp: now,
        leaseId: lease.lease_id,
        clientId: lease.client_id,
        group: lease.grp,
        permits: lease.permits - drawn,
      });
    }
  }

  const eligible = [...byGroup.values()].reduce(
    (n, clients) => n + clients.length,
    0
  );
  let remaining = Math.max(0, input.budget - footprint - input.unknownDebt);
  // With fewer permits than candidates, the oldest demand gets one each
  // rather than everyone getting a rounded-down share of nothing.
  const share = Math.min(
    policy.permitsPerLease,
    Math.max(1, Math.floor(remaining / Math.max(1, chosen.length)))
  );

  let issued = 0;
  for (const candidate of chosen) {
    if (remaining <= 0) {
      break;
    }
    const group = input.policies.get(candidate.group) ?? {};
    const groupRoom =
      group.limit === undefined
        ? share
        : group.limit -
          (input.groupUsed.get(candidate.group) ?? 0) -
          (groupFootprint.get(candidate.group) ?? 0);
    const permits = Math.min(share, groupRoom, remaining);
    if (permits <= 0) {
      continue;
    }
    remaining -= permits;
    issued += 1;
    const lease: LeaseRow = {
      lease_id: crypto.randomUUID(),
      client_id: candidate.clientId,
      grp: candidate.group,
      permits,
      released: 0,
      created_at: now,
      expires_at: now + policy.ttlMs,
      retired: 0,
    };
    store.insertLease(lease);
    footprint += permits;
    groupFootprint.set(
      candidate.group,
      (groupFootprint.get(candidate.group) ?? 0) + permits
    );
    echo(candidate.clientId, toLeaseGrant(lease, now));
    events.push({
      entityKind: 'semaphore',
      type: 'lease_granted',
      timestamp: now,
      leaseId: lease.lease_id,
      clientId: lease.client_id,
      group: lease.grp,
      permits,
    });
  }
  return { grants, footprint, events, unserved: issued < eligible };
};

Allocate a round

Details

Follow one permit request from a client object to stored grants and leases in the holder’s storage.

Durable state: Nothing is written yet.
In-flight / in memory: The helper, its round timer, and the example’s workflowLimits map exist only in this instance.

Click the diagram to open a full-screen explorer with the complete source files. Select a component to jump to its code, or select a linked code line to highlight the affected architecture. Use the ← / → keys or on-screen arrows to move through the flow; each step’s explanation sits under the diagram. Escape or the close button returns to the guide. Failure scenarios, the full step list, and durable/in-memory state notes are available under Details. These are explanatory traces of the real source, not executable simulations or guarantees about an external service.

JavaScript is disabled. Read the complete example source or follow the complete ordered walkthrough below.

examples/account-concurrency/worker.ts

packages/durability/src/semaphore.ts

packages/durability/src/semaphore-server.ts

packages/durability/src/semaphore-store.ts

packages/durability/src/semaphore-leases.ts

Allocate a round

The limit lives in an object you own. The semaphore attaches to a DurabilityScheduler built on this.ctx, so it uses this object’s storage and alarm. limit() and groupLimit() are your callbacks; the semaphore evaluates them once per round. Read these source lines. Durable state: Nothing is written yet. In memory: The helper, its round timer, and the example’s workflowLimits map exist only in this instance.
Create grant and lease tables. The constructor applies the semaphore migrations through ctx.storage.sql, then registers with the engine. Only the grants table takes part in alarm passes: due work there is reclaiming unclaimed permits. Read these source lines. Durable state: durability_semaphore_grants and durability_semaphore_leases exist in this object’s storage. In memory: The registration with the engine is repeated on each construction.
A client object asks for permits. DurabilitySemaphore.client() gets the holder with getByName(key) and calls your permits() method, which forwards the message to handle(). The message carries the client’s demand per group and cumulative reports for the grants and leases it holds. Read these source lines. Durable state: Nothing is written until the round runs. In memory: The client’s RPC call stays open until its round.
The request waits in memory for its round. handle() pushes the request onto an in-memory list and sets a timer. A round waits at least roundWaitMs (1 second) to collect concurrent requests and starts no sooner than roundPeriodMs (2 seconds) after the last one. Rounds never overlap. Read these source lines. Durable state: Nothing is written for a waiting request. In memory: The request, its resolve callback, and the round timer. The round promise is passed to ctx.waitUntil.
Apply cumulative reports first. The round runs in one storage transaction. Claimed and released counts only move forward and are clamped to what the grant can honour, so a repeated or reordered report cannot free capacity twice. A fully released grant is deleted. Read these source lines. Durable state: Grant rows advance to the reported counts. In memory: Nothing; reports are applied inside the transaction.
Grant the oldest demand first. After reclaiming expired unclaimed permits, the round sorts all demand by its oldest queued time and grants up to the group limit and what remains of the limit. Each grant row gets a claim deadline grantTimeoutMs (10 seconds) away. Read these source lines. Durable state: New grant rows with claimed = 0 count against the limit. In memory: The per-request allocations, until the reply.
Lease what is left. In strict mode leases share the limit with grants: the lease budget is what grants left, and a live lease’s unreleased permits stay reserved from the same pool. Only clients that sent a leases array are candidates. Read these source lines. Durable state: New lease rows with an expiry ttlMs away. In memory: Nothing beyond the reply being built.
Set the alarm for the claim deadline. Before the transaction commits, the scheduler sets the object’s one alarm from the earliest deadline of a grant with unclaimed permits, never sooner than alarmMinDelayMs (15 seconds). A sooner deadline is covered by an in-memory timer. Lease expiry does not set the alarm. Read these source lines. Durable state: The alarm commits with the grant rows it serves. In memory: The in-memory timer for a deadline before the alarm.
Resolve the waiting RPC. Only after the commit does each waiting promise resolve, with its grants, blocked groups, clamped claim results, leases, and nextRequestAt: when the client may ask again. Read these source lines. Durable state: Grants and leases are committed before the reply leaves. In memory: The reply in flight. If it is lost, the capacity stays reserved.

Lose the object and the reply

Evicted before the round runs. Waiting requests are promises in memory. If the holder is evicted before its round, every waiting RPC call fails and nothing was allocated; each client must ask again. Read these source lines. Durable state: Grant and lease rows from earlier rounds are untouched. In memory: The waiting requests, the round timer, and the contended flag are lost.
The round commits but the reply is lost. If the RPC breaks after the round’s transaction commits, the grants exist but the client never learns their IDs. The holder cannot tell this from a slow client, so the permits stay reserved. Read these source lines. Durable state: Unclaimed grant rows count against the limit. In memory: The reply. It is not stored, so a retried request gets new grants.
The alarm brings the holder back. At the claim deadline the alarm fires. The runtime creates a new instance, which builds a new scheduler and semaphore, and calls alarm(). Without this delegation, unclaimed permits left by an evicted instance wait for the next round. Read these source lines. Durable state: The alarm and the grant rows. In memory: A fresh helper with no waiting requests.
Your limit callbacks start over. In the example, groupLimit() reads workflowLimits, a Map on the instance. Caps set with limitWorkflow() are gone after eviction until set again; persist them if they must survive. Read these source lines. Durable state: Grant and lease rows are unaffected. In memory: Workflow caps held in the Map.
Reclaim unclaimed permits. The alarm pass shrinks each grant past its deadline to what was claimed, deletes it if that is all released, and emits grant_reclaimed after the commit. Rounds run the same reclaim first. Read these source lines. Durable state: The grant keeps only its claimed permits. In memory: The reclaim events until they are emitted.
A late claim is clamped. A client that claims after the deadline gets its claim clamped to what the grant still holds and status expired, so it requeues the items it could not admit. Read these source lines. Durable state: The claimed count never exceeds the grant. In memory: The result travels in the reply.
Claimed permits stay charged. A claimed permit counts until the client reports it released. The holder cannot stop external work; a client that never reports keeps that capacity in use, and a lowered limit only stops new admissions. Read these source lines. Durable state: Claimed, unreleased permits stay in the grant row. In memory: Nothing can reclaim them without a report.

Retire a lease nobody reports

A lease is a row with an expiry. A round inserts a lease row with its client, group, and permits, and an expiry ttlMs away, then returns the lease in the reply. The client draws from it locally, so the holder only learns what was drawn from reports. Read these source lines. Durable state: The lease row and its full footprint. In memory: Nothing; what the client drew is unknown until it reports.
Reports keep the lease alive. Every request or sync that reports a live lease moves its expiry ttlMs forward and advances the released count, clamped to what was drawn. Read these source lines. Durable state: The refreshed expiry and released count. In memory: The echoed lease in the reply.
An unreported lease is retired, not swept. The next round marks an expired lease retired: no more draws, but its whole footprint stays counted, because its client may be cut off with work still running. This happens in a round; the alarm does not watch lease expiry. Read these source lines. Durable state: retired = 1; the footprint is unchanged. In memory: The lease_retired event.
The client’s final report settles it. Only the client’s own retired report shrinks the lease to what it drew. Once every drawn permit is released, the row is deleted. Read these source lines. Durable state: The lease row shrinks, then disappears. In memory: Nothing.

Behind the scenes

Durable state: grant tables record client/group ownership, permits, cumulative claimed/released counts, and expiry. Optional lease tables account for locally drawable capacity. The request array awaiting a round is in memory, not a persisted work queue.

An allocation round first applies cumulative reports transactionally, reclaims expired unclaimed grants, and allocates grants in oldest-demand order. High-water marks only move forward and are clamped, so repeated or reordered reports cannot return the same capacity twice.

After failure: eviction before allocation loses the waiting RPC; clients must request again. A response lost after allocation does not free capacity. Unclaimed grant permits expire; confirmed claims stay charged until released. Lowering a limit stops new admissions rather than revoking running work.

Leases: the holder durably accounts for capacity that clients may draw locally. Strict provisioning counts grants and lease footprint against one limit. Expiry/retirement must not simply forget potentially running work. A permit is an admission/accounting contract, not a mechanism that can forcibly stop an external task.

Use the account concurrency composition for claim/release lifecycle, retry-safe admission, and reconciliation after a lost release notification.

Implementation: semaphore.ts, semaphore-server.ts, and semaphore-store.ts.

Semaphore (experimental)

Everything under durability/experimental/semaphore may change or be removed in a minor release; pin an exact version when depending on it. It is a separate entry point from durability/experimental because its client resolves Durable Object namespaces through cloudflare:workers.

DurabilitySemaphore holds one concurrency limit — say, an account’s maximum running jobs — inside an application-owned Durable Object, and hands permits to many client Durable Objects that queue work. Requests arriving within a round are allocated together in arrival order, so the oldest queued work across all clients is admitted first. Groups (a workflow, a tenant) may carry their own ceiling under the shared limit.

import { DurableObject } from 'cloudflare:workers';
import { timeout } from '@durability/transforms';
import { DurabilityScheduler } from 'durability';
import {
  DurabilitySemaphore,
  type SemaphoreRequest,
} from 'durability/experimental/semaphore';

/** One instance per account; the account id is the semaphore key. */
export class AccountLimit extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
  private readonly semaphore = new DurabilitySemaphore({
    scheduler: this.scheduler,
    limit: () => this.env.ACCOUNT_CONCURRENCY,
    groupLimit: (workflowId) => this.workflowLimits.get(workflowId),
  });

  permits(message: SemaphoreRequest) {
    return this.semaphore.handle(message);
  }

  override alarm(info?: AlarmInvocationInfo) {
    return this.scheduler.alarm(info);
  }
}

export const limits = DurabilitySemaphore.client({
  target: AccountLimit, // Vite injects exportName
  invoke: (stub, message) => stub.permits(message),
});

// From a client Durable Object:
const grant = await limits.with(timeout, 5_000).request(accountId, {
  type: 'request',
  v: 1,
  clientId: this.ctx.id.toString(),
  demand: [{ group: workflowId, count: queued, oldestAt: oldestEnqueuedAt }],
  grants: [{ grantId, claimed: 3, released: 1 }], // cumulative reports
});

Rounds. handle() holds a request for up to roundWaitMs (default 1 second) so concurrent requests are allocated together, and runs rounds no closer than roundPeriodMs (default 2 seconds). The response’s nextRequestAt tells the client when it may ask again.
Grants and reports. Each (client, group) that receives permits gets a grant. Clients report claimed (permits actually used) and released (claimed permits returned) per grant as cumulative high-water marks, piggybacked on requests or sent with a sync message. Counts only move forward and are clamped to the grant, so lost, repeated, or reordered reports can never free capacity twice. Permits a client never claims within grantTimeoutMs (default 10 seconds) are reclaimed; a later claim is answered with the clamped count and status: 'expired'.
Blocked attribution. Demand that stays queued is explained per group: group_limit, limit, not_permitted (canGrant returned false), or limit_resolution_error (groupLimit threw; the group fails closed).
Limits are live. limit and groupLimit are resolved once per round. Lowering a limit never evicts running work; it stops new admissions until usage is below the new ceiling.
Client side. DurabilitySemaphore.client() has the same shape as DurabilityRouting.client(): the Vite plugin injects exportName, namespace is the escape hatch, and caller transforms such as timeout compose. The underlying SemaphoreClient is transport-independent and remembers each key’s nextRequestAt.

usage() reports the limit, outstanding permits, lease footprint, and per-group usage. DurabilitySemaphore.migrate(this.ctx, null) removes the schema.

Leases

Rounds cost a round trip per batch of admissions. For bursty arrivals, the holder can hand each (client, group) with queued demand a lease: a block of permits the client may draw locally, one per item, with no round trip. Where that capacity comes from is the provisioning mode:

new DurabilitySemaphore({
  scheduler,
  limit: 100,
  leases: {
    provisioning: { mode: 'strict' }, // leases share the limit with grants
    // provisioning: { mode: 'under', reserve: 20 },   // 20 of the 100 kept for leases; grants get 80
    // provisioning: { mode: 'over', extra: { ratio: 0.5 } }, // 50 more above the limit, for leases only
    permitsPerLease: 64,
    ttlMs: 30 * 60_000,
  },
});

strict — “always below”: grants are allocated first, then leases from whatever is left; a live lease’s footprint (permits not yet released) is reserved from the same pool. Total never exceeds the limit.
under — a reserve inside the limit is kept for leases; grants may use at most limit - reserve. Still never exceeds the limit, and leases have guaranteed headroom. The reserve never takes the last permit: with a reserve at or above the limit (including after lowering the limit), grants keep one, so groups excluded from leases and clients without lease support still make progress.
over — leases come from a separate budget above the limit (the original gatekeeper behaviour); the effective ceiling is limit + extra.

Leases are issued only to clients that report leases (the queue always does), to groups without their own limit unless limitedGroups: 'withinGroupLimit', and only to a client that holds no undrawn permits for that group. Each round shares the remaining lease budget equally among candidates, oldest demand first, up to permitsPerLease; with fewer permits left than candidates, the oldest get one each. When a round leaves demand unserved, the holder marks its replies contended, and each client returns the undrawn permits of leases in groups where it has nothing queued, so an idle lease cannot keep other clients waiting. A client that stops talking to the holder keeps its lease until ttlMs. Reports (drawn, released high-water marks) heartbeat a lease, and the queue reports a lease with permits in use at half its lifetime even when nothing else is happening, so long-running work keeps its permits counted; a lease unreported for ttlMs is retired rather than forgotten: its client may be cut off from the holder with work still running, so, like a claimed grant permit, its footprint stays counted until the client reports what it drew and released. The queue keeps reporting a stopped lease until the holder settles it. A report for a lease the holder never knew is answered retired and its live footprint charged against that round. On the client, a drawn permit is one-shot: releasing the item returns footprint to the holder but never makes the permit drawable again. Lowering a group’s eligibility, or the limit, retires leases on the next round. A retired lease stops drawing at once but stays counted in full until its client confirms how much it drew, since it may have drawn after its last report.