# queue

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

| field | value |
|-------|-------|
| version | 0.1.3 |
| type | binary |
| repo | https://github.com/iii-hq/workers |
| supported_targets | x86_64-apple-darwin, aarch64-apple-darwin, i686-pc-windows-msvc, 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 worker add queue@0.1.3
```

## configuration

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

## dependencies

- `configuration` @ `^0.19.0`

## readme

# queue

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

## Install

```bash
iii worker add queue
```

`iii worker add` fetches the binary, writes a config block into
`~/.iii/config.yaml`, and the engine starts the worker on the next
`iii start`.

## 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 |
|---|---|---|
| `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 |

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

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

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

| Field | Default | Description |
|---|---|---|
| `adapter.config.store_method` | `in_memory` | `in_memory` or `file_based`. |
| `adapter.config.file_path` | `queue_store_data` | 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.

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

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

## Requires Removing The Built-In `iii-queue` Worker

The built-in `iii-queue` worker also owns `durable:subscriber`. Two owners of
the same trigger type on one engine collide, so this worker requires
`iii-queue` to be absent from the engine config.

On boot, this worker queries `engine::workers::list` and refuses to start if
`iii-queue` is active.

## Parity Vs Builtin

| Behavior | Builtin | This worker |
|---|---|---|
| Trigger type | `durable:subscriber` | same |
| Function ids | 8 public ids listed above | same, verbatim |
| 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`) | N/A — engine `QueueEnqueuer` cut (MOT-3829) |
| Enqueue failure when worker offline | n/a, in-process | invocation fails explicitly once the engine remote-enqueue cut lands |
| Latency benchmark | in-process baseline required | final budget is a pre-deprecation gate (tracked in the migration master plan) |

The full engine `TriggerAction::Enqueue` path depends on the separate
`QueueEnqueuer` engine cut tracked in the migration master plan.

## Known Gaps / Parity Notes

- **`bridge` adapter — not ported.** It was engine-internal plumbing for
  `TriggerAction::Enqueue`; that whole path is superseded by the engine's
  `QueueEnqueuer` cut (MOT-3829), which is out of scope for this worker.
- **No grouped-fifo.** The engine's `GroupedFifoWorker` partitions `fifo`
  delivery by `job.group_id` when `concurrency > 1`, running each group
  serially but different groups concurrently. This worker's `fifo` mode is
  strictly serial regardless of `concurrency` — there is no group
  partitioning.
- **`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 poller
  in-place (the engine's `FifoWorker` blocks); a message enqueued after
  the failure can be delivered before the retried one catches up.

## api reference

```json
{
  "functions": [
    {
      "description": "Browse DLQ messages",
      "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 DLQ topics with 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": "List all queue topics",
      "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 stats for a queue 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"
          },
          "depth": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          },
          "dlq_depth": {
            "format": "uint64",
            "minimum": 0,
            "type": "integer"
          }
        },
        "required": [
          "consumer_count",
          "depth",
          "dlq_depth"
        ],
        "title": "TopicStatsOutput",
        "type": "object"
      }
    },
    {
      "description": "Enqueue a message",
      "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": "Internal: reload queue configuration from the authoritative store.",
      "metadata": {},
      "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"
      }
    }
  ],
  "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": {}
    }
  ]
}
```