# queue

> Durable function queues - registers the `durable:subscriber` trigger type and the queue/DLQ service functions.

| field | value |
|-------|-------|
| version | 0.21.20 |
| type | binary |
| license | Apache-2.0 |
| repo | https://github.com/iii-hq/workers |
| supported_targets | x86_64-apple-darwin, aarch64-apple-darwin, x86_64-pc-windows-msvc, aarch64-pc-windows-msvc, x86_64-unknown-linux-gnu, aarch64-unknown-linux-gnu, x86_64-unknown-linux-musl, armv7-unknown-linux-gnueabihf |
| author | iii |

## installation

```sh
iii trigger compose::add worker=queue@0.21.20
```

## configuration

```yaml
- adapter:
    config:
      file_path: data/queue
      save_interval_ms: 5000
      store_method: file_based
    name: builtin
```

## dependencies

- `configuration` @ `latest`

## readme

# queue

Durable function queues for iii. This worker registers the
`durable:subscriber` trigger type and the queue/DLQ service functions that
replace the legacy built-in queue worker.

## Install

```bash
iii trigger compose::add worker=queue
```

`iii trigger compose::add` resolves the worker and its dependencies, writes
exact declarations to `worker-compose.yaml`, and reconciles the Compose project.

## Trigger Type

Bind a function to `durable:subscriber` to consume a topic/queue:

| Field | Required | Default | Description |
|---|---|---|---|
| `queue` | yes | - | Topic/queue name to consume. `topic` is accepted as a compatibility alias. |
| `max_retries` | no | `3` | Maximum failed deliveries before the message moves to DLQ. |
| `backoff_ms` | no | `1000` | Base retry delay in milliseconds. Retries use exponential backoff. |
| `condition_function_id` | no | - | Function invoked first. Only explicit `false` skips the handler. |

The worker also accepts the built-in subscriber `queue_config` shape for
compatibility, including `maxRetries` and `backoffDelayMs`.

## Functions

| Function id | Input | Output |
|---|---|---|
| `queue::define` | `{ "queue", "config" }` | `{ "queue", "changed" }` |
| `engine::queue::enqueue` | `{ "queue", "function_id", "data", "messageReceiptId", "traceparent"?, "baggage"? }` | `{ "messageReceiptId" }` |
| `iii::durable::publish` | `{ "queue" \| "topic", "data" }` | `null` |
| `iii::queue::redrive` | `{ "queue" \| "topic" }` | `{ "queue", "redriven" }` |
| `iii::queue::redrive_message` | `{ "queue" \| "topic", "message_id" }` | `{ "queue", "message_id", "redriven" }` |
| `iii::queue::discard_message` | `{ "queue" \| "topic", "message_id" }` | `{ "queue", "message_id", "redriven" }` |
| `engine::queue::list_topics` | `{}` | topic list |
| `engine::queue::topic_stats` | `{ "topic" \| "queue" }` | `{ "depth", "consumer_count", "dlq_depth", "config" }` |
| `engine::queue::dlq_topics` | `{}` | DLQ topic list |
| `engine::queue::dlq_messages` | `{ "topic" \| "queue", "offset", "limit" }` | DLQ messages |

### Namespace connections

When the worker runs in a project namespace, the process opens two engine
connections. This keeps project functions isolated while engine queue provider
functions remain available from `default`.

| Worker name | Namespace | Registered functions |
|---|---|---|
| `queue` | project namespace | `queue::define`, publish, redrive, redrive-message, and discard functions |
| `queue-engine` | `default` | `engine::queue::enqueue`, list, stats, and DLQ query functions |

Both names therefore appear in `engine::workers::list`. They belong to one
queue process and stop together.

## Configuration

Configuration is owned by the `configuration` worker under id `queue`.
Seed it once with `--config <file>.yaml`; runtime edits come from the
configuration worker after that.

Named queues live under `queue_configs` and can also be converged at runtime
with `queue::define`. Definitions are durable: the worker recreates their
topology and consumers on restart. A definition succeeds only after its
consumer is ready and the merged configuration has been persisted.

```yaml
queue_configs:
  harness-turn:
    type: fifo
    message_group_field: session_id
    concurrency: 10
    max_retries: 3
    backoff_ms: 1000
```

For FIFO named queues, messages with the same group-field value run in order;
different groups run concurrently up to `concurrency`.
Each target invocation has a configurable `timeout_ms`; when omitted, it
defaults to 1,800,000 milliseconds (30 minutes).
After a restart, a delivery waits until its target function is registered and
does not consume retry budget while the target worker is still booting. The
wait is event-driven: the worker binds the engine's
`engine::functions-available` and `engine::workers-available` triggers (to the
internal `queue::on-engine-change` function) and re-checks the target once per
change, never on a timer. The same feed drives the
`redeliver_on_engine_restart` restart check.

`poll_interval_ms` is deprecated and ignored (still accepted so existing
definitions validate): consumers wake on enqueue, and a backed-off retry is
delivered at exactly its due time.

`adapter.name` selects the transport: `builtin` (default), `redis`, or
`rabbitmq`. Changing the adapter config hot-swaps the transport and
restarts every consumer — the new adapter is built first, so a bad config
leaves the previous adapter serving. Pending in-memory jobs are lost on
swap/restart; file-backed jobs survive. An unreachable `redis`/`rabbitmq`
target at boot fails the boot itself (no fallback to `builtin`).

### Adapters

#### `builtin` (default)

In-process, single-worker transport. Full fan-out (every subscriber on a
topic receives every published message), retries, DLQ, redrive/discard,
and both `fifo` and `concurrent` subscriber modes. The legacy aliases
`in_memory` and `file_based` are also accepted as `adapter.name` and both
resolve to this transport. Consumers do not poll the store: an idle consumer
parks until an enqueue, retry, requeue, or redrive on its queue wakes it, or
until the earliest backed-off retry reaches its due time.

```yaml
adapter:
  name: builtin
  config:
    store_method: file_based
    file_path: data/queue
    save_interval_ms: 5000
```

| Field | Default | Description |
|---|---|---|
| `adapter.config.store_method` | `file_based` | `in_memory` or `file_based`. |
| `adapter.config.file_path` | `data/queue` | Directory used by `file_based`. |
| `adapter.config.save_interval_ms` | `5000` | Accepted for parity; this worker persists on mutation rather than on an interval. |

#### `redis`

Pub/sub only — 1:1 port of the engine builtin's `RedisAdapter`, including
its limitations. There is no DLQ, no retries, and no message durability: a
message published with no subscriber connected is lost.
Redis does not currently support named function-queue consumers, so a
persisted `queue_configs` entry fails boot with an explicit unsupported error.

```yaml
adapter:
  name: redis
  config:
    redis_url: redis://localhost:6379
```

| Field | Default | Description |
|---|---|---|
| `adapter.config.redis_url` | `redis://localhost:6379` | Redis connection string. |

Every DLQ operation (`redrive`, `redrive_message`, `discard_message`,
`dlq_count`, `dlq_peek`) returns the error
`RedisAdapter does not support DLQ operations (pub/sub only)`, verbatim
from the engine.

#### `rabbitmq`

Full-featured transport: retries, DLQ, redrive, message priority, and both
standard and FIFO queue modes. On by default (the `rabbitmq` cargo feature
is feature-default-on); building with `--no-default-features` drops it,
and selecting `adapter.name: rabbitmq` in that build fails boot with an
error naming the missing feature.

RabbitMQ's `x-max-priority` queue argument is immutable. An existing named
queue cannot change `max_priority` in place; delete and recreate its broker
topology first.

```yaml
adapter:
  name: rabbitmq
  config:
    amqp_url: amqp://localhost:5672
    max_attempts: 3
    prefetch_count: 10
    queue_mode: standard
    priority_field: priority
```

| Field | Default | Description |
|---|---|---|
| `adapter.config.amqp_url` | `amqp://localhost:5672` | AMQP connection string. |
| `adapter.config.max_attempts` | `3` | Delivery attempts before a message moves to DLQ. The retry budget is stamped on each message at publish time from this adapter-level value, so per-subscriber `queue_config.maxRetries` does not override it on this transport (it applies to the builtin adapter). |
| `adapter.config.prefetch_count` | `10` | Consumer prefetch (QoS), used in `standard` queue mode. |
| `adapter.config.queue_mode` | `standard` | `standard` or `fifo`. Unrecognized values fall back to `standard`. |
| `adapter.config.priority_field` | none | JSON field read from each published message's payload to set the AMQP message priority. Only applies to subscribers whose `queue_config.maxPriority` declares the queue as a priority queue (`x-max-priority`). |

Per-subscriber queue tuning uses the trigger's `queue_config` below;
`maxPriority` is RabbitMQ-only and ignored by the other adapters.

#### `memory` (dev/test only)

An in-process, test-only transport (`adapters::memory::MemoryAdapter`)
used by this worker's own hot-swap tests. It is gated behind the
`test-adapters` cargo feature, off in normal builds, and not a supported
`adapter.name` value for real deployments.

### Trigger `queue_config`

The `durable:subscriber` trigger's `queue_config` accepts the full builtin
`SubscriberQueueConfig` shape:

| Field | Default | Description |
|---|---|---|
| `type` | `concurrent` | `fifo` (strictly serial) or `concurrent`. |
| `maxRetries` | `3` | Failed deliveries before DLQ, per subscriber (builtin adapter; RabbitMQ uses adapter-level `max_attempts`). |
| `concurrency` | `10` (builtin) | In-flight invocations for `concurrent` mode. `0` pauses consumption. Ignored by `fifo` (always serial — this worker has no grouped-fifo partitioning; see Known Gaps). |
| `visibilityTimeout` | — | Accepted for parity. |
| `delaySeconds` | — | Accepted for parity. |
| `backoffType` | — | Accepted for parity. |
| `backoffDelayMs` | `1000` | Base retry delay; exponential backoff. |
| `maxPriority` | none | RabbitMQ-only: declares the subscriber's queue as an AMQP priority queue with this many levels (1-255). |

## Engine Compatibility

Current engines no longer load the legacy built-in by default. Engines with the
`QueueEnqueuer` cut route `TriggerAction::Enqueue` through this worker's
`engine::queue::enqueue` provider. Until that support is available, use
`iii::durable::publish` with `durable:subscriber` triggers instead of named
function queues.

When connecting to an older engine, remove the legacy built-in from its config first;
two owners of `durable:subscriber` cannot run together.

On boot, this worker queries `engine::workers::list` and refuses to start if
the legacy built-in is active.

## Parity Vs Builtin

| Behavior | Builtin | This worker |
|---|---|---|
| Trigger type | `durable:subscriber` | same |
| Function ids | 8 legacy public ids | same 8 plus `queue::define` and the engine enqueue provider |
| Retry | `max_retries` + exponential `backoff_ms * 2^(attempts - 1)` | same |
| DLQ | after retries exhausted; redrive/redrive_message/discard | same |
| Restart survival | file-backed store | same (`file_based`) |
| In-memory mode | jobs lost on restart/swap | same |
| Transports | builtin (memory/file), redis (pub/sub, no DLQ), rabbitmq (full: retry/DLQ/priority/fifo) | same as builtin |
| bridge adapter | engine-internal (`TriggerAction::Enqueue`) | replaced by `engine::queue::enqueue` |
| Enqueue failure when worker offline | n/a, in-process | invocation fails explicitly through the engine provider route |
| Latency benchmark | in-process baseline required | final budget is a pre-deprecation gate (tracked in the migration master plan) |

The `TriggerAction::Enqueue` path requires an engine version that routes named
enqueue actions through the registered `engine::queue::enqueue` provider.

## Known Gaps / Parity Notes

- **`bridge` adapter — not ported.** It was engine-internal plumbing for
  `TriggerAction::Enqueue`; the registered `engine::queue::enqueue` provider
  supersedes that path.
- **Subscriber FIFO remains global.** Grouped FIFO is implemented for named
  function queues created by `queue::define`; `durable:subscriber` keeps its
  existing strictly serial FIFO behavior.
- **`resolve_dlq_name` bare-topic behavior ported verbatim, bug and all.**
  DLQ operations (`redrive`, `dlq_count`, `topic_stats`, `dlq_peek`)
  address a bare topic name and, on the builtin adapter, aggregate across
  every subscriber's internal queue on that topic — matching the engine's
  behavior (and its naming quirk) exactly, rather than fixing it here.
  On the rabbitmq transport this makes DLQ browse/redrive against a
  subscriber topic 404 (retries write to the per-function DLQ), and the
  AMQP channel closure stops every consumer on the worker. Upstream engine
  fix tracked as MOT-3904; this worker follows once the engine lands it.
- **`dlq_topics` filters to `dlq_count > 0`.** The engine's equivalent also
  iterates every known topic but returns all of them regardless of DLQ
  depth; this worker only returns topics that currently have dead-lettered
  messages. Documented divergence, not a bug to reconcile.
- **Fifo retry via `nack` can be overtaken by newer arrivals.** A failed
  fifo message is re-queued via `nack` rather than blocking the consumer
  in-place (the engine's `FifoWorker` blocks); a message enqueued after
  the failure can be delivered before the retried one catches up.

## skills

This version ships 1 skill file for agents, also browsable on the worker page:

- [`SKILL.md`](https://workers.iii.dev/workers/queue?tab=skills&skill=SKILL.md&version=0.21.20)

### SKILL.md

# queue

The `queue` worker provides durable topic delivery for iii. Register a consumer
with a `durable:subscriber` trigger, publish with `iii::durable::publish`, and
the worker persists messages, retries failed deliveries, moves exhausted jobs to
DLQ, and exposes redrive/discard inspection functions.

Install it with `iii trigger compose::add worker=queue`. The legacy built-in queue service must be
removed from the engine config before this worker starts because both would own
the same `durable:subscriber` trigger type.

## When to Use

- A topic event must be processed reliably by a subscribed function.
- A failed consumer should retry automatically before the message lands in DLQ.
- You need to inspect, redrive, or discard dead-lettered messages.
- Local or single-instance durable queues are enough, using in-memory storage
  for development or file-backed storage for restart survival.

## Boundaries

- Not fire-and-forget broadcast. Use `pubsub` when missed events are acceptable.
- Three transports ship today: `builtin` (in-process, full DLQ/retry/fifo),
  `redis` (pub/sub only, no DLQ), and `rabbitmq` (full: retry/DLQ/priority/fifo).
  The `bridge` adapter is not ported — it was engine-internal and is superseded
  by the engine's `QueueEnqueuer` cut (MOT-3829).
- Engine `TriggerAction.Enqueue` / named function queues require the separate
  `QueueEnqueuer` engine cut. Until that lands, use `iii::durable::publish` and
  `durable:subscriber` triggers for this standalone worker.
- File-backed mode survives worker restarts. In-memory mode loses pending jobs
  on restart or transport hot-swap. An unreachable `redis`/`rabbitmq` target at
  boot fails the boot (no fallback to `builtin`).
- `fifo` mode is strictly serial — there is no grouped-fifo partitioning by
  `group_id` like the engine's `GroupedFifoWorker`.
- This worker carries work items, not key/value or stream state. Use `state` or
  `stream` for those.

## Functions

- `iii::durable::publish`: publish a JSON payload to a queue/topic. Input accepts
  `{ "queue": "...", "data": ... }` and also accepts `topic` as an alias.
- `iii::queue::redrive`: move all DLQ messages for a queue back to the main
  queue. Input accepts `queue` or `topic`.
- `iii::queue::redrive_message`: move one DLQ message back by `message_id`.
- `iii::queue::discard_message`: purge one DLQ message by `message_id`.
- `engine::queue::list_topics`: list known queue topics.
- `engine::queue::topic_stats`: return `depth`, `consumer_count`, and
  `dlq_depth` for one topic.
- `engine::queue::dlq_topics`: list DLQ topics with message counts.
- `engine::queue::dlq_messages`: browse DLQ messages with `offset` and `limit`.

## Reactive Triggers

Bind a `durable:subscriber` trigger when a function should consume every
message published to a queue/topic with retries and DLQ handling.

```typescript
iii.registerTrigger({
  type: 'durable:subscriber',
  function_id: 'orders::process',
  config: {
    queue: 'orders.created',
    max_retries: 3,
    backoff_ms: 1000,
  },
})
```

Trigger config:

- `queue` (required): topic/queue name to consume. `topic` is accepted as a
  compatibility alias.
- `max_retries` (default `3`): failed attempts before moving to DLQ.
- `backoff_ms` (default `1000`): base retry delay. Retries use the same
  exponential curve as the builtin queue: `backoff_ms * 2^(attempts - 1)`.
- `condition_function_id` (optional): called before the handler; only explicit
  `false` skips the handler.

## Configuration

Use the `configuration` worker entry named `queue`. The default is in-memory:

```yaml
adapter:
  name: builtin
```

For restart survival, use file-backed storage:

```yaml
adapter:
  name: builtin
  config:
    store_method: file_based
    file_path: data/queue
    save_interval_ms: 5000
```

For pub/sub-only delivery against a shared Redis (no DLQ, no durability):

```yaml
adapter:
  name: redis
  config:
    redis_url: redis://localhost:6379
```

For full retry/DLQ/priority/fifo delivery against RabbitMQ:

```yaml
adapter:
  name: rabbitmq
  config:
    amqp_url: amqp://localhost:5672
    max_attempts: 3
    prefetch_count: 10
    queue_mode: standard
    priority_field: priority
```

Changing the adapter config hot-swaps the transport and restarts every
consumer — the new adapter is built first, so a bad config leaves the
previous one serving. File-backed jobs survive by construction; in-memory
jobs are lost, matching the builtin adapter's in-process durability profile.

## api reference

```json
{
  "functions": [
    {
      "description": "Browse the dead-lettered (DLQ) messages of a queue",
      "metadata": {},
      "name": "engine::queue::dlq_messages",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "limit": {
            "default": 50,
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "offset": {
            "default": 0,
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "topic": {
            "type": "string"
          }
        },
        "required": [
          "topic"
        ],
        "title": "DlqMessagesInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "DlqMessage": {
            "properties": {
              "error": {
                "type": "string"
              },
              "failed_at": {
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              },
              "id": {
                "type": "string"
              },
              "payload": true,
              "retries": {
                "format": "uint32",
                "minimum": 0,
                "type": "integer"
              },
              "size_bytes": {
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              }
            },
            "required": [
              "error",
              "failed_at",
              "id",
              "payload",
              "retries",
              "size_bytes"
            ],
            "type": "object"
          }
        },
        "items": {
          "$ref": "#/definitions/DlqMessage"
        },
        "title": "Array_of_DlqMessage",
        "type": "array"
      }
    },
    {
      "description": "List dead-letter (DLQ) topics with their message counts",
      "metadata": {},
      "name": "engine::queue::dlq_topics",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Input for `engine::queue::dlq_topics` — same story as [`ListTopicsInput`].",
        "title": "DlqTopicsInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "DlqTopicInfo": {
            "properties": {
              "broker_type": {
                "type": "string"
              },
              "message_count": {
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              },
              "topic": {
                "type": "string"
              }
            },
            "required": [
              "broker_type",
              "message_count",
              "topic"
            ],
            "type": "object"
          }
        },
        "items": {
          "$ref": "#/definitions/DlqTopicInfo"
        },
        "title": "Array_of_DlqTopicInfo",
        "type": "array"
      }
    },
    {
      "description": "Internal provider for TriggerAction::Enqueue",
      "metadata": {},
      "name": "engine::queue::enqueue",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "additionalProperties": false,
        "properties": {
          "baggage": {
            "default": null,
            "type": [
              "string",
              "null"
            ]
          },
          "data": true,
          "function_id": {
            "type": "string"
          },
          "messageReceiptId": {
            "type": "string"
          },
          "namespace": {
            "default": null,
            "description": "Namespace in which the queued target must resolve. The engine sends this for non-default enqueues and omits it for legacy/default traffic.",
            "type": [
              "string",
              "null"
            ]
          },
          "queue": {
            "type": "string"
          },
          "traceparent": {
            "default": null,
            "description": "Trace context captured by the engine at the enqueue boundary and restored when the queued function is invoked.",
            "type": [
              "string",
              "null"
            ]
          }
        },
        "required": [
          "data",
          "function_id",
          "messageReceiptId",
          "queue"
        ],
        "title": "EnqueueInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "messageReceiptId": {
            "type": "string"
          }
        },
        "required": [
          "messageReceiptId"
        ],
        "title": "EnqueueOutput",
        "type": "object"
      }
    },
    {
      "description": "List all queue topics with their broker type",
      "metadata": {},
      "name": "engine::queue::list_topics",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Input for `engine::queue::list_topics`. The function takes no parameters; any provided fields are ignored (engine parity). The empty struct keeps the registered request schema typed for the registry publish gate.",
        "title": "ListTopicsInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "TopicInfo": {
            "properties": {
              "broker_type": {
                "type": "string"
              },
              "name": {
                "type": "string"
              },
              "subscriber_count": {
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              }
            },
            "required": [
              "broker_type",
              "name",
              "subscriber_count"
            ],
            "type": "object"
          }
        },
        "items": {
          "$ref": "#/definitions/TopicInfo"
        },
        "title": "Array_of_TopicInfo",
        "type": "array"
      }
    },
    {
      "description": "Get queue depth, delivered and failed counts for a topic",
      "metadata": {},
      "name": "engine::queue::topic_stats",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "topic": {
            "type": "string"
          }
        },
        "required": [
          "topic"
        ],
        "title": "TopicStatsInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "config": true,
          "consumer_count": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "delivered": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "depth": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "dlq_depth": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "failed": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          }
        },
        "required": [
          "consumer_count",
          "delivered",
          "depth",
          "dlq_depth",
          "failed"
        ],
        "title": "TopicStatsOutput",
        "type": "object"
      }
    },
    {
      "description": "Enqueue a message onto a queue topic",
      "metadata": {},
      "name": "iii::durable::publish",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "data": true,
          "topic": {
            "description": "Topic to publish to. `queue` is accepted for the migration worker API.",
            "type": "string"
          }
        },
        "required": [
          "data",
          "topic"
        ],
        "title": "PublishInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "anyOf": [
          {
            "$ref": "#/definitions/PublishAck"
          },
          {
            "type": "null"
          }
        ],
        "definitions": {
          "PublishAck": {
            "description": "Acknowledgement type for `iii::durable::publish`. The function always returns `null` on the wire (engine parity — the builtin returns `Success(None)`); this type exists so the registered response schema is typed instead of `AnyValue`, which the registry publish gate rejects.",
            "type": "object"
          }
        },
        "title": "Nullable_PublishAck"
      }
    },
    {
      "description": "Discard (purge) a single DLQ message by ID",
      "metadata": {},
      "name": "iii::queue::discard_message",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "message_id": {
            "type": "string"
          },
          "queue": {
            "type": "string"
          }
        },
        "required": [
          "message_id",
          "queue"
        ],
        "title": "RedriveSingleInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "message_id": {
            "type": "string"
          },
          "queue": {
            "type": "string"
          },
          "redriven": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          }
        },
        "required": [
          "message_id",
          "queue",
          "redriven"
        ],
        "title": "RedriveSingleResult",
        "type": "object"
      }
    },
    {
      "description": "Redrive all DLQ messages back to the main queue",
      "metadata": {},
      "name": "iii::queue::redrive",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "queue": {
            "type": "string"
          }
        },
        "required": [
          "queue"
        ],
        "title": "RedriveInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "queue": {
            "type": "string"
          },
          "redriven": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          }
        },
        "required": [
          "queue",
          "redriven"
        ],
        "title": "RedriveResult",
        "type": "object"
      }
    },
    {
      "description": "Redrive a single DLQ message by ID back to the main queue",
      "metadata": {},
      "name": "iii::queue::redrive_message",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "message_id": {
            "type": "string"
          },
          "queue": {
            "type": "string"
          }
        },
        "required": [
          "message_id",
          "queue"
        ],
        "title": "RedriveSingleInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "message_id": {
            "type": "string"
          },
          "queue": {
            "type": "string"
          },
          "redriven": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          }
        },
        "required": [
          "message_id",
          "queue",
          "redriven"
        ],
        "title": "RedriveSingleResult",
        "type": "object"
      }
    },
    {
      "description": "Define and start a durable named function queue",
      "metadata": {},
      "name": "queue::define",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "additionalProperties": false,
        "definitions": {
          "FunctionQueueConfig": {
            "additionalProperties": false,
            "description": "Per-function-queue transport settings, passed to [`QueueAdapter::setup_function_queue`].",
            "properties": {
              "backoff_ms": {
                "default": 1000,
                "description": "Base delay in milliseconds for the exponential retry backoff.",
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              },
              "concurrency": {
                "default": 10,
                "description": "Number of messages processed concurrently.",
                "format": "uint32",
                "minimum": 0,
                "type": "integer"
              },
              "max_priority": {
                "description": "Declares the queue as a RabbitMQ priority queue with this many priority levels (`x-max-priority`). `None` means not a priority queue. RabbitMQ-only; other adapters ignore it. Added (rather than part of the original minimal port) so [`QueueAdapter::setup_function_queue`] can pass it through to the RabbitMQ adapter's topology setup.",
                "format": "uint8",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              },
              "max_retries": {
                "default": 3,
                "description": "Maximum retries after the initial delivery before a message is sent to the dead-letter queue.",
                "format": "uint32",
                "minimum": 0,
                "type": "integer"
              },
              "message_group_field": {
                "description": "Payload field used to partition FIFO messages into independently ordered groups.",
                "type": [
                  "string",
                  "null"
                ]
              },
              "poll_interval_ms": {
                "default": 100,
                "description": "Deprecated and ignored. Consumers are event-driven: an enqueue wakes an idle consumer immediately and a delayed retry fires at its due time, so there is no poll interval. Still accepted (any value) so existing definitions keep deserializing under `deny_unknown_fields`.",
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              },
              "priority_field": {
                "description": "Payload field whose non-negative integer value supplies message priority when the adapter supports priority queues.",
                "type": [
                  "string",
                  "null"
                ]
              },
              "redeliver_on_engine_restart": {
                "default": false,
                "description": "Re-deliver an invocation inside the same queue attempt when the engine restarts while it is in flight. This is only safe for consumers whose own durable checkpoints make duplicate delivery idempotent.",
                "type": "boolean"
              },
              "timeout_ms": {
                "default": 1800000,
                "description": "Maximum time in milliseconds allowed for one target-function invocation.",
                "format": "uint64",
                "minimum": 0,
                "type": "integer"
              },
              "type": {
                "default": "standard",
                "description": "Queue scheduling mode. Supported values are `standard` and `fifo`.",
                "type": "string"
              }
            },
            "type": "object"
          }
        },
        "properties": {
          "config": {
            "allOf": [
              {
                "$ref": "#/definitions/FunctionQueueConfig"
              }
            ],
            "default": {
              "backoff_ms": 1000,
              "concurrency": 10,
              "max_retries": 3,
              "poll_interval_ms": 100,
              "redeliver_on_engine_restart": false,
              "timeout_ms": 1800000,
              "type": "standard"
            }
          },
          "queue": {
            "type": "string"
          }
        },
        "required": [
          "queue"
        ],
        "title": "DefineQueueInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "changed": {
            "type": "boolean"
          },
          "queue": {
            "type": "string"
          }
        },
        "required": [
          "changed",
          "queue"
        ],
        "title": "DefineQueueOutput",
        "type": "object"
      }
    },
    {
      "description": "Internal: reload queue configuration from the authoritative store.",
      "metadata": {
        "internal": true
      },
      "name": "queue::on-config-change",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "ConfigChangeRequest",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "ok": {
            "type": "boolean"
          }
        },
        "required": [
          "ok"
        ],
        "title": "ConfigChangeAck",
        "type": "object"
      }
    },
    {
      "description": "Internal: wake queue deliveries held on an engine change (a target function registering, an engine restart) when functions or workers come and go.",
      "metadata": {
        "internal": true
      },
      "name": "queue::on-engine-change",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Payload of the engine's registry-change triggers. Only the advisory event tag is read; the larger fields (`functions`, `worker_id`) are ignored because a change is only a cue to re-check.",
        "properties": {
          "event": {
            "default": null,
            "type": [
              "string",
              "null"
            ]
          }
        },
        "title": "EngineChangeEvent",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "ok": {
            "type": "boolean"
          }
        },
        "required": [
          "ok"
        ],
        "title": "EngineChangeAck",
        "type": "object"
      }
    }
  ],
  "triggers": [
    {
      "description": "Durable queue subscriber",
      "invocation_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "SubscriberQueueConfig": {
            "properties": {
              "backoff_delay_ms": {
                "format": "uint64",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              },
              "backoff_type": {
                "type": [
                  "string",
                  "null"
                ]
              },
              "concurrency": {
                "format": "uint32",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              },
              "delay_seconds": {
                "format": "uint64",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              },
              "max_priority": {
                "description": "Declares this subscriber's queue as a RabbitMQ priority queue with this many levels (`x-max-priority`, 1–255). RabbitMQ-only; the priority value of each message comes from the adapter-level `priority_field`.",
                "format": "uint8",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              },
              "max_retries": {
                "format": "uint32",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              },
              "type": {
                "type": [
                  "string",
                  "null"
                ]
              },
              "visibility_timeout": {
                "format": "uint64",
                "minimum": 0,
                "type": [
                  "integer",
                  "null"
                ]
              }
            },
            "type": "object"
          }
        },
        "properties": {
          "backoff_ms": {
            "format": "uint64",
            "minimum": 0,
            "type": [
              "integer",
              "null"
            ]
          },
          "condition_function_id": {
            "type": [
              "string",
              "null"
            ]
          },
          "max_retries": {
            "format": "uint32",
            "minimum": 0,
            "type": [
              "integer",
              "null"
            ]
          },
          "queue": {
            "description": "Queue/topic name to consume.",
            "type": "string"
          },
          "queue_config": {
            "anyOf": [
              {
                "$ref": "#/definitions/SubscriberQueueConfig"
              },
              {
                "type": "null"
              }
            ]
          }
        },
        "required": [
          "queue"
        ],
        "title": "SubscriberSpec",
        "type": "object"
      },
      "metadata": {},
      "name": "durable:subscriber",
      "return_schema": {}
    }
  ]
}
```