Scaling the engine

The engine scales in two directions. Replicas add CPU for the engine, FEEL and the API. Shards add database capacity: a shard is a PostgreSQL database that holds a share of the engine's runtime data. Engine describes how replicas and partitions work. This page covers when to add which, how to size them, and how to add a shard.

Replicas or shards

Every step is one transaction, and every job poll and completion is a query, so the database is usually the first limit. Look at where the time goes:

• The backend pods are busy, near their CPU limit, while the database has headroom. Add replicas, or give them more CPU.
• The step workers are busy while steps stay fast (see Diagnosing with the metrics). Raise ENGINE_STEP_WORKERS.
• The database is busy: its CPU, WAL writes or disk are near their limit, and steps slow down as you add step workers or replicas. Add a shard on another PostgreSQL server.

How shards work

Every instance lives on one partition, and the engine has 256 partitions by default. A shard holds a share of the partitions, with their instances, jobs, user tasks, timers, messages and history. The app database is always shard quantumbpm-1 and also holds what the whole cluster shares: the deployed processes and decisions, the partition map and the partition leases. Extra shards hold runtime data only.

• An instance stays on its partition for its whole life, so a step reads and writes one shard. A message, a signal or a call activity that reaches an instance on another shard goes through the outbox, as it does within one shard.
• Every replica connects to every shard. A replica owns an even share of the partitions whichever shards they live on, so replicas and shards are sized independently.
• A shard adds capacity only on its own PostgreSQL server. Shard databases on one server share its CPU, WAL and disk. They are still a useful start: a shard database moved whole to a new server, with the backend stopped, keeps its shard ID and partitions. Only its URL changes, and no rebalance is needed.
• Shards can be added, not removed. A shard that holds partitions must stay in the configuration, or the backend refuses to start.

Configuring shards

Each extra shard is an ID and a PostgreSQL URL:

engine:
  shards:
    - id: quantumbpm-2
      url: postgres://bpm:<password>@shard2.example:5432/bpm?sslmode=require
    - id: quantumbpm-3
      url: postgres://bpm:<password>@shard3.example:5432/bpm?sslmode=require

ENGINE_SHARDS takes the same list as a JSON array, so a single secret can carry every shard with its credentials:

ENGINE_SHARDS='[{"id":"quantumbpm-2","url":"postgres://bpm:<password>@shard2.example:5432/bpm?sslmode=require"}]'

The Helm chart reads it from a Secret, see engineShards in the chart's README.md.

• IDs use lowercase letters, digits, - and _, up to 63 characters. quantumbpm-1 is the app database and is not listed. An ID is permanent once its shard holds partitions.
• The database starts empty and is owned by the URL's user. The backend creates the engine's schema in it on startup, as it does in the app database.
• The pool to each extra shard holds up to ENGINE_SHARD_MAX_CONNS connections per replica. A URL that sets pool_max_conns keeps its own.
• A new shard holds no partitions until a rebalance gives it its share.

Every shard database records the cluster it belongs to, its shard ID and the version of the partition map. The backend refuses to start, and names the shard, when:

• a shard that holds partitions is missing from the configuration (partitions live on shards missing from the config),
• a shard URL points at another shard's database, for example two URLs swapped (check the shard URLs), or at another cluster's database,
• a shard holds partitions but its database has no engine data, which happens when the URL points at an empty database,
• the databases were restored to different points, or a rebalance did not finish (run it again).

Sizing

• Partitions: keep 256. The count is fixed when the engine first starts, and caps the cluster at 256 shards.
• Replicas: set them by backend CPU (FEEL, the interpreter, API traffic), at least 2 for availability. They need no relation to the shard count, and the HPA can scale them freely.
• Step workers: the steps running at once on a shard are about replicas x ENGINE_STEP_WORKERS / shards. More shards give more write capacity, so raise ENGINE_STEP_WORKERS to use it. Start from the concurrency one shard server handles well, roughly 2 to 4 per core, and work back: ENGINE_STEP_WORKERS = per-shard target x shards / replicas.
• Relay and correlation workers follow the shard count, 2 of each per shard on every replica. There is nothing to set.
• Pools: each replica opens one pool per extra shard, up to ENGINE_SHARD_MAX_CONNS connections (default ENGINE_STEP_WORKERS + 16). Keep DB_MAX_CONNS larger, as the app database also serves the API, the catalog and shard quantumbpm-1.
• max_connections: at least replicas x ENGINE_SHARD_MAX_CONNS on each shard server and replicas x DB_MAX_CONNS on the app database's server, plus headroom for admin sessions. Several shard databases on one server add up. With the HPA, count maxReplicas.
• No PgBouncer in transaction mode in front of any engine database. Migrations and the rebalance take session-level advisory locks, and the backend caches prepared statements per connection.
• PostgreSQL in Docker gets a 64 MB /dev/shm by default, which parallel queries outgrow. Start the container with --shm-size=1g or more.

An example with 4 replicas and 6 shard servers:

database:
  maxConns: 100            # app database: API, catalog, shard quantumbpm-1
engine:
  stepWorkers: 32          # 4 x 32 = 128 steps at once, about 21 per shard
  shardMaxConns: 48        # the default for 32 step workers, 4 x 48 = 192 connections on each shard server
  shards:                  # quantumbpm-1 is the app database and is not listed
    - { id: quantumbpm-2, url: "postgres://bpm:<password>@shard2.example:5432/bpm" }
    - { id: quantumbpm-3, url: "postgres://bpm:<password>@shard3.example:5432/bpm" }
    - { id: quantumbpm-4, url: "postgres://bpm:<password>@shard4.example:5432/bpm" }
    - { id: quantumbpm-5, url: "postgres://bpm:<password>@shard5.example:5432/bpm" }
    - { id: quantumbpm-6, url: "postgres://bpm:<password>@shard6.example:5432/bpm" }

Operations lists the metrics per shard: bpmn_engine_shard_up, step duration, relay lag, backlog and table sizes, and the pgxpool_* pool series.

Adding a shard

A rebalance gives every shard an even share of the partitions. It runs with every replica stopped and copies the rows of the partitions that move, so the downtime grows with the data on them.

1. Create the new database, empty and owned by the backend's user. If you back up with restore points, set up its WAL archiving now (see Backups with shards).

2. Add it to engine.shards (ENGINE_SHARDS).

3. Stop every replica.

4. Run the rebalance with --dry-run first. It lists the partitions that would move with their row counts, and changes nothing:

   docker run --rm --env-file backend.env <backend image> rebalance --dry-run
   docker run --rm --env-file backend.env <backend image> rebalance

   The command runs in the backend image with the backend's configuration, including ENGINE_SHARDS. It needs no license key. It logs Rebalance finished with the partitions and rows it moved.

5. Start the replicas.

6. Take a base backup of the new shard, or a fresh offline backup of every database. Until then no backup can restore the rows that moved to it.

The chart's README.md has the same steps as a Kubernetes Job.

• Duration: plan for about 2 GB per minute of live engine data on the moving partitions, plus about 50 ms per partition. Going from N to 2N shards moves about half the data. Dead space in the tables does not move.
• The rebalance refuses while a replica runs. A replica that was killed drops out within 10 seconds. A replica that starts during a rebalance waits for it to finish.
• A rebalance that stops partway, from a crash, a signal or a failed copy, leaves the backend refusing to start with a rebalance did not finish, run it again. Run it again and it resumes. Each partition copies in one transaction, so no shard holds a partial copy.
• The moves are as few as possible: shards that already hold partitions keep as many of them as their share allows.