---
title: "Push subscriptions"
description: "Have Queues push batches of messages to your HTTPS endpoint, verify their signatures, and report partial failures."
---

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

# Push subscriptions

A subscription is a queue's managed consumer: Queues receives the queue's messages for you and
POSTs them in batches to your HTTPS endpoint. Your endpoint answers `2xx` to acknowledge a batch,
or lists the messages it couldn't handle. A queue has at most one subscription.

## Create a subscription

```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" \
  -d '{
    "queueId": "01999f3c-7a1e-7b2a-9c4d-5e6f7a8b9c0d",
    "target": { "kind": "https", "url": "https://hooks.example.com/queues/orders", "timeoutSeconds": 30 },
    "batchSize": 10,
    "maxConcurrency": 10
  }'
```

| Field | Default | Meaning |
| --- | --- | --- |
| `queueId` | | The queue to consume. Fixed |
| `target.url` | | Your endpoint ([rules below](#endpoints-queues-calls)) |
| `target.timeoutSeconds` | 30 | How long a delivery may take, 1 to 60 seconds |
| `batchSize` | 10 | Messages per delivery, at most; 1 to 10 |
| `maxConcurrency` | 10 | Deliveries in flight at once, 1 to 100 |
| `state` | `active` | `paused` creates it without delivering |

[CreateSubscription](/queues-api/CreateSubscription/) is checked as
`queues.subscriptions.create` on the queue, and answers `409 subscription_exists` if the queue
already has one. **The answer carries the signing secret** (`signingSecret`, `whsec_…`) **once:**
store it where your endpoint can read it. A repeated create with the same `Idempotency-Key`
answers `signingSecret: null`; if you lost the secret, [rotate it](#rotate-the-signing-secret).

Deliveries start at once, and pull consumers can still receive from the queue: they and the
subscription compete for messages.

With the CLI: `ezgh queues subscriptions create orders --url https://hooks.example.com/q`
prints the signing secret once; `ezgh queues subscriptions` also lists, pauses, resumes, changes,
deletes and rotates the secret of subscriptions.

## Endpoints Queues calls

- `https://` on port 443, with a publicly trusted certificate. No user name, password or
  fragment in the URL, at most 2,048 characters.
- A public DNS name: not an IP address, a single label, `localhost`, or a name under `.local`,
  `.internal`, `.test`, `.svc`, `.localhost`, `ezghcloud.com` or `ezgh.app`.
- At every delivery, the name must resolve to public addresses only. Private, loopback,
  link-local and shared addresses are refused.
- Redirects aren't followed: a `3xx` answer is a failed delivery.

## The request

Each delivery is a `POST` with `Content-Type: application/json`, these headers:

| Header | Value |
| --- | --- |
| `EZGH-Signature` | `t=<unix seconds>,v1=<hex HMAC-SHA256>`, one `v1` per active signing secret. See [Verify the signature](#verify-the-signature) |
| `EZGH-Delivery-Id` | This delivery's ID, also in the body as `deliveryId` |

and a body like this:

```json
{
  "subscription": "qsb_7m2k9x0q4r1s",
  "queue": "que_4k2m9x0a7qpe",
  "deliveryId": "01999f41-6c3b-7a2d-9e1f-0a1b2c3d4e5f",
  "messages": [
    {
      "messageId": "01999f3d-02b4-7c1d-8e2f-3a4b5c6d7e8f",
      "body": "{\"orderId\": \"o-123\"}",
      "attributes": { "source": "checkout" },
      "sentAt": "2026-10-13T17:02:40Z",
      "receiveCount": 1
    }
  ]
}
```

`messages` holds 1 to `batchSize` messages. On a FIFO queue each message also has `groupId` and
`seq`, and a batch can hold several groups, each in order. `receiveCount` counts delivery attempts
and pull receives together.

## Verify the signature

Check `EZGH-Signature` on every request before you trust its body. It's the scheme Workflows push
apps use:

1. Split the header on `,` into `t=` (Unix seconds) and one or more `v1=` (hex).
2. Refuse the request if `t` is more than 300 seconds from now, either way.
3. Decode each signing secret: the text after `whsec_` is the base64url (unpadded) encoding of the
   32-byte HMAC key.
4. Compute `HMAC-SHA256(key, t + "." + body)` over the **raw body bytes**, before parsing them.
   The request is genuine if any `v1` equals it for any of your secrets. Compare in constant time.

### Node.js

```js title="server.mjs"
import { createServer } from "node:http"
import { createHmac, timingSafeEqual } from "node:crypto"

// During a rotation, list the new secret and the previous one.
const secrets = [process.env.EZGH_QUEUES_SIGNING_SECRET]

function verifySignature(header, body, secrets, now = Date.now()) {
  let t = -1
  const signatures = []
  for (const part of (header ?? "").split(",")) {
    const [key, value = ""] = part.trim().split("=")
    if (key === "t" && /^\d+$/.test(value)) t = Number(value)
    if (key === "v1" && /^[0-9a-f]{64}$/.test(value)) signatures.push(Buffer.from(value, "hex"))
  }
  if (t < 0 || signatures.length === 0) return false
  if (Math.abs(now / 1000 - t) > 300) return false // more than 5 minutes old or ahead
  return secrets.some((secret) => {
    const key = Buffer.from(secret.slice("whsec_".length), "base64url")
    const expected = createHmac("sha256", key).update(`${t}.`).update(body).digest()
    return signatures.some((signature) => timingSafeEqual(expected, signature))
  })
}

createServer(async (req, res) => {
  const chunks = []
  for await (const chunk of req) chunks.push(chunk)
  const body = Buffer.concat(chunks)
  if (!verifySignature(req.headers["ezgh-signature"], body, secrets)) {
    res.writeHead(401).end()
    return
  }
  const delivery = JSON.parse(body.toString("utf8"))
  const failed = []
  for (const message of delivery.messages) {
    try {
      await handle(message) // your code
    } catch {
      failed.push(message.messageId)
    }
  }
  res.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ failed }))
}).listen(8080)
```
### Go

```go title="deliveries.go"
package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"
)

func verifySignature(header string, body []byte, secrets []string, now time.Time) bool {
	t := int64(-1)
	var signatures [][]byte
	for _, part := range strings.Split(header, ",") {
		key, value, _ := strings.Cut(strings.TrimSpace(part), "=")
		switch key {
		case "t":
			if n, err := strconv.ParseUint(value, 10, 63); err == nil {
				t = int64(n)
			}
		case "v1":
			if b, err := hex.DecodeString(value); err == nil && len(b) == sha256.Size {
				signatures = append(signatures, b)
			}
		}
	}
	if t < 0 || len(signatures) == 0 {
		return false
	}
	if age := now.Unix() - t; age > 300 || age < -300 { // more than 5 minutes old or ahead
		return false
	}
	for _, secret := range secrets {
		key, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
		if err != nil {
			continue
		}
		mac := hmac.New(sha256.New, key)
		mac.Write([]byte(strconv.FormatInt(t, 10) + "."))
		mac.Write(body)
		expected := mac.Sum(nil)
		for _, signature := range signatures {
			if hmac.Equal(expected, signature) {
				return true
			}
		}
	}
	return false
}

type message struct {
	MessageID  string            `json:"messageId"`
	Body       string            `json:"body"`
	Attributes map[string]string `json:"attributes"`
}

// deliveries serves a subscription's endpoint. During a rotation, pass the new secret and the
// previous one.
func deliveries(secrets []string, handle func(*http.Request, message) error) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 16<<20))
		if err != nil {
			http.Error(w, "unreadable body", http.StatusBadRequest)
			return
		}
		if !verifySignature(r.Header.Get("EZGH-Signature"), body, secrets, time.Now()) {
			http.Error(w, "bad signature", http.StatusUnauthorized)
			return
		}
		var delivery struct {
			Messages []message `json:"messages"`
		}
		if err := json.Unmarshal(body, &delivery); err != nil {
			http.Error(w, "bad JSON", http.StatusBadRequest)
			return
		}
		failed := []string{}
		for _, m := range delivery.Messages {
			if err := handle(r, m); err != nil {
				failed = append(failed, m.MessageID)
			}
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string][]string{"failed": failed})
	}
}
```
### Python

```python title="app.py"
import base64
import hashlib
import hmac
import json
import os
import time

from fastapi import FastAPI, Request, Response


def verify_signature(header, body, secrets, now=None):
    t, signatures = None, []
    for part in (header or "").split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isascii() and value.isdigit():
            t = int(value)
        elif key == "v1" and len(value) == 64 and value.isascii():
            signatures.append(value)
    if t is None or not signatures:
        return False
    if abs((time.time() if now is None else now) - t) > 300:  # more than 5 minutes old or ahead
        return False
    for secret in secrets:
        encoded = secret.removeprefix("whsec_")
        key = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
        expected = hmac.new(key, f"{t}.".encode() + body, hashlib.sha256).hexdigest()
        if any(hmac.compare_digest(expected, s) for s in signatures):
            return True
    return False


app = FastAPI()
# During a rotation, list the new secret and the previous one.
SECRETS = [os.environ["EZGH_QUEUES_SIGNING_SECRET"]]


@app.post("/queues/orders")
async def deliveries(request: Request):
    body = await request.body()
    if not verify_signature(request.headers.get("EZGH-Signature"), body, SECRETS):
        return Response(status_code=401)
    failed = []
    for message in json.loads(body)["messages"]:
        try:
            await handle(message)  # your code
        except Exception:
            failed.append(message["messageId"])
    return {"failed": failed}
```

## Answer the delivery

Answer within the subscription's `timeoutSeconds`:

| Your answer | What happens |
| --- | --- |
| `2xx` | The whole batch is acknowledged, whatever the body says, unless it's the JSON below |
| `2xx` with `{"failed": ["<messageId>", …]}` | Every message except those listed is acknowledged. The listed ones are retried. IDs that weren't in the batch are ignored |
| Any other status, `3xx` included | The whole batch is retried |
| No answer within `timeoutSeconds`, or a connection or TLS error | The whole batch is retried |

A `2xx` body that isn't JSON, or isn't that shape, acknowledges the whole batch: list failures
only in `failed`. On a FIFO queue a failed message also fails the later messages of its group in
the same batch, so they're retried after it, in order ([FIFO queues](/queues/fifo#push-subscriptions)).

While a delivery is open, its messages are hidden for the longer of the queue's visibility timeout
and `timeoutSeconds` plus 30 seconds, so a slow endpoint doesn't get a second copy while it's
still working. Deliveries are at least once all the same: an endpoint that handled a message but
didn't answer in time gets it again. Use `messageId` to recognize repeats.

## Retries

- **A failed message** becomes visible again after `2^(receiveCount − 1)` seconds (1, 2, 4, 8,
  16 …), with ±20 % jitter, and at most the queue's visibility timeout. It's then delivered again,
  or received by a pull consumer.
- **`429` or `503` with `Retry-After`** (in seconds): the batch waits that long, at most the
  queue's visibility timeout.
- **Each attempt counts as a receive.** With a [redrive policy](/queues/dead-letter-queues), a
  message moves to the dead-letter queue after `maxReceiveCount` attempts. Without one, it's
  retried until the queue's retention ends.
- **Slowing down:** after a `429`, a `503` or a timeout, Queues halves the subscription's
  deliveries in flight, then adds one back per successful delivery, up to `maxConcurrency`.

## Status

[GetSubscription](/queues-api/GetSubscription/) shows how deliveries are going:

| Field | Meaning |
| --- | --- |
| `status.health` | `healthy`, or `failing` after a failed delivery until one succeeds |
| `status.lastError` | The last failure: `at`, `class` (`http`, `timeout`, `connection`, `refused` or `tls`), the endpoint's `status` when it answered, and a `message` of at most 200 characters. `refused` is an address or name the [endpoint rules](#endpoints-queues-calls) don't allow |
| `status.lastSuccessAt` | The last successful delivery |

Individual delivery attempts aren't kept.

## Pause, resume, change and delete

- **[PauseSubscription](/queues-api/PauseSubscription/)** stops deliveries. Messages wait in
  the queue, or go to pull consumers. **[ResumeSubscription](/queues-api/ResumeSubscription/)**
  starts them again. Both need `queues.subscriptions.pause`.
- **[UpdateSubscription](/queues-api/UpdateSubscription/)** changes `target`, `batchSize`
  or `maxConcurrency`. The queue is fixed.
- **[DeleteSubscription](/queues-api/DeleteSubscription/)** stops deliveries; the queue and
  its messages stay.

## Rotate the signing secret

[RotateSubscriptionSecret](/queues-api/RotateSubscriptionSecret/) returns a new secret,
once, and needs `queues.subscriptions.update`. For 24 hours, deliveries are signed with both the
new and the previous secret (two `v1` values), then with the new one only. Deploy the new secret
alongside the old, then remove the old one.

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