/** * LaneQueue — per-lane FIFO queue for serialising async work. * * Each "lane" (keyed by session ID) processes work items one at a time. * If a lane is idle, work starts immediately. If it's busy, the work * is queued and a promise is returned that resolves when it's this * entry's turn to execute. * * Independent lanes run in parallel — only items within the same lane * are serialised. */ interface QueueEntry { work: () => Promise; resolve: (value: T) => void; reject: (reason: unknown) => void; policy: LaneQueueConfig; metadata?: LaneQueueEnqueueMetadata; } interface Lane { active: boolean; queue: QueueEntry[]; debounceTimer?: ReturnType; } export type LaneQueueMode = 'collect' | 'followup' | 'steer' | 'steer_backlog' | 'interrupt'; export type LaneQueueOverflow = 'drop_old' | 'drop_new'; export interface LaneQueueConfig { mode: LaneQueueMode; cap: number; overflow: LaneQueueOverflow; debounceMs: number; summarizeOverflow: boolean; } export interface LaneQueueEnqueueMetadata { requestId?: string; label?: string; } export interface LaneQueueEnqueueOptions { policy?: Partial; metadata?: LaneQueueEnqueueMetadata; } export type LaneQueueRejectCode = 'superseded' | 'overflow' | 'cancelled'; export interface LaneQueueRejectDetails { code: LaneQueueRejectCode; laneId: string; mode: LaneQueueMode; overflow?: LaneQueueOverflow; droppedCount?: number; message: string; } export class LaneQueueRejectedError extends Error { readonly details: LaneQueueRejectDetails; constructor(details: LaneQueueRejectDetails) { super(details.message); this.name = 'LaneQueueRejectedError'; this.details = details; } } export class LaneQueue { private lanes: Map = new Map(); private config: LaneQueueConfig; constructor(config?: Partial) { this.config = { mode: config?.mode ?? 'collect', cap: Math.max(1, config?.cap ?? 50), overflow: config?.overflow ?? 'drop_old', debounceMs: Math.max(0, config?.debounceMs ?? 0), summarizeOverflow: config?.summarizeOverflow ?? true, }; } /** * Enqueue a unit of work for the given lane. * Returns a promise that resolves with the work's return value * once it has been executed (which may be immediately if the lane is idle). */ async enqueue( laneId: string, work: () => Promise, policyOrOptions?: Partial | LaneQueueEnqueueOptions, ): Promise { const options = this.normalizeEnqueueOptions(policyOrOptions); const effective = this.resolvePolicy(options.policy); let lane = this.lanes.get(laneId); if (!lane) { lane = { active: false, queue: [] }; this.lanes.set(laneId, lane); } // If nothing is running on this lane, execute immediately if (!lane.active && !lane.debounceTimer) { lane.active = true; try { return await work(); } finally { lane.active = false; this.processNext(laneId); } } if (effective.mode === 'steer' || effective.mode === 'steer_backlog' || effective.mode === 'interrupt') { this.rejectPending(laneId, lane, { code: 'superseded', laneId, mode: effective.mode, message: 'Superseded by newer request', }); } else if (effective.mode === 'followup' && lane.queue.length > 0) { this.rejectPending(laneId, lane, { code: 'superseded', laneId, mode: effective.mode, message: 'Superseded by newer follow-up request', }); } if (lane.queue.length >= effective.cap) { if (effective.overflow === 'drop_new') { return Promise.reject( new LaneQueueRejectedError({ code: 'overflow', laneId, mode: effective.mode, overflow: 'drop_new', droppedCount: 1, message: effective.summarizeOverflow ? `Lane queue full (drop_new): request rejected with ${lane.queue.length} pending` : 'Lane queue full (drop_new)', }), ); } // drop_old const dropped = lane.queue.shift(); dropped?.reject( new LaneQueueRejectedError({ code: 'overflow', laneId, mode: effective.mode, overflow: 'drop_old', droppedCount: 1, message: effective.summarizeOverflow ? 'Lane queue overflow (drop_old): oldest pending request dropped' : 'Lane queue overflow (drop_old)', }), ); } // Otherwise, queue the work and return a deferred promise return new Promise((resolve, reject) => { lane.queue.push({ work: work as () => Promise, resolve: resolve as (value: unknown) => void, reject, policy: effective, metadata: options.metadata, }); }); } /** Check if a lane currently has active work executing. */ isProcessing(laneId: string): boolean { const lane = this.lanes.get(laneId); return (lane?.active ?? false) || Boolean(lane?.debounceTimer); } /** Get the number of pending (not yet started) items in a lane. */ queueLength(laneId: string): number { return this.lanes.get(laneId)?.queue.length ?? 0; } /** Get the total number of pending items across all lanes. */ totalPending(): number { let total = 0; for (const lane of this.lanes.values()) { total += lane.queue.length; } return total; } /** * Cancel all pending entries in a lane. * Active work is NOT interrupted — only queued items are rejected. * Rejected promises receive an Error with message "Lane cancelled". */ cancel(laneId: string): void { const lane = this.lanes.get(laneId); if (!lane) {return;} if (lane.debounceTimer) { clearTimeout(lane.debounceTimer); lane.debounceTimer = undefined; } this.rejectPending(laneId, lane, { code: 'cancelled', laneId, mode: this.config.mode, message: 'Lane cancelled', }); // Clean up empty idle lanes if (!lane.active && lane.queue.length === 0 && !lane.debounceTimer) { this.lanes.delete(laneId); } } private rejectPending(laneId: string, lane: Lane, details: LaneQueueRejectDetails): void { const pending = lane.queue.splice(0); for (const entry of pending) { entry.reject(new LaneQueueRejectedError({ ...details, laneId, mode: entry.policy.mode })); } } private resolvePolicy(policy?: Partial): LaneQueueConfig { return { mode: policy?.mode ?? this.config.mode, cap: Math.max(1, policy?.cap ?? this.config.cap), overflow: policy?.overflow ?? this.config.overflow, debounceMs: Math.max(0, policy?.debounceMs ?? this.config.debounceMs), summarizeOverflow: policy?.summarizeOverflow ?? this.config.summarizeOverflow, }; } /** * Process the next queued entry for a lane (called after current work finishes). * Runs asynchronously so the caller's finally block completes first. */ private processNext(laneId: string, skipDebounce = false): void { const lane = this.lanes.get(laneId); if (!lane) {return;} if (lane.active || lane.debounceTimer) { return; } const next = lane.queue[0]; if (!next) { this.lanes.delete(laneId); return; } if (!skipDebounce && next.policy.debounceMs > 0) { lane.debounceTimer = setTimeout(() => { lane.debounceTimer = undefined; this.processNext(laneId, true); }, next.policy.debounceMs); return; } const entry = lane.queue.shift(); if (!entry) { // Lane is empty — clean up this.lanes.delete(laneId); return; } lane.active = true; entry.work() .then((value) => entry.resolve(value)) .catch((err) => entry.reject(err)) .finally(() => { lane.active = false; this.processNext(laneId); }); } private normalizeEnqueueOptions( policyOrOptions?: Partial | LaneQueueEnqueueOptions, ): LaneQueueEnqueueOptions { if (!policyOrOptions) { return {}; } if ('policy' in policyOrOptions || 'metadata' in policyOrOptions) { return policyOrOptions as LaneQueueEnqueueOptions; } return { policy: policyOrOptions as Partial, }; } }