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 — it never runs anything on this process; by default it also publishes a Start through the transport, so a worker runs the initial on enter (pass start_on_create=False to defer that and call start yourself later) — send further events (also published), 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)  # PENDING — never runs on this process; Start is already
drain()                       # published (start_on_create=True, the default) — a worker
                               # (here, our own drain) runs it

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.

Only stop_event ends that loop. A message whose handling raises — a store or transport outage, a machine version the worker can’t run, an engine bug — is logged with its traceback and goes back to the queue, to come back after suspend_recheck seconds (default 5); a claim or a timer sweep that raises is logged, and the loop carries on after the same pause. An action’s own error isn’t one of these: the driver routes it (on error, else the execution fails). The sync Worker.run and the async AsyncWorker.run behave the same.

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 HAREL_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()) run the same logic either on a shared background event loop (an anyio blocking portal, the default) or, with execution="inline", in the caller’s own thread — so the deterministic step()-in-a-loop style works for embedding and tests, and a web view can create an execution and queue its Start inside its own transaction. See execution models.

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.

A step can run twice without anything crashing¶

A worker commits a step with a CAS on the execution’s version, so its commit loses to any other write of the same execution that landed while it ran the step: a control-plane command (suspend, cancel, terminate, redrive) from the runner, a process() from a DurableRunner on the same store, or another worker after the lease expired under an action slower than visibility. The losing worker gets a StoreConflict, puts the event back (nack), and the event is processed again on the new version. Nothing committed half, but the step’s actions ran on the first attempt and run again on the next — at-least-once, with no failure involved. Each attempt hands an action the same stm.idempotency_key, so a side effect deduped on it happens once (see durability).

Deploying a new version of a machine¶

The workers share one definition of record: the store. When python -m harel.worker starts it loads every *.stm at the top of HAREL_DEFINITIONS_DIR and publishes them — each machine version, with the files it imports. A version published already with the same source is left as it is; one published with other content (a machine changed without a new version) stops the worker before it claims anything, so the deploy fails, not the executions.

From then on a worker runs any execution, whatever version it is on:

a version it has loaded runs from its own files;
a version it hasn’t — an older one whose file is gone, or a newer one a newer worker published — is read from the store and kept (a published version never changes);
a version it can’t use is left for another worker: the message goes back to the queue (DefinitionNotLoaded in the log) and comes back after suspend_recheck. That is a version nobody published, one calling an action module the worker doesn’t allow, or one calling an action the worker’s code can’t import — an older worker meeting a version a newer deploy published, during a rolling deploy.

A rolling deploy of order version 2 therefore needs nothing special: the new workers publish version 2 and start creating its executions’ children and timers; the old ones keep running version 1’s executions, and leave version 2’s that call code they don’t have to the new ones. Run harel publish DIR in the pipeline to stop a bad deploy before any worker rolls out.

A version read from the store may call only the action modules the worker’s own machines call, plus those in HAREL_ACTION_MODULES (comma-separated; a module allows the modules under it). A runner of your own takes the same as action_modules=, and actions= / guards= for what your code binds instead of a bind in the DSL. See changing a machine.

Scaling & throughput¶

Two independent dials:

Concurrency within a worker (HAREL_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.

Warning

The numbers in bench/RESULTS.md were measured on one 8-core laptop with the backends in Docker Desktop, workers and backends sharing the cores — so a multi-worker plateau there is often the laptop, not the backend. Read them as relative; a backend on its own hardware, with workers on other machines, 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 a toy FSM (start, then one more event, then done), same Postgres + laptop, measured the same way for both — the timed window covers sending and processing every event, with the clients and the workers in separate processes:

	best case	one worker process
harel-on-Postgres	~900 events/s	~860 events/s
DBOS — durable workflow + `send`/`recv`	~470–500 events/s (4 workers)	~205 events/s

(DBOS was still gaining at 4 worker processes, the most this host could run without them competing for cores.) 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.

Dedicating workers to machines¶

A worker can be given capacity for some machines only — so a busy machine gets workers of its own, and doesn’t take them from the others:

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

ORDERS = """
event Go {}
machine fulfilment { initial Packing state Packing {} final Shipped success from Packing to Shipped on Go }
"""
fulfilment = definition_from_dsl(ORDERS, "fulfilment")
runner = DistributedRunner(DictStore(), InMemoryTransport(), {fulfilment.id: fulfilment})

dedicated = runner.worker("fulfilment-1", machines={"fulfilment"})  # only fulfilment's work
anyone = runner.worker("general-1")                                 # everything, fulfilment included

python -m harel.worker takes the same as HAREL_MACHINES=fulfilment,publishing, and logs at start which machines it claims for.

A tree travels with its root. An execution’s routing_key is the machine its tree started from: its orthogonal regions, fan-out instances and invoked submachines carry the root’s, so a worker given fulfilment runs its parcels.parcel children too. Name only root machines.
Dedicated capacity, not exclusivity. A worker given no machines claims everything, the dedicated machines’ work included: it is the safety net when the dedicated workers are down or busy. To keep a machine’s work off the general workers, give every worker its machines.
Nothing claims a machine no worker covers when every worker is given machines: its executions wait. A worker warns at start about a machine it is given but has no file for.
Transports. The in-memory, SQLite, libSQL, rqlite, Postgres and Mongo transports route claims; Redis and SQS can’t, and a worker given machines on them fails at its first claim (NotImplementedError) rather than claim every machine’s work. See transports.

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

Independent work is marked as such in the runners’ logic, and run with coroutines (the async runners, and the sync ones in the default execution="background") it overlaps on the event loop — concurrency even in a single-process DurableRunner. In the caller’s thread (execution="inline") the same work runs in order, with the same result:

Spawn relay: 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): a worker publishes all the timers due in one sweep concurrently. The headless runner delivers them one at a time, since two timers of the same execution must not race on its version.
Control-plane propagation (cancel/suspend/resume): status changes propagate to all children at each level of the tree concurrently.