Queues and Workflows

EZGH Queues and EZGH Workflows work together in four ways, all within one project:

• Start a run per message: 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: 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: a step sends messages to a queue, exactly once, with sendToQueue.
• 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: each message starts a run of the workflow, and is acknowledged once the start is accepted.

ezgh queues subscriptions create orders --workflow-app billing-jobs --workflow chargeCustomer

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 }"

• 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:

{
  "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, except for the caller’s permission check:

A subscription’s 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

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

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
}

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.

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

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:

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

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 } },
  ])
})

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}},
})

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.

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

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 } }"

• 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:

{
  "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 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 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.