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 }
}

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.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");
// → 42

b.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");
// → 42

b.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 DLQ_UNSUPPORTED when the configured backend does not implement a dead-letter view.

var dead = await b.queue.dlqList("ingest", { limit: 50 });
dead.length;
// → 3

b.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");
// → true

b.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 DLQ_UNSUPPORTED when the configured backend does not implement a dead-letter view.

var stuck = await b.queue.dlqSize("ingest");
// → 3

b.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.

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 — bad graphs reject with FLOW_CYCLE / FLOW_UNKNOWN_DEP before any row lands.

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;
// → 3

b.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 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

Last updated 2026-08-08T16:39:49.652Z by seeder.