---
title: "Queues and Workflows"
description: "Start an EZGH Workflows run for each message of a queue, resume runs waiting for a message, send messages to a queue from a workflow step, and get a message when a run ends."
---

> Documentation Index
> Fetch the complete documentation index at: https://docs.ezghcloud.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Queues and Workflows

EZGH Queues and [EZGH Workflows](/workflows) work together in four ways, all within one project:

- **[Start a run per message](#start-a-workflow-run-from-a-queue):** a queue's subscription starts
  a run of a workflow for each message, with the message as the run's input.
- **[Wait for a message](#wait-for-a-message-in-a-workflow):** a run waits in
  `waitForMessage` until a message with its key arrives on a queue, and a queue's subscription
  resumes it.
- **[Send from a workflow](#send-messages-from-a-workflow):** a step sends messages to a queue,
  exactly once, with `sendToQueue`.
- **[Run-result notifications](#run-result-notifications):** when an app's runs end, Workflows
  sends a message describing each one to a queue.

## Start a workflow run from a queue

A subscription can deliver to a workflow instead of an [HTTPS endpoint](/queues/push): each
message starts a run of the workflow, and is acknowledged once the start is accepted.

### Console

Open the queue, then **Subscription**, and **Create subscription**. Choose **Start a workflow
run**, pick the Workflows app and the workflow, and set the concurrency.
### CLI

```sh
ezgh queues subscriptions create orders --workflow-app billing-jobs --workflow chargeCustomer
```
### API

```sh
curl -sS -X POST "https://queues-api.ezghcloud.com/v1/organizations/$ORG_ID/projects/$PROJECT_ID/subscriptions" \
  -H "Authorization: Bearer $EZGH_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d "{ \"queueId\": \"$QUEUE_ID\",
        \"target\": { \"kind\": \"workflow\", \"mode\": \"start\", \"appId\": \"$APP_ID\", \"workflow\": \"chargeCustomer\" },
        \"maxConcurrency\": 10 }"
```

[CreateSubscription](/queues-api/CreateSubscription/) answers the subscription with
`batchSize: 1` and `signingSecret: null`: deliveries to a workflow aren't signed.

- **Who may create it:** `queues.subscriptions.create` on the queue, and `workflows.runs.start` on
  the workflow (`<the app's resource name>/<workflow>`). Both are checked when you create or change
  the subscription; deliveries aren't checked again.
- **The app** is a push or pull app in the queue's project, and the workflow one of its
  workflows. Until a push app has synced (or a pull app's workers have announced its workflows),
  starts are refused and retried.
- **One message, one run.** Deliveries are one message each (`batchSize` is 1), up to
  `maxConcurrency` at once. On a FIFO queue, a message group's next message waits until the
  previous one's run has started, so a group's runs start in order.

The run's input is the message:

```json
{
  "messageId": "01999f50-1a2b-7c3d-8e4f-5a6b7c8d9e0f",
  "body": "{\"orderId\":1042}",
  "attributes": { "source": "web" },
  "sentAt": "2026-10-13T18:00:00Z",
  "queueId": "01999f3b-9e8d-7c6b-a5f4-e3d2c1b0a9f8",
  "groupId": "acct-9",
  "seq": "18446"
}
```

`groupId` and `seq` are there on FIFO queues only. The body is the message's text: parse it in the
workflow when it's JSON.

**Delivery.** A start is accepted when Workflows has created the run, or found the run this
message already started: the run's ID comes from the project, queue and message, so a message
delivered again reaches the same run instead of starting another. Starts are admitted like
[runs started through the API](/workflows/limits), except for the caller's permission check:

| Workflows answers | Queues |
| --- | --- |
| The run was started, or already existed | Acknowledges the message |
| `429` or `503` | Tries again after `Retry-After` |
| The app or workflow isn't found, isn't synced or announced, the organization isn't active, or the input is refused | Tries again with backoff; after the queue's `maxReceiveCount` deliveries the message moves to its [dead-letter queue](/queues/dead-letter-queues) |

A subscription's [status](/queues/push#status) shows the last failure, as for an HTTPS endpoint.
**Pause**, **Resume**, **Delete** and changing `maxConcurrency` work the same way; the target's
kind can't change.

## Wait for a message in a workflow

A run can pause until a message about it arrives: a payment, an approval, a reply from another
system. `waitForMessage` waits on a queue and a key, and a subscription in **resume** mode delivers
the queue's messages to the waiting runs by one of the message's attributes.

### In the workflow

### TypeScript

```ts
export const order = workflow("order", async ({ input, step }) => {
  await step.run("place", () => shop.place(input))
  const paid = await step.waitForMessage("paid", {
    queueId: process.env.PAYMENTS_QUEUE_ID!,
    key: `order:${input.orderId}`,
    timeoutSeconds: 86_400,
  })
  if (paid === null) return step.run("cancel-order", () => shop.cancel(input.orderId))
  await step.run("ship", () => shop.ship(input.orderId, JSON.parse(paid.body)))
})
```
### Go

```go
paid, ok, err := workflows.WaitForMessage(ctx, "paid", workflows.WaitForMessageOptions{
	QueueID: os.Getenv("PAYMENTS_QUEUE_ID"),
	Key:     "order:" + in.OrderID,
	Timeout: 24 * time.Hour,
})
if err != nil {
	return err
}
if !ok {
	return cancelOrder(ctx, in.OrderID) // nobody paid in time
}
```
### Python

```python
paid = await ctx.step.wait_for_message(
    "paid", queue_id=os.environ["PAYMENTS_QUEUE_ID"], key=f"order:{order_id}", timeout_seconds=86_400,
)
if paid is None:
    return await ctx.step.run("cancel-order", lambda: shop.cancel(order_id))
await ctx.step.run("ship", lambda: shop.ship(order_id, json.loads(paid["body"])))
```

- **The queue** is named by its ID (a UUID). **The key** is 1 to 256 letters, digits and
  `_ . : -`; make it name what the run waits for, such as `order:1042`.
- **The message** arrives as `{ messageId, body, attributes, sentAt, queueId, groupId?, seq? }`
  (`groupId` and `seq` on FIFO queues). It's recorded with the step, so a replay gets the same
  message.
- **A timeout isn't an error.** After `timeoutSeconds` (1 second to 30 days) without a message, the
  step completes with nothing: `null` in TypeScript, `ok` false (and a nil error) in Go, `None` in
  Python. Handle it like any other result.
- **Cancelling** the run ends a pending wait with the SDK's cancellation error.
- Waits need Workflows SDK v0.3.0 and later (protocol 1.2): on an earlier SDK the run can't
  wait for a message.

### The resume subscription

The queue the run waits on needs a subscription in resume mode for the app's workflow. Its
**correlation attribute** names the message attribute whose value is the key: with
`correlationAttribute: "orderKey"`, a message with the attribute `orderKey: order:1042` resumes the
runs waiting on `order:1042`.

### Console

Open the queue, then **Subscription**, and **Create subscription**. Choose **Resume waiting
runs**, pick the Workflows app and the workflow, and enter the correlation attribute.
### CLI

```sh
ezgh queues subscriptions create payments --workflow-app billing-jobs --workflow order \
  --mode resume --correlation-attribute orderKey
```
### API

```sh
curl -sS -X POST "https://queues-api.ezghcloud.com/v1/organizations/$ORG_ID/projects/$PROJECT_ID/subscriptions" \
  -H "Authorization: Bearer $EZGH_API_KEY" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d "{ \"queueId\": \"$QUEUE_ID\",
        \"target\": { \"kind\": \"workflow\", \"mode\": \"resume\", \"appId\": \"$APP_ID\",
                      \"workflow\": \"order\", \"correlationAttribute\": \"orderKey\" } }"
```

Creating it needs `queues.subscriptions.create` on the queue and `workflows.runs.resume` on the
workflow (`<the app's resource name>/<workflow>`). Then send the message with the attribute:

```sh
curl -sS -X POST "https://queues-api.ezghcloud.com/v1/organizations/$ORG_ID/projects/$PROJECT_ID/queues/$QUEUE_ID/messages" \
  -H "Authorization: Bearer $EZGH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "messages": [ { "body": "{\"amount\":\"19.99\"}", "attributes": { "orderKey": "order:1042" } } ] }'
```

### How messages reach waiting runs

- **Every run waiting at that moment.** A message resumes every run of the app waiting on that
  queue and key when it's delivered, each with the same message, and is then acknowledged. A run
  that starts waiting afterwards waits for the next message.
- **Nobody waiting yet.** A message that no run is waiting for, or without the correlation
  attribute (or with a value that isn't a valid key), isn't acknowledged: it's delivered again with
  the subscription's backoff, so a run that starts waiting within the retry window still gets it.
  After the queue's `maxReceiveCount` deliveries it moves to the
  [dead-letter queue](/queues/dead-letter-queues). The subscription's status shows these as
  **No run to resume** (`message`).
- **So send after the run waits, or let retries cover the gap.** If a message can arrive before
  the run reaches its wait, give the queue a `maxReceiveCount` high enough for the retries to
  cover that time, or the message ends up in the dead-letter queue, from which a
  [redrive](/queues/dead-letter-queues#redrive-messages-back) sends it again.
- **One message per call**, up to `maxConcurrency` at once; on a FIFO queue, a group's next
  message waits for the previous one's delivery.

## Send messages from a workflow

`sendToQueue` (Workflows SDK v0.3.0 and later) sends 1 to 10 messages to a queue as a step. It returns the recorded result, one
entry per message (`messageId`, and on FIFO queues `seq`), so a replay doesn't send again. It
needs the Workflows SDK's `QueuesClient`, with an API key that has `queues.messages.send` on the
queue: push apps hold no API key of their own, so the step never borrows one.

### TypeScript

```ts
import { QueuesClient, workflow } from "@ezgh/workflows"

const queues = new QueuesClient({
  apiKey: process.env.EZGH_API_KEY!,
  organizationId: process.env.EZGH_ORG_ID!,
  projectId: process.env.EZGH_PROJECT_ID!,
})

export const order = workflow("order", async ({ input, step }) => {
  const placed = await step.run("place", () => shop.place(input))
  await step.sendToQueue("notify", queues, process.env.ORDERS_QUEUE_ID!, [
    { body: JSON.stringify(placed), attributes: { orderId: placed.id } },
  ])
})
```
### Go

```go
queues, err := workflows.NewQueuesClient(workflows.QueuesClientOptions{
	APIKey:         os.Getenv("EZGH_API_KEY"),
	OrganizationID: os.Getenv("EZGH_ORG_ID"),
	ProjectID:      os.Getenv("EZGH_PROJECT_ID"),
})

// In the workflow:
sent, err := workflows.SendToQueue(ctx, "notify", queues, os.Getenv("ORDERS_QUEUE_ID"), []workflows.QueueMessage{
	{Body: string(placed), Attributes: map[string]string{"orderId": id}},
})
```
### Python

```python
from ezgh_workflows import Context, QueuesClient, workflow

queues = QueuesClient(
    api_key=os.environ["EZGH_API_KEY"],
    organization_id=os.environ["EZGH_ORG_ID"],
    project_id=os.environ["EZGH_PROJECT_ID"],
)


@workflow("order")
async def order(ctx: Context) -> None:
    placed = await ctx.step.run("place", lambda: shop.place(ctx.input))
    await ctx.step.send_to_queue(
        "notify", queues, os.environ["ORDERS_QUEUE_ID"],
        [{"body": json.dumps(placed), "attributes": {"orderId": placed["id"]}}],
    )
```

The queue is named by its ID (a UUID). The client's base URL is `https://queues-api.ezghcloud.com`.

- **Sent once.** Every attempt of the step sends the same `Idempotency-Key`,
  `wfq_<hex SHA-256 of "<runId>:<stepId>">`, which Queues honors for 48 hours: an attempt that
  sends again after a crash gets the first answer instead of storing a second copy. Keep the
  messages the same on every attempt; the same key with a different body is
  `409 idempotency_conflict`.
- **FIFO queues.** Give each message a `groupId`. A message without its own `dedupId` gets
  `<that hex>:<index>`.
- **Errors.** The client resends `429` (except `quota_exceeded`), `5xx` and network errors after
  `Retry-After`, up to 3 times, before the step's retry policy takes over. Any other `4xx` (a bad
  message, a missing queue, a refused permission) fails the step at once, without retries.

`client.send` works outside workflows too. See each SDK's README for the details.

## Run-result notifications

An app can send a message to a queue each time one of its runs ends: a dashboard, an alert or
another service then receives run results like any other messages, instead of polling.

### Console

Open **Workflows**, the app, then **Settings**. In the **Notifications** card, pick the queue, the
run statuses, every workflow or only some, and on a FIFO queue the message group, then **Save**.
**Turn off** stops them.
### CLI

```sh
ezgh workflows apps update billing-jobs --notify-queue run-results --notify-on failed,timedOut
ezgh workflows apps update billing-jobs --no-notify
```
### API

```sh
curl -sS -X PATCH "https://workflows-api.ezghcloud.com/v1/organizations/$ORG_ID/projects/$PROJECT_ID/apps/$APP_ID" \
  -H "Authorization: Bearer $EZGH_API_KEY" \
  -H "Content-Type: application/json" \
  -d "{ \"notifications\": { \"queueId\": \"$QUEUE_ID\", \"on\": [\"failed\", \"timedOut\"], \"workflows\": null } }"
```

[UpdateApp](/workflows-api/UpdateApp/) with `"notifications": null` turns them off.

- **`on`:** the statuses that send a message: `completed`, `failed`, `cancelled`, `timedOut`.
- **`workflows`:** only these workflows' runs, or `null` for every workflow of the app.
- **`groupBy`:** on a FIFO queue, each message's `groupId`: `run` (the run's ID, the default) or
  `workflow` (the workflow's name, so a workflow's results arrive in the order its runs ended).
- **Who may set it:** `workflows.apps.update` on the app, and `queues.messages.send` on the queue.
  The queue is in the app's project.

Each message's body is JSON:

```json
{
  "type": "workflows.run.ended",
  "runId": "01999f60-4d5e-7f6a-8b9c-0d1e2f3a4b5c",
  "appId": "01999f3b-9e8d-7c6b-a5f4-e3d2c1b0a9f8",
  "workflow": "chargeCustomer",
  "status": "failed",
  "attempt": 1,
  "error": { "code": "user", "name": "CardDeclinedError", "message": "Card declined" },
  "startedAt": "2026-10-13T18:00:00Z",
  "endedAt": "2026-10-13T18:01:35Z"
}
```

- **`output`** is a completed run's output, inlined when the message stays within Queues' 256 KiB.
  A larger output is `outputRef` instead: the path of
  [the run's output](/workflows-api/GetRunOutput/) to read with `workflows.runs.getPayload`.
- **`error`** is a failed or timed-out run's error.
- **`attempt`** is the run's attempt: a [retried run](/workflows/retrying-runs) that ends again
  sends another message.
- **Once per ending.** Each message is sent with the `Idempotency-Key`, and on a FIFO queue the
  `dedupId`, `<runId>:<attempt>:<status>`, so a resend never stores it twice.
- **When Queues can't take it,** sending is retried with backoff for up to 24 hours after the run
  ended, then dropped. A queue that's been deleted drops it at once.

Source: https://docs.ezghcloud.com/queues/workflows/index.mdx