Queues

The queue worker decouples producers from consumers: a function publishes a message to a named topic and returns right away, and any function subscribed to that topic processes the message in the background, with retries and a dead-letter queue (DLQ) for messages that keep failing.

Before adding the worker

compose::add is served by a running Compose daemon. Keep the engine and a daemon for this project running in separate terminals before using any of the commands below. If this project does not have a Compose file yet, create worker-compose.yaml containing containers: {} first.

# terminal 1
iii --config config.yaml

# terminal 2, from the directory that contains worker-compose.yaml
iii compose --namespace dev --engine ws://127.0.0.1:49134

iii compose is an intentional verbless daemon invocation, documented in the CLI reference. The -n used below is the short form of --namespace for iii call. Run the remaining commands from a third terminal in that same project directory:

iii call -n dev compose::add worker=queue

This page covers common queue patterns. For the complete configuration and function API, see the queue worker docs.

Named Queues

When you need to control function execution for time consuming operations, or guarantee a certain number of retries, call the function with the enqueue action. The queue field of the action names the queue to use, so configure that queue in the queue worker configuration first.

Enqueue a function

An enqueue call is a function call with the enqueue action. The queue runs the function later, and the caller gets a receipt before the function starts:

Node / TypeScript
Python
Rust
Go

import { CallAction, registerWorker, type EnqueueResult } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, { workerName: "email-worker", namespace: "email" });

const { messageReceiptId } = await worker.call<unknown, EnqueueResult>({
  function_id: "email::send",
  payload: { to: "a@b.com", subject: "hi" },
  action: CallAction.Enqueue({ queue: "email-jobs" }), // "queue" names the queue
});
// messageReceiptId identifies the enqueued job

import os

from iii import CallAction, InitOptions, register_worker

worker = register_worker(
    os.environ["III_URL"],
    InitOptions(worker_name="email-worker", namespace="email"),
)

receipt = worker.call(
    {
        "function_id": "email::send",
        "payload": {"to": "a@b.com", "subject": "hi"},
        "action": CallAction.Enqueue(queue="email-jobs"),  # "queue" names the queue
    }
)
# receipt["messageReceiptId"] identifies the enqueued job

use iii_sdk::{register_worker, CallAction, CallRequest, InitOptions};
use serde_json::json;

let worker = register_worker(
    &std::env::var("III_URL").expect("III_URL must be set"),
    InitOptions {
        namespace: Some("email".into()),
        ..Default::default()
    },
);

let receipt = worker
    .call(CallRequest {
        function_id: "email::send".to_string(),
        payload: json!({ "to": "a@b.com", "subject": "hi" }),
        action: Some(CallAction::Enqueue {
            queue: "email-jobs".to_string(),
        }),
        timeout_ms: None,
    })
    .await?;
// receipt["messageReceiptId"] identifies the enqueued job

import (
	"context"
	"encoding/json"
	"log"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
	"github.com/iii-hq/iii/sdk/packages/go/iii/helpers/queue"
)

client := iii.RegisterWorker(iii.EngineURLFromEnv())
defer client.Close()

res, err := client.Call(context.Background(), iii.CallRequest{
    FunctionID: "email::send",
    Payload:    json.RawMessage(`{"to":"a@b.com","subject":"hi"}`),
    Action:     iii.EnqueueAction("email-jobs"), // "email-jobs" names the queue
})
if err != nil {
    log.Fatal(err)
}

var receipt queue.EnqueueResult
if err := json.Unmarshal(res, &receipt); err != nil {
    log.Fatal(err)
}
// receipt.MessageReceiptID identifies the enqueued job

Pub/Sub Queues

Queues can also be used in a publish/subscribe form when multiple listeners need to subscribe to the same data and it’s important that the messages be durable (ie. will succeed).

Consume messages

An observer of queue::on-message with a topic in its args runs the target function for each message published to that topic, with the message data as the input. A normal return acknowledges the message. An error retries it, and the last failed delivery moves the message to the dead-letter queue. In a worker, register the consumer function and observe queue::on-message with the topic. If you do not have a worker yet, follow Create a new worker, then edit its source:

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, { workerName: "email-worker", namespace: "email" });

// receives the `data` from each published message
worker.registerFunction("email::send", async (msg: { to: string; subject: string }) => {
  // do the work here; throw to nack and let the message retry
  return { sent: true };
});

worker.observe({ function_id: "queue::on-message", args: { topic: "emails" } }, "email::send");

import os

from iii import InitOptions, register_worker

worker = register_worker(
    os.environ["III_URL"],
    InitOptions(worker_name="email-worker", namespace="email"),
)

# receives the `data` from each published message
def send(msg: dict) -> dict:
    # do the work here; raise to nack and let the message retry
    return {"sent": True}

worker.register_function("email::send", send)

worker.observe({"function_id": "queue::on-message", "args": {"topic": "emails"}}, "email::send")

use iii_sdk::{register_worker, InitOptions, RegisterFunction, Source};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::json;

#[derive(Deserialize, JsonSchema)]
struct Email {
    to: String,
    subject: String,
}

let worker = register_worker(
    &std::env::var("III_URL").expect("III_URL must be set"),
    InitOptions {
        namespace: Some("email".into()),
        ..Default::default()
    },
);

// receives the `data` from each published message
worker.register_function(
    "email::send",
    RegisterFunction::new(|_msg: Email| Ok::<_, iii_sdk::Error>(json!({ "sent": true }))),
);

worker.observe(
    Source::function("queue::on-message").args(json!({ "topic": "emails" })),
    "email::send",
);

import (
	"context"
	"encoding/json"
	"log"
	"os"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

client := iii.RegisterWorker(os.Getenv("III_URL"),
    iii.WithName("email-worker"),
    iii.WithNamespace("email"),
)
defer client.Close()

// receives the `data` from each published message
client.RegisterFunction("email::send", func(ctx context.Context, data json.RawMessage) (any, error) {
    var msg struct {
        To      string `json:"to"`
        Subject string `json:"subject"`
    }
    if err := json.Unmarshal(data, &msg); err != nil {
        return nil, err
    }
    // do the work here; return an error to nack and let the message retry
    return map[string]bool{"sent": true}, nil
})

if _, err := client.Observe(iii.Source{
    FunctionID: "queue::on-message",
    Args:       iii.Args{"topic": "emails"},
}, "email::send"); err != nil {
    log.Fatal(err)
}

Add the worker to start it:

iii call -n dev compose::add worker=./email-worker

Observer filters select messages by their data as input. A message that the filters reject is acknowledged without running the target.

Publish a message

With the consumer running, publish to its topic:

# publish a message to the "emails" topic
iii call iii::durable::publish --json '{"topic":"emails","data":{"to":"a@b.com","subject":"hi"}}'

Open the console and go to the Traces tab to watch the message flow from the publish through to email::send running.

Subscriber args

Every observer of queue::on-message carries its subscription in args. The args fall into three classes.

Consumption

topic names the topic to consume. It is required.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, { workerName: "email-worker", namespace: "email" });

worker.observe({ function_id: "queue::on-message", args: { topic: "emails" } }, "email::send");

import os

from iii import InitOptions, register_worker

worker = register_worker(
    os.environ["III_URL"],
    InitOptions(worker_name="email-worker", namespace="email"),
)

worker.observe({"function_id": "queue::on-message", "args": {"topic": "emails"}}, "email::send")

use iii_sdk::{register_worker, InitOptions, Source};
use serde_json::json;

let worker = register_worker(
    &std::env::var("III_URL").expect("III_URL must be set"),
    InitOptions {
        namespace: Some("email".into()),
        ..Default::default()
    },
);

worker.observe(
    Source::function("queue::on-message").args(json!({ "topic": "emails" })),
    "email::send",
);

import (
	"log"
	"os"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

client := iii.RegisterWorker(os.Getenv("III_URL"),
    iii.WithName("email-worker"),
    iii.WithNamespace("email"),
)
defer client.Close()

if _, err := client.Observe(iii.Source{
    FunctionID: "queue::on-message",
    Args:       iii.Args{"topic": "emails"},
}, "email::send"); err != nil {
    log.Fatal(err)
}

Retries

max_retries counts the deliveries of one message, the first one included, before the message moves to the dead-letter queue. The default is 3. backoff_ms sets the first retry delay, and each later retry doubles it, so a message that always fails reaches the dead-letter queue about 3 seconds after it is published.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, { workerName: "email-worker", namespace: "email" });

worker.observe(
  {
    function_id: "queue::on-message",
    args: { topic: "emails", max_retries: 5, backoff_ms: 500 },
  },
  "email::send",
);

import os

from iii import InitOptions, register_worker

worker = register_worker(
    os.environ["III_URL"],
    InitOptions(worker_name="email-worker", namespace="email"),
)

worker.observe(
    {
        "function_id": "queue::on-message",
        "args": {"topic": "emails", "max_retries": 5, "backoff_ms": 500},
    },
    "email::send",
)

use iii_sdk::{register_worker, InitOptions, Source};
use serde_json::json;

let worker = register_worker(
    &std::env::var("III_URL").expect("III_URL must be set"),
    InitOptions {
        namespace: Some("email".into()),
        ..Default::default()
    },
);

worker.observe(
    Source::function("queue::on-message")
        .args(json!({ "topic": "emails", "max_retries": 5, "backoff_ms": 500 })),
    "email::send",
);

import (
	"log"
	"os"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

client := iii.RegisterWorker(os.Getenv("III_URL"),
    iii.WithName("email-worker"),
    iii.WithNamespace("email"),
)
defer client.Close()

if _, err := client.Observe(iii.Source{
    FunctionID: "queue::on-message",
    Args:       iii.Args{"topic": "emails", "max_retries": 5, "backoff_ms": 500},
}, "email::send"); err != nil {
    log.Fatal(err)
}

The engine bounds each delivery to a target by observers.activation_timeout_ms (30 seconds by default). A target that runs longer fails the delivery with TIMEOUT, which is retried like any other failure. For longer jobs, raise it in the engine config.

Delivery

queue_config sets the delivery settings of this subscriber, for example { type: "fifo" } to process messages one at a time. group names the consumer group: subscribers that name the same group share one consumer, which handles each message once.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, { workerName: "email-worker", namespace: "email" });

worker.observe(
  {
    function_id: "queue::on-message",
    args: { topic: "emails", queue_config: { type: "fifo" }, group: "email-senders" },
  },
  "email::send",
);

import os

from iii import InitOptions, register_worker

worker = register_worker(
    os.environ["III_URL"],
    InitOptions(worker_name="email-worker", namespace="email"),
)

worker.observe(
    {
        "function_id": "queue::on-message",
        "args": {
            "topic": "emails",
            "queue_config": {"type": "fifo"},
            "group": "email-senders",
        },
    },
    "email::send",
)

use iii_sdk::{register_worker, InitOptions, Source};
use serde_json::json;

let worker = register_worker(
    &std::env::var("III_URL").expect("III_URL must be set"),
    InitOptions {
        namespace: Some("email".into()),
        ..Default::default()
    },
);

worker.observe(
    Source::function("queue::on-message").args(json!({
        "topic": "emails",
        "queue_config": { "type": "fifo" },
        "group": "email-senders",
    })),
    "email::send",
);

import (
	"log"
	"os"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

client := iii.RegisterWorker(os.Getenv("III_URL"),
    iii.WithName("email-worker"),
    iii.WithNamespace("email"),
)
defer client.Close()

if _, err := client.Observe(iii.Source{
    FunctionID: "queue::on-message",
    Args: iii.Args{
        "topic":        "emails",
        "queue_config": iii.Args{"type": "fifo"},
        "group":        "email-senders",
    },
}, "email::send"); err != nil {
    log.Fatal(err)
}

A subscriber without a group has its own durable queue, so two observers of one topic each receive every published message. To spread one stream of messages across several consumers, give their observers the same group.

RabbitMQ queue names changed in 0.23.x and are not migrated automatically. See Upgrading from 0.22.x.

Inspecting Queue Topics

engine::queue::list_topics lists every topic. engine::queue::topic_stats reports the counters of one topic: depth is the messages waiting for the consumer and dlq_depth is the dead-lettered messages. A topic with no waiting messages reports depth: 0. For a named queue, consumer_count reports its active delivery slots. A topic that no observer subscribes to does not register, so there is nothing to inspect. List every topic:

iii call engine::queue::list_topics

[{ "name": "emails", "broker_type": "builtin", "subscriber_count": 1 }]

Get the counters of the emails topic:

iii call engine::queue::topic_stats topic=emails

{ "depth": 0, "consumer_count": 1, "dlq_depth": 0, "config": null }

The list holds two kinds of queue. A pub/sub topic appears once an observer subscribes to it and shows broker_type: "builtin". A configured named queue shows broker_type: "function_queue".

Inspecting Dead Letter Queue Messages

A message reaches the dead-letter queue only once its target exhausts its retries, so the DLQ functions return empty until something fails.

Forcing a message into the dead-letter queue

To see the DLQ populated, make email::send fail by changing the handler to throw. Each message then fails all of its deliveries and dead-letters.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, { workerName: "email-worker", namespace: "email" });

worker.registerFunction("email::send", async () => {
  throw new Error("forced failure");
});

worker.observe({ function_id: "queue::on-message", args: { topic: "emails" } }, "email::send");

import os

from iii import InitOptions, register_worker

worker = register_worker(
    os.environ["III_URL"],
    InitOptions(worker_name="email-worker", namespace="email"),
)

# the message is the input; the error below nacks every delivery
def send(_msg: dict) -> dict:
    raise Exception("forced failure")

worker.register_function("email::send", send)

worker.observe({"function_id": "queue::on-message", "args": {"topic": "emails"}}, "email::send")

use iii_sdk::{register_worker, InitOptions, RegisterFunction, Source};
use serde_json::json;

let worker = register_worker(
    &std::env::var("III_URL").expect("III_URL must be set"),
    InitOptions {
        namespace: Some("email".into()),
        ..Default::default()
    },
);

worker.register_function(
    "email::send",
    RegisterFunction::new(|_msg: serde_json::Value| {
        Err::<serde_json::Value, _>(iii_sdk::Error::Handler("forced failure".into()))
    }),
);

worker.observe(
    Source::function("queue::on-message").args(json!({ "topic": "emails" })),
    "email::send",
);

import (
	"context"
	"encoding/json"
	"errors"
	"log"
	"os"

	iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

client := iii.RegisterWorker(os.Getenv("III_URL"),
    iii.WithName("email-worker"),
    iii.WithNamespace("email"),
)
defer client.Close()

client.RegisterFunction("email::send", func(ctx context.Context, data json.RawMessage) (any, error) {
    return nil, errors.New("forced failure")
})

if _, err := client.Observe(iii.Source{
    FunctionID: "queue::on-message",
    Args:       iii.Args{"topic": "emails"},
}, "email::send"); err != nil {
    log.Fatal(err)
}

Now publish a message; after the retries run out it lands in the DLQ (exact ids, timestamps, and sizes vary per run):

iii call iii::durable::publish --json '{"topic":"emails","data":{"to":"a@b.com","subject":"hi"}}'

Listing topics with dead-lettered messages

iii call engine::queue::dlq_topics

[{ "topic": "emails", "broker_type": "builtin", "message_count": 1 }]

Browsing dead-lettered messages

iii call engine::queue::dlq_messages topic=emails

[
  {
    "id": "0b9c…",
    "payload": { "to": "a@b.com", "subject": "hi" },
    "error": "ErrorBody { code: \"invocation_failed\", message: \"forced failure\"...",
    "failed_at": 1718900000,
    "retries": 3,
    "size_bytes": 64
  }
]

Redriving dead-lettered messages

Restore the handler, then move the topic’s dead-lettered messages back to the main queue for reprocessing:

iii call iii::queue::redrive topic=emails

{ "queue": "emails", "redriven": 1 }

The fixed function now processes them, so the DLQ is empty again:

iii call engine::queue::dlq_messages topic=emails

Redriving or discarding a single message

To handle one message instead of the whole topic, pass the id from engine::queue::dlq_messages. Redrive one message back to the main queue:

iii call iii::queue::redrive_message topic=emails message_id=0b9c…

{ "queue": "emails", "message_id": "0b9c…", "redriven": 1 }

Or discard it, deleting it from the DLQ permanently:

iii call iii::queue::discard_message topic=emails message_id=0b9c…

{ "queue": "emails", "message_id": "0b9c…", "redriven": 1 }

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Before adding the worker

Named Queues

Enqueue a function

Pub/Sub Queues

Consume messages

Publish a message

Subscriber args

Consumption

Retries

Delivery

Inspecting Queue Topics

Inspecting Dead Letter Queue Messages

Forcing a message into the dead-letter queue

Listing topics with dead-lettered messages

Browsing dead-lettered messages

Redriving dead-lettered messages

Redriving or discarding a single message

​Before adding the worker

​Named Queues

​Enqueue a function

​Pub/Sub Queues

​Consume messages

​Publish a message

​Subscriber args

​Consumption

​Retries

​Delivery

​Inspecting Queue Topics

​Inspecting Dead Letter Queue Messages

​Forcing a message into the dead-letter queue

​Listing topics with dead-lettered messages

​Browsing dead-lettered messages

​Redriving dead-lettered messages

​Redriving or discarding a single message

Before adding the worker

Named Queues

Enqueue a function

Pub/Sub Queues

Consume messages

Publish a message

Subscriber args

Consumption

Retries

Delivery

Inspecting Queue Topics

Inspecting Dead Letter Queue Messages

Forcing a message into the dead-letter queue

Listing topics with dead-lettered messages

Browsing dead-lettered messages

Redriving dead-lettered messages

Redriving or discarding a single message