Coordinate concurrent workflows

When several executions can touch the same record, use an atomic data store step. It performs the check and the write together, so only one execution wins.

Atomic Concurrency Primitives

Two workflows (or two runs of the same workflow) that operate on the same row can race: each reads the same value, each decides independently, each writes — and the writes overlap. The atomic primitives run the read, the check, and the write as one indivisible operation, so this can’t happen.

Three primitives cover the patterns below:

Step	What it does	Used in
data-store.atomic-update	Conditionally update one row by key.	Token bucket, Queue dispatcher, Stale-claim recovery
data-store.atomic-claim	Claim N rows matching a filter.	Queue dispatcher, Stale-claim recovery
data-store.atomic-upsert	Insert if absent, or mutate on conflict.	Process event once, Lease acquisition, Token bucket, Rate limit per key

Note

An empty result is not an error. If the guard didn’t match, the queue was empty, or you lost a race, the return value tells you so — updated: false, claimed: [], inserted: false. Branch on the return; don’t try to catch an exception.

Shared building blocks

All three primitives share two pieces of configuration:

• where — the same {field, operator, value} filter shape that Query uses, applied to the stored values. Same operators, same path syntax for nested fields.
• set — a list of {field, operation, value} mutations applied to the value. operation is either set (overwrite) or increment (add a signed number; a missing field starts from 0).

Both accept templates: {{ $util.now }}, {{ $execution.id }}, references to prior step outputs, and so on.

Atomic Update

Conditionally mutate a single row identified by key. The where clause is the compare-and-swap guard: the update applies only if all (or any) conditions match the row’s current value. If the guard fails, nothing changes and updated is false.

Field	Description
Table	The table containing the row
Key	Exact key of the row (supports templates)
Guard Conditions (optional)	Preconditions on the current value; the update applies only if they match
Match	all (AND) or any (OR) for combining guards
Set	One or more mutations to apply when the guard matches
Return Fields (optional)	Limit the returned value to specific top-level fields

Output:

{{ decrement-tokens.updated }}      // true if the guard matched and the row was updated
{{ decrement-tokens.value }}        // the new value if updated, else null
{{ decrement-tokens.key }}          // the key that was targeted

atomic-update doesn’t create the row if it’s absent. Seed it first with atomic-upsert.

A cross-execution counter is exactly this shape: seed the row with atomic-upsert, then atomic-update with a count increment 1 mutation on each run. The increment happens inside the row, so concurrent runs never lose a count the way a separate Get-then-Set would.

Atomic Claim

Claim up to limit rows that match where, marking them via set in the same operation. The data store guarantees that two concurrent claims never grab the same row; losers skip past locked rows and continue. This is the queue-drain primitive.

Field	Description
Table	The queue table
Select Rows (optional)	Conditions identifying claimable rows (e.g. status equals pending)
Match	all or any for combining conditions
Claim Mutation	How to mark claimed rows. Defaults to status → in_flight, claimedAt → {{ $util.now }} (so stale claims can be recovered), and claimedBy → {{ $execution.id }} (so claimed rows correlate to the run that holds them in traces).
Order By	Column (createdAt or recordTimestamp) and direction. Default: FIFO (createdAt asc).
Limit	Max rows per call (default 10, max 1,000)
Return Fields (optional)	Limit returned value to specific top-level fields

Output:

{{ drain-queue.claimed }}            // array of { key, value } that this call won
{{ drain-queue.count }}              // number of rows claimed (0 = nothing available)
{{ drain-queue.claimed[0].value }}   // first claimed row's value

An empty claimed array means nothing was available, not an error.

Atomic Upsert

Insert a row when it doesn’t exist, or apply set to the existing row when it does. The table and key identify the record. The inserted flag in the output tells you which branch fired — useful for any “first writer initializes, others update” pattern.

Field	Description
Table	The table
Key	The conflict key
Insert Value	Full value to insert when the row doesn’t yet exist
Update Mutation	Mutations to apply when the row already exists (the conflict branch)
Return Fields (optional)	Limit returned value to specific top-level fields

Output:

{{ seed-bucket.inserted }}   // true if a new row was inserted, false if an existing row was updated
{{ seed-bucket.value }}      // the resulting value
{{ seed-bucket.key }}        // the key

---

Concurrency Patterns

Idempotency / Process Once

Handle each event exactly once, even when the source delivers duplicates. The event’s unique ID becomes the dedupe key; atomic-upsert with set: [] claims the event in one operation — the first caller inserts, every subsequent delivery sees the existing row and skips.

Step	Configuration
Atomic Upsert	Table: processed_events, Key: {{ initial.eventId }}, Insert Value: {"claimedAt": "{{ $util.now }}", "payload": "{{ initial.payload }}"}, Update Mutation: (empty list — pure seed-if-missing semantics)
If	Condition: {{ claim.inserted }} equals true — first delivery, do the work; else skip as duplicate

The insert is atomic, so two concurrent deliveries can’t both pass the dedupe check; one succeeds, and the others see the row already exists.

Recipe: Process Event Once — copy-pasteable workflow.

Token Bucket (rate limiting)

Limit how many parallel callers can proceed in a window. Each caller decrements a shared counter; when the counter hits zero, further callers lose the race and back off.

Seed the bucket once (idempotent: repeated runs leave it alone):

Step	Configuration
Atomic Upsert	Table: rate_limits, Key: external-api, Insert Value: {"tokens": 100}, Update Mutation: (empty — nothing to do on conflict)

Decrement on each call:

Step	Configuration
Atomic Update	Table: rate_limits, Key: external-api, Guard: tokens gt 0, Set: tokens increment -1
If	Condition: {{ decrement.updated }} equals false (back off, retry, or give up)

When tokens reaches 0, the guard tokens gt 0 no longer matches; the row is untouched, updated is false, and the caller knows to back off.

Rate limit per key

Rather than a fixed pool of tokens, the per-key pattern advances a shared “next available at” timestamp by a fixed interval on each call. Each caller atomically reserves its slot and waits until that slot before proceeding. Different from the token bucket in that calls space out evenly rather than burst-then-refill.

Recipe: Rate-Limited HTTP Calls — the full workflow with reset-if-stale handling.

Lease Acquisition

Only one workflow at a time holds the lease (for example, a single dispatcher). The first caller creates the lease; subsequent callers see it already held and back off until it’s released.

Step	Configuration
Atomic Upsert	Table: leases, Key: dispatcher, Insert Value: {"holder": "{{ $execution.id }}", "expiresAt": "..."}, Update Mutation: (empty — leave the existing lease alone)
If	Condition: {{ acquire.inserted }} equals true (you hold the lease; otherwise back off)

Lease renewal is an atomic-update against the same key with a guard on holder (only the current holder may extend) and a fresh expiresAt.

Recipe: Leader Election Lease — full workflow with the if/else branch and the release step.

Queue Drain (Producer / Dispatcher)

Producers enqueue work into a shared table; one or more dispatchers drain at a controlled rate. The atomic claim guarantees no row is processed twice across concurrent dispatcher runs — concurrent dispatchers either get disjoint batches or one gets nothing.

Producer workflows add entries to a pending_ops table using atomic-upsert (or plain Set) with {"status": "pending", "payload": ...}.

Dispatcher workflow runs on a schedule:

Step	Configuration
Atomic Claim	Table: pending_ops, Select Rows: status equals pending, Order By: createdAt asc (FIFO), Limit: 10
For Each	Items: {{ drain.claimed }} (process each claimed op via HTTP, etc.)
Atomic Update (inside For Each)	Table: pending_ops, Key: {{ $item.key }}, Guard: claimedBy equals {{ $execution.id }}, Set: status → done (or status → failed with error details)

The claim marked the rows in_flight in the same step that returned them, so no other dispatcher run can pick them up. If the workflow crashes mid-process, the rows stay in_flight — which the stale-claim recovery pattern below handles.

Recipe: Async Queue Dispatcher — full workflow including stale-claim recovery as a separate scheduled run.

Stale-Claim Recovery

If a dispatcher crashes after claiming but before completing, the claimed rows are stuck at status = in_flight. A second scheduled workflow re-claims any in-flight row whose claimedAt is older than a threshold:

Step	Configuration
Atomic Claim	Table: pending_ops, Select Rows: status equals in_flight AND claimedAt lt {{ stale_threshold }}, Set: status → in_flight, claimedAt → {{ $util.now }} (re-stamps the claim)
For Each	Process exactly as in the main drain

The default atomic-claim set rows already write claimedAt = {{ $util.now }}, so this works out of the box.

Note

The filter combines conditions with a single all or any mode; it can’t express “pending OR (in_flight AND stale)” in one call. Run the normal drain and the stale-claim recovery as two separate scheduled passes.

What’s not supported

These primitives are deliberately scoped:

• Multi-step transactions that span workflow steps. Each primitive is one atomic operation, not a transaction block.
• Bulk multi-row updates by filter. That’s what atomic-claim is for. atomic-update is single-row by design.
• Claim-and-delete. atomic-claim marks rows in place (at-least-once delivery). Disposable queues would need a future mode: 'delete'.
• Priority queues (ordering by a stored value field). orderBy supports createdAt and recordTimestamp only.
• A reaper daemon for stuck rows. Use the stale-claim recovery pattern above instead.

---