$worker

queue

v0.1.2

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

macOS: arm64 · x64
Linux: arm64 · armv7 · x64
Windows: arm64 · x64 · x86

agent-ready brief for v0.1.2

install + config + dependencies + readme + api reference, all in one place. fetch as agent-context.md for an llm to consume.

full markdown

open as agent-context.md

the same content rendered as discrete blocks below is exposed as a single markdown document at /workers/queue.md?version=0.1.2. paste it into an llm prompt or pipe it through curl from a worker.

install

$iii worker add queue@0.1.2

configuration

iii-config.yaml

- adapter:
    name: builtin

dependencies

dependencies1

nameversion

configuration^0.19.0

readme

README.md

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

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

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.

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.

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)

agent-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": {}
    }
  ]
}

queue

full markdown

install

configuration

dependencies

readme

queue

Install

Trigger Type

Functions

Configuration

Adapters

builtin (default)

redis

rabbitmq

memory (dev/test only)

Trigger queue_config

Requires Removing The Built-In iii-queue Worker

Parity Vs Builtin

Known Gaps / Parity Notes

api reference (json)

`builtin` (default)

`redis`

`rabbitmq`

`memory` (dev/test only)

Trigger `queue_config`

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