Queue
Durable, pluggable job queue with priority-aware leasing, retry + deterministic backoff, graceful shutdown, parent/child flows, and a dead-letter surface for jobs that exhaust their retries.
Same dispatcher shape as b.objectStore: every operator-named backend declares a protocol plus protocol-specific options. The built-in local protocol is SQLite-backed (rows live in the framework's main DB so persistence survives crashes / restarts without external infrastructure), and can be pointed at an operator's own database handle, table, and schema via the local config (db / table / schema). redis and sqs ship; amqp and nats are listed as deferred and surface a clear error if selected.
local and redis are driven by the generic b.queue.consume loop and the lifecycle below (framework-side leasing, deterministic backoff, DLQ, and the sweep timer). sqs is an SQS-native adapter with a different model: complete / fail delete or re-deliver by the message's receiptHandle (returned by lease(), threaded back by the caller), and DLQ + visibility-expiry are handled server-side by the SQS queue's RedrivePolicy — so sqs is driven directly (lease → handle → complete/fail), not by b.queue.consume, and it does not use the framework DLQ / sweep described below. See lib/queue-sqs.js for its action map and the features that require operator wiring.
Job lifecycle: enqueued (status='pending', availableAt set by delaySeconds) ↓ availableAt reached + consumer leases inflight (status='inflight', lease expires after leaseDurationMs) ↓ handler returns ↓ handler throws done (status='done') attempts < maxAttempts: pending (with deterministic backoff) else: failed → DLQ row written
A 30-second sweep timer re-pends inflight rows whose lease expired without completion (crashed handlers, OOM kills) so no job is abandoned. Within a single millisecond, higher priority jobs lease before lower-priority ones (deterministic — see b.queue.enqueue opts).
Dead-letter handling: jobs that exhaust maxAttempts write a system.queue.dlq.write audit event and stay queryable via b.queue.dlqList. Operator decides whether to retry (b.queue.dlqRetry) — never automatic.
b.queue.init(opts) #
{
backends: {
[name: string]: {
protocol: "local" | "redis" | "sqs",
breaker?: { ... }, // see b.retry.CircuitBreaker opts
retry?: { ... }, // see b.retry.withRetry opts
// local protocol — bring-your-own database (all optional):
db?: object, // store handle (execute/executeOne/executeAll); default cluster-storage
table?: string, // table name (validated + quoted); default "_blamejs_jobs"
schema?: string, // schema/namespace qualifier (validated + quoted)
// ...other protocol-specific opts (e.g. redis url, sqs queueUrl)
},
},
defaultBackend?: string, // name to use when enqueue/consume omit { backend }
sweepIntervalMs?: number, // expired-lease sweep period; default 30s, 0 = no timer
}One-time initialization. Wires every named backend through the protocol dispatcher, wraps mutating ops with the retry helper + circuit breaker, and starts the 30-second expired-lease sweep. Idempotent — calling init after the queue is already initialized is a no-op (boot order doesn't have to be exact).
Throws when opts.backends is missing — operators catch the typo at boot rather than discovering it on first enqueue.
The local protocol defaults to the framework's own database (the main SQLite in single-node mode, the operator-supplied external DB in cluster mode) and the _blamejs_jobs table. An operator who wants the queue rows to live in their own database, table, or schema supplies db / table / schema in the local backend config. The db handle must expose the same execute / executeOne / executeAll surface as b.clusterStorage; table / schema are validated as SQL identifiers and quoted through b.safeSql (an identifier that isn't a safe name is refused at init time, not interpolated into SQL). Sealed columns (payload, lastError) stay sealed regardless of where the rows land.
b.queue.init({
backends: {
primary: { protocol: "local" },
app: { protocol: "local", table: "app_jobs", schema: "work" },
},
defaultBackend: "primary",
});
b.queue.listBackends();
// → [{ name: "primary", protocol: "local", breakerState: "closed" }, ...]b.queue.enqueue(queueName, payload, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
priority?: number, // higher leases first (default 0)
delaySeconds?: number, // park before becoming leaseable
maxAttempts?: number, // retries before DLQ (backend default applies)
classification?: string, // operator metadata, surfaced in audit
traceId?: string, // cross-request correlation id
}Persists a single job to the named queue and returns a promise that resolves with the assigned jobId. The job's payload is stored verbatim by the backend (the local protocol JSON-encodes; redis and sqs follow their wire formats). Resolves before any consumer actually leases the job — enqueue is durable handoff, not synchronous execution.
Higher priority jobs lease ahead of lower ones within the same availableAt window. delaySeconds parks the job until the timestamp arrives. maxAttempts overrides the queue default; on the final attempt the job moves to the dead-letter view rather than retrying again.
var result = await b.queue.enqueue("ingest", { url: "https://example.com" }, {
priority: 5,
maxAttempts: 3,
});
result.jobId;
// → "job-7c2f8e1a..."b.queue.consume(queueName, handler, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
concurrency?: number, // max in-flight handlers (default 1)
leaseDurationMs?: number, // lease window before sweep re-pends (default 30s)
pollIntervalMs?: number, // idle backoff between empty leases (default 1s)
fastPollMs?: number, // delay between non-empty lease batches (default 50ms)
rateLimit?: {
max: number, // positive integer
perSeconds: number, // positive finite seconds
},
}Starts a long-running consumer that leases jobs and runs them through handler(job, ctx). Handler resolution marks the job done; rejection bumps the attempt counter and either re-pends with deterministic exponential backoff (1s base, 5min cap, no jitter) or routes to the DLQ when attempts >= maxAttempts. This loop drives the local and redis backends; the sqs backend uses SQS-native receipt-handle complete/fail and server-side redrive and is driven directly rather than by this consumer (see the module intro).
Returns a consumer state handle whose .cancel() aborts the poll loop immediately (without waiting for the next pollIntervalMs tick) and stops leasing new work. In-flight handlers complete on their own; b.queue.shutdown waits for them with a deadline.
ctx carries extendLease(additionalMs) for long-running handlers about to overrun their lease, and progress(0..100) for audit-chain progress markers (rate-limited so a chatty handler can't flood the chain).
var consumer = b.queue.consume("ingest", async function (job, ctx) {
ctx.progress(10);
// ...do work...
ctx.progress(100);
}, { concurrency: 4 });
// Later, on shutdown signal:
consumer.cancel();b.queue.tick(opts) #
{
queue: string, // required — the queue name to drain
handler: function, // required — (job, ctx) => Promise, same as consume
max: number, // default: 10 — most jobs to lease this tick
leaseMs: number, // default: 30s — lease duration for the batch
concurrency: number, // default: 1 — jobs run in parallel
backend: string, // named backend; default the init default
}Drain what is due, then return. Leases at most max available jobs, runs handler over them, applies the same leasing, deterministic backoff and dead-letter transitions b.queue.consume applies, and settles. Nothing is left running when the promise resolves: no resident loop, no timer, no consumer registered for shutdown to wait on.
Pair it with b.queue.init({ sweepIntervalMs: 0 }). init otherwise starts a 30-second expired-lease sweep timer, and in a frozen isolate that timer either never fires or fires inside a later invocation, touching the backend outside any request the platform is accounting for. tick sweeps once per invocation itself, so the timer would repeat a sweep that has already run.
This is the driver for a scheduled invocation — a Cloudflare scheduled() handler, a Lambda on EventBridge, a Kubernetes CronJob. consume cannot serve those: it is a resident loop and it starts a background sweep, and such a runtime freezes between invocations, so the timer never fires and the consumer is either blocking the handler or killed mid-lease. Without a per-tick entry point the durable half — attempt accounting, backoff schedule, DLQ transition — gets reimplemented by the caller, which is the part that is genuinely hard to get right.
Returns { leased, succeeded, failed, stale, unsettled, movedToDlq, mayHaveMore, queueDepth }.
Two of those counts are not terminal states and mean different things. stale is benign: the backend answered that this worker no longer owns the job, so someone else is running it. unsettled is not: the backend REFUSED to record the outcome — complete or fail rejected after its own retries — so the job is still inflight and this tick does not know whether the handler succeeded. Those are recovered by a later sweep, and if the handler had already succeeded its side effects run a second time, so a non-zero unsettled is worth alerting on rather than counting as noise.
mayHaveMore is the tick-again signal: the batch came back full, so more may be due right now. Loop on that, not on a depth count — queueDepth is the queue's total active depth (pending plus inflight) and includes jobs whose availableAt is still in the future, so a caller looping on it would spin through empty ticks waiting for a delayed job to come due. queueDepth is null when the backend could not be asked: unknown is not empty.
stale counts jobs this tick ran but did not settle: the handler outlived the lease, the job was swept and re-leased, and the backend's attempt fence refused the completion. Those are not failures — another worker owns them — and a non-zero stale means leaseMs is too short for the handler, or the batch too large to finish inside it.
Jobs run sequentially by default. Pass concurrency to overlap them; the lease is held for leaseMs either way, so a batch that will take longer than the lease should either raise leaseMs or call ctx.extendLease(ms) from the handler.
var result = await b.queue.tick({
queue: "webhooks",
max: 50,
handler: async function (job) { await deliver(job.payload); },
});
while (result.mayHaveMore) result = await b.queue.tick({ queue: "webhooks", max: 50, handler: deliver });
// → { leased: 50, succeeded: 48, failed: 2, stale: 0, unsettled: 0,
// movedToDlq: 1, mayHaveMore: true, queueDepth: 62 }b.queue.size(queueName, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
}Resolves with the number of pending + inflight jobs in the queue — the live backlog. Excludes done and failed rows. Operators wire this to dashboards and autoscalers.
var pending = await b.queue.size("ingest");
// → 42b.queue.purge(queueName, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
}Deletes every job in the named queue and resolves with the deleted count. Emits a system.queue.purge audit event for forensic traceability. Use during operator-driven cleanups; never in normal traffic — purged jobs are not recoverable.
var deleted = await b.queue.purge("ingest");
// → 42b.queue.dlqList(queueName, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
limit?: number, // backend-specific paging cap
}Resolves with an array of dead-letter rows — jobs that exhausted their retries and were parked for human review. Each row carries the original payload, attempt count, last failure reason, and trace correlation id. Rejects with queue/dlq-unsupported when the configured backend does not implement a dead-letter view, queue/not-initialized before b.queue.init(), and queue/unknown-backend for a backend name that was never registered.
var dead = await b.queue.dlqList("ingest", { limit: 50 });
dead.length;
// → 3b.queue.dlqRetry(jobId, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
}Resets a single dead-letter row back to pending so consumers pick it up again. Operator-driven only — the framework never auto-retries failed-after-retries jobs because the failure mode usually requires human investigation. Resolves with true when the row was found and reset, false otherwise.
var ok = await b.queue.dlqRetry("job-7c2f8e1a");
// → trueb.queue.dlqSize(queueName, opts) #
{
backend?: string, // backend name; defaults to defaultBackend
}Resolves with the number of dead-letter rows for the named queue — jobs that exhausted their retries and were parked for human review. Operators wire this to dashboards / alerting so a growing DLQ surfaces before it becomes a backlog. Rejects with queue/dlq-unsupported when the configured backend does not implement a dead-letter view, queue/not-initialized before b.queue.init(), and queue/unknown-backend for a backend name that was never registered.
var stuck = await b.queue.dlqSize("ingest");
// → 3b.queue.shutdown(opts) #
{
timeoutMs?: number, // drain deadline in ms (default 30000)
}Cancels every active consumer and waits for in-flight handlers to drain, then stops the expired-lease sweep timer. Honors a deadline — handlers that exceed timeoutMs are abandoned (their leases expire and the sweep re-pends them on the next process). Idempotent — calling shutdown before init is a no-op so SIGTERM handlers can be wired unconditionally.
process.on("SIGTERM", async function () {
await b.queue.shutdown({ timeoutMs: 15000 });
});b.queue.listBackends() #
Returns an array of { name, protocol, breakerState } rows — one per configured backend. breakerState is "closed" / "open" / "half-open" from the per-backend circuit breaker. Operators wire this to a /health/queue endpoint or readiness probe so a tripped breaker surfaces in the orchestrator before silent backlog growth.
Calling it before b.queue.init() throws queue/not-initialized.
var status = b.queue.listBackends();
// → [{ name: "primary", protocol: "local", breakerState: "closed" }]b.queue.enqueueFlow(spec) #
{
queueName: string,
children: [
{
name: string, // unique within the flow
payload: any,
dependsOn?: string[], // sibling names this child waits on
priority?: number,
maxAttempts?: number,
classification?: string,
traceId?: string,
},
...
],
}Atomically registers a parent-child job graph. Each child enqueues with a parking-lot availableAt = MAX_SAFE_INTEGER until every dependsOn row reaches done, at which point the dependent's availableAt drops to "now" and consumers pick it up. Cycle detection runs at registration time, so a bad graph rejects with queue/flow-cycle or queue/flow-unknown-dep before any row lands.
A malformed spec rejects with queue/bad-flow: a spec that is not an object, a missing queueName or children, a child that is not an object, a child name that is not a non-empty string, and a name repeated within the flow. A queue backend without patchFlowDeps cannot release a parked child, so it rejects with queue/flow-unsupported rather than registering a graph it could never advance. It also rejects with queue/not-initialized before b.queue.init(), and queue/unknown-backend for a backend name that was never registered.
Resolves with { flowId, jobs: [{ name, jobId }, ...] }. The returned jobId array is in declaration order, not topological order — callers that need a specific child's id look it up by name.
var flow = await b.queue.enqueueFlow({
queueName: "ingest",
children: [
{ name: "fetch", payload: { url: "https://example.com" } },
{ name: "transform", payload: { stage: 1 }, dependsOn: ["fetch"] },
{ name: "publish", payload: { topic: "out" }, dependsOn: ["transform"] },
],
});
flow.jobs.length;
// → 3b.queue.bootFromEnv(opts) #
{
env?: object, // override process.env for testing / fixtures
}Env-driven init mirroring b.network.bootFromEnv and b.logStream.bootFromEnv. Reads BLAMEJS_QUEUE_* and calls queue.init({ backends: { default: ... } }) so operators get a working queue without writing build-app code. Idempotent — a second call after init already ran is a no-op.
Recognized env vars: BLAMEJS_QUEUE_PROTOCOL "local" | "redis" (default "local") BLAMEJS_QUEUE_REDIS_URL redis://host:port/db (required when protocol=redis) BLAMEJS_QUEUE_REDIS_PASSWORD auth password BLAMEJS_QUEUE_REDIS_USERNAME ACL username BLAMEJS_QUEUE_REDIS_TLS "1" / "true" forces TLS (else inferred from rediss://) BLAMEJS_QUEUE_REDIS_KEY_PREFIX key prefix (default "blamejs:queue")
Throws queue/invalid-config when BLAMEJS_QUEUE_PROTOCOL is unknown or when redis is selected without BLAMEJS_QUEUE_REDIS_URL — operators catch the typo at boot rather than first enqueue.
process.env.BLAMEJS_QUEUE_PROTOCOL = "local";
b.queue.bootFromEnv();
b.queue.listBackends().length;
// → 1