Something went wrong. Try again.
I'm going to build my own OpenClaw, with blackjack... and bun!
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208/** * queue.ts – Lane-aware execution queue with retry for agent runs. * * Ensures only one task runs at a time per logical lane (for example one chat), * while allowing different lanes to make progress in parallel. Failed tasks are * retried with exponential back-off up to DEFAULT_MAX_RETRIES times. * Items can be deduplicated by an optional string `id` (useful for scheduled * tasks that shouldn't stack up). * * Consumers: * - runtime.ts creates a single AgentQueue and passes it to the router * and task scheduler. * - router.ts calls enqueue() for each inbound message that needs processing. * - task-scheduler.ts calls enqueueTask() for due scheduled tasks. * - channels/web.ts calls enqueue() for web channel interactions. */
import { DEFAULT_BASE_RETRY_MS, DEFAULT_MAX_RETRIES, getRetryDelay, shouldRetry } from "./queue/retry-policy.js";import { createLogger } from "./utils/logger.js";
const log = createLogger("queue");const DEFAULT_LANE_KEY = "__default__";
/** Internal representation of a queued work item. */interface QueueItem { /** Optional deduplication key. */ id?: string; /** The async function to execute. */ fn: () => Promise<void>; /** Logical lane key; tasks in the same lane are serialized. */ laneKey: string; /** Number of times this item has been retried after failure. */ retries: number;}
interface LaneState { running: boolean; pending: QueueItem[]; current: QueueItem | null; runningPromise: Promise<void> | null;}
/** In-memory counters for queue behavior/health visibility. */export interface QueueMetrics { enqueued: number; deduplicated: number; started: number; succeeded: number; failed: number; retriesScheduled: number;}
/** * A lane-aware execution queue: items run FIFO within each lane, while * unrelated lanes may execute concurrently. */export class AgentQueue { private lanes = new Map<string, LaneState>(); private shuttingDown = false; private metrics: QueueMetrics = { enqueued: 0, deduplicated: 0, started: 0, succeeded: 0, failed: 0, retriesScheduled: 0, };
private getLane(laneKey?: string): LaneState { const resolved = laneKey || DEFAULT_LANE_KEY; let lane = this.lanes.get(resolved); if (!lane) { lane = { running: false, pending: [], current: null, runningPromise: null }; this.lanes.set(resolved, lane); } return lane; }
private removeLaneIfIdle(laneKey: string): void { const lane = this.lanes.get(laneKey); if (!lane) return; if (lane.running) return; if (lane.current) return; if (lane.pending.length > 0) return; if (lane.runningPromise) return; this.lanes.delete(laneKey); }
private hasQueuedId(id: string): boolean { for (const lane of this.lanes.values()) { if (lane.current?.id === id) return true; if (lane.pending.some((item) => item.id === id)) return true; } return false; }
/** * Add a work item to the queue. * Items in the same lane run sequentially; items in different lanes may run * concurrently. Items with the same `id` are deduplicated globally. */ enqueue(fn: () => Promise<void>, id?: string, laneKey?: string): void { if (this.shuttingDown) return;
if (id && this.hasQueuedId(id)) { this.metrics.deduplicated += 1; return; }
const resolvedLaneKey = laneKey || DEFAULT_LANE_KEY; const lane = this.getLane(resolvedLaneKey); const item: QueueItem = { id, fn, laneKey: resolvedLaneKey, retries: 0 }; this.metrics.enqueued += 1;
if (lane.running) { lane.pending.push(item); return; }
this.runItem(lane, item); }
/** Convenience wrapper that prefixes the id with "task:" for scheduled tasks. */ enqueueTask(taskId: string, fn: () => Promise<void>, laneKey?: string): void { this.enqueue(fn, `task:${taskId}`, laneKey); }
/** Start executing an item inside a lane. */ private runItem(lane: LaneState, item: QueueItem): void { lane.running = true; lane.current = item; this.metrics.started += 1; lane.runningPromise = this.executeItem(lane, item); }
/** Run the item's function, handle errors with retry, and advance within the lane. */ private async executeItem(lane: LaneState, item: QueueItem): Promise<void> { try { await item.fn(); this.metrics.succeeded += 1; } catch (error) { this.metrics.failed += 1; log.error("Queue item failed", { operation: "execute_item", laneKey: item.laneKey, itemId: item.id ?? null, err: error, }); this.scheduleRetry(item); } finally { lane.running = false; lane.current = null; lane.runningPromise = null; if (lane.pending.length > 0 && !this.shuttingDown) { const next = lane.pending.shift()!; this.runItem(lane, next); } else { this.removeLaneIfIdle(item.laneKey); } } }
/** Schedule an exponential-backoff retry for a failed item. */ private scheduleRetry(item: QueueItem): void { if (!shouldRetry(item.retries, DEFAULT_MAX_RETRIES, this.shuttingDown)) return; item.retries++; this.metrics.retriesScheduled += 1; const delay = getRetryDelay(item.retries, DEFAULT_BASE_RETRY_MS); log.info("Scheduling queue retry", { operation: "schedule_retry", retries: item.retries, maxRetries: DEFAULT_MAX_RETRIES, delayMs: delay, itemId: item.id ?? null, }); setTimeout(() => { if (this.shuttingDown) return; const lane = this.getLane(item.laneKey); if (lane.running) { lane.pending.push(item); } else { this.runItem(lane, item); } }, delay); }
/** * Gracefully shut down the queue: clear pending items and wait up to `ms` * milliseconds for currently running lane tasks to finish. */ async shutdown(ms = 5000): Promise<void> { this.shuttingDown = true; const runningPromises: Promise<void>[] = []; for (const lane of this.lanes.values()) { lane.pending = []; if (lane.runningPromise) runningPromises.push(lane.runningPromise); } if (runningPromises.length > 0) { await Promise.race([Promise.allSettled(runningPromises), Bun.sleep(ms)]); } }
/** Snapshot queue counters for diagnostics and tests. */ getMetrics(): QueueMetrics { return { ...this.metrics }; }}