/**
* Bounded per-site work scheduling primitives.
*/
class ConcurrencyGate {
private active = 0
private readonly waiters: Array<() => void> = []
constructor(private readonly maxConcurrency: number) {
if (!Number.isSafeInteger(maxConcurrency) || maxConcurrency < 1) {
throw new Error('maxConcurrency must be a positive safe integer')
}
}
get inUse(): number {
return this.active
}
async run(handler: () => Promise<void>): Promise<void> {
await this.acquire()
try {
await handler()
} finally {
this.release()
}
}
private acquire(): Promise<void> {
if (this.active < this.maxConcurrency) {
this.active++
return Promise.resolve()
}
return new Promise((resolve) => this.waiters.push(resolve))
}
private release(): void {
const next = this.waiters.shift()
if (next) {
// Hand the occupied slot directly to the next task.
next()
return
}
this.active--
}
}
export interface SchedulerDrainResult {
outcome: 'drained' | 'forced'
forced: boolean
remainingWork: number
activeHandlers: number
}
/**
* A per-site FIFO scheduler with a shared concurrency limit. The returned
* promise preserves the handler failure for the cursor tracker, while the
* internal settled queue lets a later event for that site run in order and be
* recovered by replay if necessary.
*/
export class SiteWorkScheduler {
private readonly siteQueues = new Map<string, Promise<void>>()
private readonly work = new Set<Promise<void>>()
private readonly gate: ConcurrencyGate
private queued = 0
constructor(maxConcurrency: number) {
this.gate = new ConcurrencyGate(maxConcurrency)
}
get queuedHandlers(): number {
return this.queued
}
get activeHandlers(): number {
return this.gate.inUse
}
schedule(siteKey: string, handler: () => Promise<void>): Promise<void> {
const previous = this.siteQueues.get(siteKey) ?? Promise.resolve()
this.queued++
const task = previous.catch(() => undefined).then(() => this.gate.run(handler))
// Do not make one failed event permanently block later work for the same
// site. The caller still observes task's rejection and keeps its cursor
// reservation pending until the relay replays it.
let settled: Promise<void>
const finish = () => {
this.queued = Math.max(0, this.queued - 1)
this.work.delete(settled)
if (this.siteQueues.get(siteKey) === settled) this.siteQueues.delete(siteKey)
}
settled = task.then(
() => finish(),
() => finish(),
)
this.siteQueues.set(siteKey, settled)
this.work.add(settled)
return task
}
async drain(gracePeriodMs: number): Promise<SchedulerDrainResult> {
if (this.work.size === 0) return this.drainResult('drained')
if (gracePeriodMs <= 0) return this.drainResult('forced')
const didDrain = await new Promise<boolean>((resolve) => {
const timer = setTimeout(() => resolve(false), gracePeriodMs)
void this.waitForIdle().then(() => {
clearTimeout(timer)
resolve(true)
})
})
return this.drainResult(didDrain ? 'drained' : 'forced')
}
private async waitForIdle(): Promise<void> {
while (this.work.size > 0) {
await Promise.all(this.work)
}
}
private drainResult(outcome: SchedulerDrainResult['outcome']): SchedulerDrainResult {
return {
outcome,
forced: outcome === 'forced',
remainingWork: this.queued,
activeHandlers: this.gate.inUse,
}
}
}