Skip to main content

Sharding

info

Sharding is a DBOS Transact Enterprise feature.

A DBOS application using a single Postgres system database can run about 1,000 workflows or steps per second per Postgres vCPU, or 20K workflows or steps per second on a 16-vCPU server. To scale beyond that, you can shard your system database across multiple Postgres servers using a Postgres-compatible sharding layer such as Citus or PgDog. With sharded Postgres, DBOS scales linearly to >1M workflows or steps per second.

When you shard your system database, every workflow lives under a shard key that you assign it. All of a workflow's data (its status, inputs, outputs, steps, events, messages, and streams) is stored on the shard that owns its key, so every operation on a workflow touches a single shard.

Setting Up a Sharded System Database​

To use a sharded system database, set sharding in your DBOS configuration and point system_database_url at your sharding router (for example, a Citus coordinator or PgDog router):

config: DBOSConfig = {
    "name": "dbos-app",
    "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    "conductor_key": os.environ.get("DBOS_CONDUCTOR_KEY"),
    "sharding": True,
}
DBOS(config=config)

When DBOS launches, it creates the sharded system database schema through your sharding layer. In production, we recommend you create the schema ahead of time with dbos migrate --sharding.

Every DBOS table has a shard_key column. You may need to configure your sharding engine to shard on that column, for example in PgDog's pgdog.toml:

pgdog.toml
[[sharded_tables]]
database = "dbos"
column = "shard_key"
data_type = "varchar"

To connect a DBOSClient to a sharded system database, construct it with sharding=True.

Assigning Shard Keys​

On a sharded system database, you must assign a shard key to every workflow you start or enqueue. Use SetShardKey to set the shard key of every workflow started, enqueued, or called in a block:

from dbos import DBOS, SetShardKey

queue = DBOS.register_queue("orders")

with SetShardKey("orders"):
    handle = queue.enqueue(process_order, order)

From a DBOSClient, set shard_key in the enqueue options:

options: EnqueueOptions = {
    "workflow_name": "process_order",
    "queue_name": "orders",
    "shard_key": "orders",
}
handle = client.enqueue(options, order)

Workflow IDs are unique only within a shard key: two workflows with the same ID under different shard keys are different workflows. Thus, to identify a workflow on a sharded system database, you need both its ID and its shard key. Within a workflow, DBOS.shard_key returns the shard key of the current workflow.

Choosing Shard Keys​

Queue flow control limits (its global concurrency limit, rate limit, and partition limits) are enforced separately within each shard key: for example, a queue with a global concurrency of 10 can run 10 workflows under each shard key. Worker concurrency limits still apply across all shard keys.

Therefore, we recommend choosing shard keys based on how you enqueue workflows:

  • For workflows enqueued on a partitioned queue, use their partition key as their shard key. Each partition then lives on one shard, and its partition limits work as they would on an unsharded database.
  • For workflows enqueued on a queue that isn't partitioned, if you need queue-wide flow control, use the queue's name as their shard key, so the queue's limits apply to all its workflows.

Child Workflows​

A workflow doesn't inherit its caller's shard key, so you must set the shard key of each child workflow. A child workflow that you start or call directly must have the same shard key as its parent:

@DBOS.workflow()
def parent_workflow():
    with SetShardKey(DBOS.shard_key):
        child_workflow()

A child workflow that you enqueue can have any shard key:

@DBOS.workflow()
def parent_workflow():
    with SetShardKey("reports"):
        handle = report_queue.enqueue(generate_report)
    return handle.get_result()

Interacting With Workflows​

Every method that operates on a workflow by its ID, such as DBOS.send, DBOS.get_event, DBOS.retrieve_workflow, and the workflow management methods, takes a shard_key parameter. On a sharded system database, you must pass the shard key of the workflow:

DBOS.send(workflow_id, message, "topic", shard_key="orders")
DBOS.cancel_workflow(workflow_id, shard_key="orders")

Methods that operate on several workflows at once, such as DBOS.cancel_workflows and DBOS.send_bulk, only accept workflows under a single shard key per call.

DBOS.list_workflows and DBOS.list_queued_workflows list workflows across all shards. Each returned WorkflowStatus includes the workflow's shard_key, and you can filter by shard_key or parent_shard_key.

Scheduled workflows run under the shard key of their schedule, which you set with the shard_key parameter of DBOS.create_schedule.

Limitations​

The following features are not supported on a sharded system database:

Of the DBOS CLI commands, only dbos migrate supports a sharded system database.