Queues

Workflow queues allow you to ensure that workflow functions will be run, without starting them immediately. Queues are useful for controlling the number of workflows run in parallel, or the rate at which they are started.

Queue configuration is persisted to the system database, so any DBOS process or DBOSClient connected to the same system database can register, retrieve, and reconfigure queues.

Queue Management

DBOS.registerQueue

DBOS.registerQueue(
  name: string,
  options?: RegisterQueueOptions,
): Promise<WorkflowQueue>

interface RegisterQueueOptions {
  // Applied to the queue as a whole
  globalConcurrency?: number;
  workerConcurrency?: number;
  rateLimit?: QueueRateLimit;
  // Applied to each partition separately
  partitionConcurrency?: number;
  partitionWorkerConcurrency?: number;
  partitionRateLimit?: QueueRateLimit;
  minPollingIntervalMs?: number;
  onConflict?: QueueConflictResolution;
}

interface QueueRateLimit {
  limitPerPeriod: number;
  periodSec: number;
}

type QueueConflictResolution =
  | 'update_if_latest_version'
  | 'always_update'
  | 'never_update';

Register a queue and persist its configuration to the system database, returning a WorkflowQueue. DBOS must be launched before calling registerQueue. If the queue already exists in the database, the onConflict option controls whether its configuration is overwritten.

Parameters:

• name: The name of the queue. Must be unique among all queues in the application.
• globalConcurrency: The maximum number of workflows from this queue that may run concurrently across all DBOS processes. Defaults to no limit.
• workerConcurrency: The maximum number of workflows from this queue that may run concurrently within a single DBOS process. Must be less than or equal to globalConcurrency.
• rateLimit: A limit on the maximum number of functions which may be started in a given period.
  • rateLimit.limitPerPeriod: The number of workflows that may be started within the specified time period.
  • rateLimit.periodSec: The time period across which limitPerPeriod applies.
• partitionConcurrency: The maximum number of workflows from any one partition of this queue that may run concurrently across all DBOS processes. Must be at least 1 and less than or equal to globalConcurrency.
• partitionWorkerConcurrency: The maximum number of workflows from any one partition of this queue that may run concurrently within a single DBOS process. Must be at least 1 and less than or equal to partitionConcurrency, workerConcurrency, and globalConcurrency.
• partitionRateLimit: A limit on the maximum number of workflows which may be started from any one partition in a given period. Takes the same fields as rateLimit.
• minPollingIntervalMs: The minimum interval, in milliseconds, between dequeue attempts for this queue. Defaults to 1000ms. The actual polling interval includes random jitter and increases with backoff under contention, then scales back down when contention clears.
• onConflict: How to behave when a queue with this name already exists in the system database:
  • 'update_if_latest_version' (default): overwrite the existing configuration only if the running application is the latest registered application version. This prevents older versions in a rolling deploy from overwriting a newer configuration.
  • 'always_update': always overwrite the existing configuration.
  • 'never_update': leave the existing configuration unchanged. The returned queue reflects the persisted configuration, not the supplied parameters.

Setting any partition limit makes the queue partitioned: every enqueue must supply a queuePartitionKey, and deduplication is not supported. The queue-wide limits (globalConcurrency, workerConcurrency, and rateLimit) continue to apply across all partitions.

Queues are owned by the application (identified by its configured name) that registers them, and queue names are globally unique across all applications sharing a system database: if the queue is already registered by a different application, registerQueue throws an error regardless of onConflict.

Example syntax:

const queue = await DBOS.registerQueue("email", {
  globalConcurrency: 10,
  rateLimit: { limitPerPeriod: 100, periodSec: 60 },
});

DBOS.retrieveQueue

DBOS.retrieveQueue(name: string): Promise<WorkflowQueue | null>

Retrieve a queue by name from the system database, or null if no queue with that name has been registered.

Example syntax:

const queue = await DBOS.retrieveQueue("email");
if (queue !== null) {
  console.log(await queue.getGlobalConcurrency());
}

DBOS.listQueues

DBOS.listQueues(applicationName?: string | string[]): Promise<WorkflowQueue[]>

List all database-backed queues registered in the system database. Returns an empty list if no queues have been registered.

Parameters:

• applicationName: List only queues owned by this application (or one of these applications). Queues owned by no application are always included. If unset, list only this application's queues.

Example syntax:

for (const queue of await DBOS.listQueues()) {
  console.log(queue.name, queue.concurrency, queue.partitionConcurrency);
}

DBOS.deleteQueue

DBOS.deleteQueue(name: string): Promise<void>

Delete a queue from the system database. No-op if no queue with that name exists.

warning

Workflows already enqueued on a deleted queue can no longer be dequeued, executed, or recovered. However, if a queue with the same name is later registered, it will dequeue the leftover workflows. Do not rely on this: stale workflows unexpectedly resuming on a future queue is rarely the intended behavior. Instead, cancel or drain pending workflows on the queue before deleting it.

class WorkflowQueue

A WorkflowQueue is returned from DBOS.registerQueue, DBOS.retrieveQueue, or DBOS.listQueues. Its cached fields reflect the queue's configuration as of the most recent read from the database; use the get methods to refresh from the database, and the set methods to update.

class WorkflowQueue {
  readonly name: string;

  // Cached configuration. May be stale if another process has reconfigured
  // the queue. Use the get methods below to refresh from the database.
  concurrency?: number;  // The queue-wide (global) concurrency limit
  workerConcurrency?: number;
  rateLimit?: QueueRateLimit;
  partitionConcurrency?: number;
  partitionWorkerConcurrency?: number;
  partitionRateLimit?: QueueRateLimit;
  minPollingIntervalMs?: number;

  // The application that owns this queue; undefined for in-memory and unowned queues.
  applicationName?: string;

  // Read the latest values from the database.
  getGlobalConcurrency(): Promise<number | undefined>;
  getWorkerConcurrency(): Promise<number | undefined>;
  getRateLimit(): Promise<QueueRateLimit | undefined>;
  getPartitionConcurrency(): Promise<number | undefined>;
  getPartitionWorkerConcurrency(): Promise<number | undefined>;
  getPartitionRateLimit(): Promise<QueueRateLimit | undefined>;
  getMinPollingIntervalMs(): Promise<number | undefined>;

  // Update the configuration in the database.
  setGlobalConcurrency(value: number | undefined): Promise<void>;
  setWorkerConcurrency(value: number | undefined): Promise<void>;
  setRateLimit(value: QueueRateLimit | undefined): Promise<void>;
  setPartitionConcurrency(value: number | undefined): Promise<void>;
  setPartitionWorkerConcurrency(value: number | undefined): Promise<void>;
  setPartitionRateLimit(value: QueueRateLimit | undefined): Promise<void>;
  setMinPollingIntervalMs(value: number): Promise<void>;
}

A queue is partitioned if any of its partition limits is set. Each set method validates the new value against the queue's other limits: a per-partition concurrency limit must be less than or equal to its queue-wide counterpart, and partitionWorkerConcurrency must be less than or equal to partitionConcurrency. Pass undefined to any set method to remove that limit.

Reconfiguring Queues

Because queue configuration lives in the system database, you can change a queue's configuration at runtime without redeploying or restarting your workers. Workers pick up the new configuration on their next polling iteration.

const queue = await DBOS.retrieveQueue("email");
if (queue !== null) {
  await queue.setGlobalConcurrency(50);
  await queue.setRateLimit({ limitPerPeriod: 500, periodSec: 60 });
}

info

Setting any partition limit makes the queue partitioned; clearing all of them makes it unpartitioned again. Take care when partitioning a queue at runtime: workflows already enqueued on it have no partition key and will not be dequeued until the queue is unpartitioned.

The set methods may only be called on a queue returned from DBOS.registerQueue, DBOS.retrieveQueue, DBOS.listQueues, or the equivalent DBOSClient methods.

warning

If your application calls DBOS.registerQueue on startup, the next process to start can overwrite settings you applied at runtime via set methods. Either update the registerQueue call to match the new configuration, or pass onConflict: "never_update" to preserve the runtime changes.

Calling them on a queue created with the legacy new WorkflowQueue(...) constructor throws an error.

Enqueueing Workflows

Workflows are enqueued by providing a queueName argument to DBOS.startWorkflow. This enqueues a function for processing and returns a handle to it. Through arguments to DBOS.startWorkflow, you can optionally provide a custom priority or deduplication ID to an enqueued workflow.

The DBOS.startWorkflow method durably enqueues your function; after it returns, your function is guaranteed to eventually execute even if your app is interrupted.

Example syntax:

import { DBOS } from "@dbos-inc/dbos-sdk";

async function taskFunction(task) {
    // ...
}
const taskWorkflow = DBOS.registerWorkflow(taskFunction, {"name": "taskWorkflow"});

async function queueFunction(tasks) {
  await DBOS.registerQueue("example_queue");
  const handles = []

  // Enqueue each task so all tasks are processed concurrently.
  for (const task of tasks) {
    handles.push(await DBOS.startWorkflow(taskWorkflow, { queueName: "example_queue" })(task))
  }

  // Wait for each task to complete and retrieve its result.
  // Return the results of all tasks.
  const results = []
  for (const h of handles) {
    results.push(await h.getResult())
  }
  return results
}
const queueWorkflow = DBOS.registerWorkflow(queueFunction, {"name": "taskWorkflow"})

Example syntax using decorated workflows:

import { DBOS } from "@dbos-inc/dbos-sdk";

class Tasks {
  @DBOS.workflow()
  static async processTask(task) {
    // ...
  }

  @DBOS.workflow()
  static async processTasks(tasks) {
    await DBOS.registerQueue("example_queue");
    const handles = []

    // Enqueue each task so all tasks are processed concurrently.
    for (const task of tasks) {
      handles.push(await DBOS.startWorkflow(Tasks, { queueName: "example_queue" }).processTask(task));
    }

    // Wait for each task to complete and retrieve its result.
    // Return the results of all tasks.
    const results = [];
    for (const h of handles) {
      results.push(await h.getResult());
    }
    return results;
  }
}

DBOS.enqueueWorkflowWithOptions

DBOS.enqueueWorkflowWithOptions<T = unknown>(
  options: EnqueueWorkflowOptions,
  ...args: unknown[]
): Promise<WorkflowHandle<T>>

Enqueue a workflow by name, without a reference to its function. Takes the same options as DBOSClient.enqueue (except duplicationPolicy), so the workflow may be implemented by another process or another application, as long as it shares this system database. Can safely be called from inside a workflow: the enqueued workflow is recorded as a child of the calling workflow.

Unlike DBOS.startWorkflow, options are not validated against the local registry, and appVersion is left unset unless given (an unset appVersion is only dequeued by an executor running the latest registered application version). The enqueued workflow is owned by this application unless the applicationName option names another one, in which case that application dequeues and runs it.

DBOS.enqueueWorkflowWithOptionsPortable

DBOS.enqueueWorkflowWithOptionsPortable<T = unknown>(
  options: EnqueueWorkflowOptions,
  positionalArgs: unknown[],
  namedArgs?: Record<string, unknown>,
): Promise<WorkflowHandle<T>>

Like enqueueWorkflowWithOptions, but serializes arguments in portable format for target workflows that take named arguments (for example, a Python workflow with keyword arguments).

Legacy: In-Memory Queues

class WorkflowQueue {
  constructor(name: string, queueParameters?: QueueParameters);
}

interface QueueParameters {
  globalConcurrency?: number;
  workerConcurrency?: number;
  rateLimit?: QueueRateLimit;
  partitionConcurrency?: number;
  partitionWorkerConcurrency?: number;
  partitionRateLimit?: QueueRateLimit;
  minPollingIntervalMs?: number;
}

The new WorkflowQueue(...) constructor declares an in-memory queue whose configuration is fixed at construction and lives only in process memory. It must be called before DBOS.launch(), and the resulting queue cannot be reconfigured at runtime.

warning

This API is deprecated. Use DBOS.registerQueue instead.

const queue = new WorkflowQueue(
    "example_queue",
    {
        workerConcurrency: 5,
        rateLimit: { limitPerPeriod: 50, periodSec: 30 }
    },
);