Distribution

Durability lets one process resume an execution. Distribution lets many workers share the load while guaranteeing each execution is only ever advanced by one of them at a time.

The transport seam

Alongside the store sits an independent transport — a queue with single-active-consumer per group, where the group is the execution id. That property is what makes concurrency safe: events for one execution are processed in order by one worker, even with a pool of workers running. Backends:

Transport	Exclusivity mechanism
InMemoryTransport	in-process (tests, single process)
SqliteTransport	BEGIN IMMEDIATE (global write-lock) + lease
RedisTransport	SET lock NX PX group-lock-as-lease + FIFO list; one ready:{prio} ZSET per priority level, scored by available-at, so claim is O(log N + K) per tier and a high-priority group isn’t starved behind a low-priority backlog
PostgresTransport	per-group row claimed with SELECT … FOR UPDATE SKIP LOCKED — concurrent workers lease different groups in parallel (no global lock)
RqliteTransport	queue table on Raft-replicated SQLite
MongoTransport	per-group ready-index/lock document (available_at + token; find_one_and_update lease)
LibsqlTransport	libSQL (Turso’s SQLite fork): BEGIN IMMEDIATE + lease, like SqliteTransport; local file, sqld, or a Turso replica. Experimental (local-file tested; Turso/sqld path unvalidated against a real account)
SqsTransport	SQS FIFO MessageGroupId (runs on LocalStack)

Backends without a cheap global serialization primitive (Redis, Mongo) build per-group exclusivity by hand with a per-group lock record indexed by when it next becomes claimable, so claim reads only the few due groups rather than scanning the whole queue; SQLite and Postgres lean on the database instead (SQLite’s write-lock; Postgres’s row lock via SKIP LOCKED). For how each transport enforces the per-group invariant and lays its queue out — the lease, claim, FIFO and parking mechanics — see the transports reference.

Store and transport are independent — mix freely, or unify on one backend (all-Postgres, all-rqlite, all-Mongo, all-libSQL: no Redis needed).

Running with workers

DistributedRunner(store, transport, definitions) is the façade: create an execution, send it events (published to the transport), and run one or more workers that claim → load → dedupe → route → ack. A worker’s step() processes exactly one available message and returns whether it did — which lets us drive it deterministically here, without thread timing:

from harel import definition_from_dsl, DictStore, Event
from harel.engine.distributed import DistributedRunner
from harel.engine.transport import InMemoryTransport

SOURCE = """
event PlaceOrder {}
event Deliver {}

machine order {
  initial Cart
  state Cart {}
  state AwaitingPayment {}
  final Delivered success {}
  from Cart to AwaitingPayment on PlaceOrder
  from AwaitingPayment to Delivered on Deliver
}
"""

defn = definition_from_dsl(SOURCE, "order")
runner = DistributedRunner(DictStore(), InMemoryTransport(), {defn.id: defn})
worker = runner.worker()


def drain():
    while worker.step():
        pass


exe = runner.create(defn.id)
drain()

runner.send(exe.id, Event(kind="PlaceOrder"))
drain()
print("after PlaceOrder ->", runner.store.load(exe.id).active_path)

runner.send(exe.id, Event(kind="Deliver"))
drain()
final = runner.store.load(exe.id)
print("after Deliver    ->", final.active_path, "/", final.status.name, "/", final.outcome)

after PlaceOrder -> AwaitingPayment
after Deliver    -> Delivered / DONE / success

In production you don’t call step() in a loop — a worker runs run(stop_event), looping claim→process and, on its idle path, sweeping due timers (fire_due_timers) so timeouts fire without a separate scheduler.

Async-native workers

The worker is async-native. The bundled python -m harel.worker runs one asyncio event loop (via anyio.run) that drives up to STM_CONCURRENCY events in flight at once (AsyncWorker.run, default 256): while one execution’s action awaits IO, the loop processes others — so one worker process overlaps many events’ IO on a single event loop, with no thread per event. The synchronous façades shown above (DistributedRunner, worker.step()) are thin wrappers over that async core through an anyio blocking portal, so the deterministic step()-in-a-loop style still works for embedding and tests.

Scale out by running more worker processes (or machines) against the same store and transport — the single-active-consumer-per-group property keeps each execution on one worker at a time regardless of how many are running.

Scaling & throughput

Two independent dials:

• Concurrency within a worker (STM_CONCURRENCY): how many events one event loop keeps in flight. Raising it lifts throughput until either the backend’s per-event round-trips (load → commit → claim → ack) or the event loop’s own per-event CPU becomes the limit — which of the two binds depends on the backend’s per-op latency on your host (a slow-I/O backend is round-trip-bound; a fast one is loop-CPU-bound). Too high a value degrades throughput: more in-flight coroutines cost more to schedule than they save.

• More workers / shards: adding worker processes lifts the aggregate sublinearly until something shared saturates. That ceiling is either the host (CPU-bound worker processes oversubscribing the cores) or the single backend instance — which one binds depends on your hardware. On a busy single instance you scale horizontally by sharding: because executions are independent (single-consumer per group, no cross-execution coordination), you partition them across independent (store, transport) instances — each shard shares nothing with the others. This is the same model Temporal (hash the id to a shard, add shards) and DBOS (partition, “your ceiling is your Postgres”) use.

Each backend’s claim/commit is folded into the fewest round-trips it can be — an atomic Lua script on Redis, one sorted find_one_and_update on Mongo, a plpgsql function (FOR UPDATE SKIP LOCKED inside) on Postgres — because the per-worker limit is usually round-trips, not the server (a single Postgres/Redis sat at single-digit % CPU even at the top rates we measured).

Warning

The numbers in bench/RESULTS.md are A/B comparisons on one laptop with the backends in Docker Desktop — its VM/network proxy and the shared cores cap the absolute throughput, so the multi-worker plateau there is the laptop, not the backend. Read them as relative gains; a backend on native hardware / managed cloud, with workers on a multi-core host, scales far higher on a single instance before sharding is needed. Run bench/bench_async.py, bench_workers.py, bench_shards.py against your own backend for real numbers.

harel and durable-execution engines (illustrative)

harel is a statechart engine; DBOS, Temporal and friends are durable-execution / workflow engines. They overlap — both keep long-lived state alive across crashes — but model the problem differently: declarative named states versus imperative durable code.

Warning

The figures below are illustrative only — NOT a fair benchmark of DBOS, and should not be cited as one. They run the same toy FSM on the same laptop + Docker Postgres (so both carry the handicap above), but the two tools do different work. DBOS ran in its default single-instance config. The point is the shape of the difference, not the ratio. Full caveats and the script (bench/bench_dbos.py) are in bench/RESULTS.md.

On the toy FSM (Idle → Working → Done), single process, same Postgres + laptop, measured the same way for both (timed window = enqueue and process every event):

	events/s
harel-on-Postgres	~800
DBOS — durable workflow + send/recv	~390
DBOS — durable workflow per event	~230

(harel scales further with more worker processes — ~2050 at 8 workers — but the single-process row is the like-for-like one here.) harel comes out ahead, but that’s the paradigm, not a verdict on DBOS: DBOS does full durable-workflow bookkeeping per event — workflow-status rows, automatic recovery of arbitrary imperative code, queues, SERIALIZABLE transactions — which is exactly what you want for “run these steps, with retries, and survive a crash mid-way”, and is overkill for the trivial state transition this toy does. harel’s per-event path is a lean claim → load → commit → ack. So pick by the shape of the problem: a statechart engine when the domain is a machine of named states with hierarchy and guards; a durable-execution engine when it’s imperative code that must survive crashes.

Fair queuing and priority

Round-robin fairness

By default workers use round-robin across active executions. After a group is processed (its message acked), its “last claimed” timestamp is set to now, so it moves to the back of the queue. Groups never yet claimed have timestamp 0 and are always preferred. This prevents a handful of high-traffic executions from monopolizing all worker capacity — the hot-partition problem that arises with pure FIFO queues.

Execution priority

Assign a priority level at creation time (0 = normal, up to 4):

urgent = runner.create(defn.id, priority=2)

The priority is stored in the transport on the first publish for that execution. It is fixed at creation: subsequent send() calls carry the same execution priority when publishing to a group that does not yet have messages in the transport, and INSERT OR IGNORE / $setOnInsert / HSETNX semantics ensure the first-published value wins.

Worker priority routing — high_ratio + priority_threshold

Workers decide which priority tier to serve on each claim step:

worker = runner.worker(
    high_ratio=0.8,        # 80 % of claim attempts target priority >= threshold first
    priority_threshold=2,  # "high" means priority 2, 3, or 4
)

On each step():

• With probability high_ratio, the worker tries claim(min_priority=priority_threshold). If nothing is available at that priority, it falls back to claim(min_priority=0) so the worker never idles when only normal-priority work is present.

• With probability 1 - high_ratio, it goes directly to claim(min_priority=0).

The default is high_ratio=0.0 — no priority routing, every claim considers all groups equally. A worker configured with high_ratio=1.0 always attempts high-priority work first and falls back to any priority, effectively draining high-priority executions before touching normal ones whenever both are available.

Orthogonal & fan-out, distributed

The same machinery carries orthogonal regions and fan-out: each region or addressed child is its own execution with its own group, so they genuinely run on different workers, and their Finished events travel back through the transport to the parent’s join. The model you wrote in the tutorial distributes without change. Region and fan-out children inherit the parent’s priority, so a high-priority workflow’s parallel work is claimed at that priority too (not demoted to 0).

The async engine uses asyncio.gather throughout to parallelise independent work on the event loop. In the headless DurableRunner this gives true parallelism even in a single-process deployment:

• Spawn relay (_flush): all pending children start concurrently — their on enter actions (LLM calls, HTTP requests, …) overlap rather than running one after another.

• Broadcast (inject): a domain event delivered to multiple live regions runs each region’s handler concurrently.

• Timer sweep (fire_due_timers): all timers due in the same sweep are fired concurrently; per-timer order (deliver before delete) is preserved via a local coroutine.

• Control-plane propagation (cancel/suspend/resume): status changes propagate to all children at each level of the tree concurrently.