Skip to content
QUICKFLO DOCS

Async Queue Dispatcher

A scheduled dispatcher workflow that drains a queue of pending operations safely under concurrent runs. The atomic claim grabs a disjoint batch each time — two dispatchers that fire at the same moment either claim non-overlapping rows or one claims everything available and the other claims nothing. No row is ever processed twice.

This is the workflow side of a producer/dispatcher split: any number of producer workflows (webhooks, schedules, manual triggers) can write rows into a pending_ops table; this dispatcher pulls them out at whatever rate you schedule it to fire.

Workflow JSON

Section titled “Workflow JSON”

Copy this and paste it into a new workflow with Cmd+V:

async-queue-dispatcher.json
{
  "name": "async-queue-dispatcher",
  "description": "Drain a queue of pending operations at a controlled rate. Atomically claims up to N pending rows so two concurrent dispatcher runs never grab the same row, processes each one, then marks it done. Pair with the stale-claim recovery workflow (see What to customize first) to handle crashed runs.",
  "steps": [
    {
      "stepId": "drain",
      "stepType": "data-store.atomic-claim",
      "input": {
        "tableName": "pending_ops",
        "where": [{ "field": "status", "operator": "equals", "value": "pending" }],
        "orderBy": { "field": "createdAt", "direction": "asc" },
        "limit": 10
      }
    },
    {
      "stepId": "process-each",
      "stepType": "core.for-each",
      "input": {
        "items": "{{ drain.claimed }}",
        "concurrency": 5,
        "continueOnError": true,
        "steps": [
          {
            "stepId": "do-work",
            "stepType": "core.log",
            "input": {
              "message": "Processing op {{ $item.key }} with payload {{ $item.value.payload }}. Replace this step with the real work (HTTP call, API request, etc.). If it throws, the row stays in_flight and stale-claim recovery will reclaim it later."
            }
          },
          {
            "stepId": "mark-done",
            "stepType": "data-store.atomic-update",
            "input": {
              "tableName": "pending_ops",
              "key": "{{ $item.key }}",
              "where": [{ "field": "claimedBy", "operator": "equals", "value": "{{ $execution.id }}" }],
              "set": [{ "field": "status", "operation": "set", "value": "done" }]
            }
          }
        ]
      }
    }
  ]
}

Setup

Section titled “Setup”

Connections needed: none — the example uses core.log as the work step. Replace it with whatever your dispatcher actually does (HTTP call, downstream workflow, API request).

Trigger: a Schedule trigger firing at whatever cadence you want to drain at — every minute, every 5 minutes, etc. Concurrent fires are safe.

Tables:

  • pending_ops — created by producer workflows when they enqueue work. Each row’s value carries the payload and a status: "pending" field.

How it works

Section titled “How it works”

Two steps:

  1. drain — data-store.atomic-claim against pending_ops where status = pending. Claims up to 10 rows, ordered FIFO by createdAt. The claim atomically flips status to in_flight and stamps claimedAt and claimedBy (the executing run’s ID) on each row, so a second dispatcher firing at the same moment skips these rows.
  2. process-each — core.for-each over drain.claimed with concurrency: 5 (process five at a time). For each item:
    • do-work — the placeholder. Replace with the real work. If it throws, this row’s status stays in_flight and stale-claim recovery (below) will reclaim it.
    • mark-done — data-store.atomic-update guarded by claimedBy equals {{ $execution.id }}. The guard matters: if stale-claim recovery already reclaimed this row to a different run, this update is a no-op. Otherwise it flips status to done.

Producing work

Section titled “Producing work”

Other workflows enqueue items by inserting into the same table. A typical producer step:

{
  "stepId": "enqueue",
  "stepType": "data-store.set",
  "input": {
    "tableName": "pending_ops",
    "key": "{{ $util.uuid }}",
    "value": {
      "status": "pending",
      "payload": { "userId": "user_42", "action": "send-email" }
    }
  }
}

Any number of producer workflows can write at any rate. The dispatcher catches up on its own schedule.

Stale-claim recovery

Section titled “Stale-claim recovery”

If a dispatcher crashes between drain and mark-done, the claimed rows are stuck at status = in_flight. Run a second scheduled workflow on a slower cadence (say, every 5 minutes) that reclaims them:

{
  "steps": [
    {
      "stepId": "reclaim-stale",
      "stepType": "data-store.atomic-claim",
      "input": {
        "tableName": "pending_ops",
        "where": [
          { "field": "status", "operator": "equals", "value": "in_flight" },
          { "field": "claimedAt", "operator": "lt", "value": "{{ initial.staleBeforeTimestamp }}" }
        ],
        "filterMode": "all",
        "orderBy": { "field": "createdAt", "direction": "asc" },
        "limit": 50
      }
    },
    {
      "stepId": "process-recovered",
      "stepType": "core.for-each",
      "input": {
        "items": "{{ reclaim-stale.claimed }}",
        "concurrency": 5,
        "continueOnError": true,
        "steps": [
          {
            "stepId": "do-recovered-work",
            "stepType": "core.log",
            "input": { "message": "Recovered abandoned op {{ $item.key }}." }
          },
          {
            "stepId": "mark-recovered-done",
            "stepType": "data-store.atomic-update",
            "input": {
              "tableName": "pending_ops",
              "key": "{{ $item.key }}",
              "where": [{ "field": "claimedBy", "operator": "equals", "value": "{{ $execution.id }}" }],
              "set": [{ "field": "status", "operation": "set", "value": "done" }]
            }
          }
        ]
      }
    }
  ]
}

The recovery claim re-stamps claimedAt and claimedBy (the default atomic-claim set), so it won’t reclaim the same row on a subsequent pass unless this run also fails. The trigger should pass a staleBeforeTimestamp (e.g. “now minus 5 minutes”); compute it in the trigger or in a preceding step.

What to customize first

Section titled “What to customize first”
  • Replace the work step. The core.log step inside the for-each is a placeholder. Put the actual processing there.
  • Tune the limit and concurrency. limit: 10 claims up to 10 rows per drain; concurrency: 5 processes 5 in parallel. Adjust based on how long the work step takes and what downstream capacity is.
  • Add a leader-election lease on the dispatcher itself if you don’t want even one concurrent dispatch. The atomic claim makes concurrent dispatchers safe (they get disjoint batches), but if your work step needs strict serialization (e.g. ordered consumer), wrap the whole dispatcher in a Leader Election Lease so only one dispatcher runs at a time.
Section titled “Related recipes”