Use Queues

Goal

Use queues to run work asynchronously with built-in retries, dead-letter support, and concurrency control. iii offers two queue modes — topic-based (pub/sub fan-out) and named queues (direct function targeting) — so you can pick the delivery model that fits your use case.

What Are Queues

A queue sits between the code that produces work and the code that processes it. Instead of calling a function and waiting for it to finish, you hand the work to a queue. The queue stores the message, delivers it to a consumer, retries on failure, and routes permanently failed messages to a dead-letter queue (DLQ) for later inspection. This separation solves three problems:

Speed — the producer responds immediately instead of blocking on slow downstream work.
Reliability — transient failures (network blips, service restarts) are retried automatically instead of being lost.
Load control — concurrency limits prevent consumers from overwhelming downstream systems.

When to Use Queues

Scenario	Use a queue?	Why
HTTP handler must respond fast, but downstream work is slow	Yes	Enqueue the work and return `202 Accepted` immediately
Multiple functions must react to the same event	Yes	Topic-based queues fan out to every subscriber
Work must survive process restarts	Yes	Queues persist messages and retry on failure
External API has rate limits	Yes	Concurrency control throttles parallel requests
Transactions for the same entity must be ordered	Yes	FIFO queues guarantee per-group ordering
You need the function’s return value right now	No	Use a synchronous trigger instead
The work is non-critical and losing it is acceptable	Maybe	TriggerAction.Void() is simpler if you don’t need retries

Two Queue Modes

iii supports two ways to use queues. Both share the same adapter, retry engine, and DLQ infrastructure — they differ in how producers address consumers.

	Topic-based	Named queues
Producer	`trigger({ function_id: 'enqueue', payload: { topic, data } })`	`trigger({ function_id, payload, action: TriggerAction.Enqueue({ queue }) })`
Consumer	Registers `registerTrigger({ type: 'queue', config: { topic } })`	No registration — function is the target
Delivery	Fan-out: each subscribed function gets every message; replicas compete	Single target function per enqueue call
Config	Optional `queue_config` on trigger	`queue_configs` in `iii-config.yaml`
Best for	Durable pub/sub with retries and fan-out	Direct function invocation with retries, FIFO, DLQ

Named queues use the Enqueue trigger action. If you are new to trigger actions, read Trigger Actions first.

Topic-Based Queues

Topic-based queues work like durable pub/sub: you publish a message to a topic, and every function subscribed to that topic receives a copy. If a function has multiple replicas, they compete on a shared per-function queue — only one replica processes each message.

Subscribe one or more functions to the same topic. Each function gets its own internal queue.

Node / TypeScript
Python
Rust

import { registerWorker } from 'iii-sdk'

const iii = registerWorker(process.env.III_URL ?? 'ws://localhost:49134')

iii.registerFunction(
  { id: 'notify::email' },
  async (data) => {
    await sendEmail(data.userId, `Order ${data.orderId} created`)
    return {}
  },
)

iii.registerFunction(
  { id: 'audit::log' },
  async (data) => {
    await writeAuditLog('order.created', data)
    return {}
  },
)

iii.registerTrigger({
  type: 'queue',
  function_id: 'notify::email',
  config: { topic: 'order.created' },
})

iii.registerTrigger({
  type: 'queue',
  function_id: 'audit::log',
  config: { topic: 'order.created' },
})

from iii import register_worker

iii = register_worker("ws://localhost:49134")


def send_email_notification(data):
    send_email(data["userId"], f"Order {data['orderId']} created")
    return {}


def write_audit(data):
    write_audit_log("order.created", data)
    return {}


iii.register_function({"id": "notify::email"}, send_email_notification)
iii.register_function({"id": "audit::log"}, write_audit)

iii.register_trigger({
    "type": "queue",
    "function_id": "notify::email",
    "config": {"topic": "order.created"},
})

iii.register_trigger({
    "type": "queue",
    "function_id": "audit::log",
    "config": {"topic": "order.created"},
})

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

let iii = register_worker("ws://localhost:49134", InitOptions::default());

iii.register_function(RegisterFunction::new_async(
    "notify::email",
    |data: Value| async move {
        send_email(data["userId"].as_str().unwrap_or(""), &format!("Order {} created", data["orderId"])).await?;
        Ok(json!({}))
    },
));

iii.register_function(RegisterFunction::new_async(
    "audit::log",
    |data: Value| async move {
        write_audit_log("order.created", &data).await?;
        Ok(json!({}))
    },
));

iii.register_trigger(RegisterTriggerInput {
    trigger_type: "queue".into(),
    function_id: "notify::email".into(),
    config: json!({ "topic": "order.created" }),
    metadata: None,
})?;

iii.register_trigger(RegisterTriggerInput {
    trigger_type: "queue".into(),
    function_id: "audit::log".into(),
    config: json!({ "topic": "order.created" }),
    metadata: None,
})?;

Both notify::email and audit::log are now subscribed to order.created. Every message published to that topic reaches both functions.

Publish events to the topic

From any function, publish a message using the builtin enqueue function. The engine fans it out to every subscribed function.

Node / TypeScript
Python
Rust

import { TriggerAction } from 'iii-sdk'

await iii.trigger({
  function_id: 'enqueue',
  payload: {
    topic: 'order.created',
    data: { orderId: 'ord_789', userId: 'usr_42', total: 149.99 },
  },
  action: TriggerAction.Void(),
})

from iii import TriggerAction

iii.trigger({
    "function_id": "enqueue",
    "payload": {
        "topic": "order.created",
        "data": {"orderId": "ord_789", "userId": "usr_42", "total": 149.99},
    },
    "action": TriggerAction.Void(),
})

use iii_sdk::{TriggerAction, TriggerRequest};
use serde_json::json;

iii.trigger(TriggerRequest {
    function_id: "enqueue".into(),
    payload: json!({
        "topic": "order.created",
        "data": { "orderId": "ord_789", "userId": "usr_42", "total": 149.99 },
    }),
    action: Some(TriggerAction::Void),
    timeout_ms: None,
}).await?;

The producer does not need to know which functions are subscribed — it only knows the topic name.

Understand fan-out delivery

Topic-based queues use fan-out per function:

Each distinct function subscribed to a topic receives a copy of every message.
If a function has multiple replicas running, they compete on a shared per-function queue — only one replica processes each message.

This gives you pub/sub-style event distribution with the durability and retry guarantees of a queue.

Filter messages with conditions (optional)

Attach a condition function to a queue trigger to filter which messages reach the handler. The condition receives the message data and returns true or false. If false, the handler is not called — no error is surfaced.

Node / TypeScript
Python
Rust

iii.registerFunction(
  { id: 'conditions::is_high_value' },
  async (data) => data.total > 1000,
)

iii.registerTrigger({
  type: 'queue',
  function_id: 'notify::vip-team',
  config: {
    topic: 'order.created',
    condition_function_id: 'conditions::is_high_value',
  },
})

def is_high_value(data):
    return data.get("total", 0) > 1000


iii.register_function({"id": "conditions::is_high_value"}, is_high_value)

iii.register_trigger({
    "type": "queue",
    "function_id": "notify::vip-team",
    "config": {
        "topic": "order.created",
        "condition_function_id": "conditions::is_high_value",
    },
})

iii.register_function(RegisterFunction::new_async(
    "conditions::is_high_value",
    |data: Value| async move {
        Ok(json!(data["total"].as_f64().unwrap_or(0.0) > 1000.0))
    },
));

iii.register_trigger(RegisterTriggerInput {
    trigger_type: "queue".into(),
    function_id: "notify::vip-team".into(),
    config: json!({
        "topic": "order.created",
        "condition_function_id": "conditions::is_high_value",
    }),
    metadata: None,
})?;

See Conditions for the full pattern including HTTP and state trigger conditions.

Named Queues

Named queues target a specific function directly. You define queue settings in iii-config.yaml and reference the queue name when enqueuing work.

Define named queues in config

Declare one or more named queues under queue_configs. Each queue has independent retry, concurrency, and ordering settings.

iii-config.yaml

modules:
  - class: modules::queue::QueueModule
    config:
      queue_configs:
        default:
          max_retries: 5
          concurrency: 10
          type: standard
        payment:
          max_retries: 10
          concurrency: 2
          type: fifo
          message_group_field: transaction_id
        email:
          max_retries: 8
          concurrency: 5
          type: standard
          backoff_ms: 2000
      adapter:
        class: modules::queue::BuiltinQueueAdapter
        config:
          store_method: file_based
          file_path: ./data/queue_store

See the Queue module reference for every field, type, and default value.

Enqueue work via trigger action

From any function, enqueue a job by calling trigger() with TriggerAction.Enqueue and the target queue name. The caller receives an acknowledgement (messageReceiptId) once the engine accepts the job — it does not wait for processing.

Node / TypeScript
Python
Rust

import { registerWorker, TriggerAction } from 'iii-sdk'

const iii = registerWorker(process.env.III_URL ?? 'ws://localhost:49134')

const receipt = await iii.trigger({
  function_id: 'orders::process-payment',
  payload: { orderId: 'ord_789', amount: 149.99, currency: 'USD' },
  action: TriggerAction.Enqueue({ queue: 'payment' }),
})

console.log(receipt.messageReceiptId)

from iii import register_worker, TriggerAction

iii = register_worker("ws://localhost:49134")

receipt = iii.trigger({
    "function_id": "orders::process-payment",
    "payload": {"orderId": "ord_789", "amount": 149.99, "currency": "USD"},
    "action": TriggerAction.Enqueue(queue="payment"),
})

print(receipt["messageReceiptId"])

use iii_sdk::{register_worker, InitOptions, TriggerAction, TriggerRequest};
use serde_json::json;

let iii = register_worker("ws://localhost:49134", InitOptions::default());

let receipt = iii.trigger(TriggerRequest {
    function_id: "orders::process-payment".to_string(),
    payload: json!({
        "orderId": "ord_789",
        "amount": 149.99,
        "currency": "USD",
    }),
    action: Some(TriggerAction::Enqueue { queue: "payment".to_string() }),
    timeout_ms: None,
}).await?;

println!("{}", receipt["messageReceiptId"]);

The target function receives the payload as its input — it does not need to know it was invoked via a queue.

Handle the enqueue result

The enqueue call can fail synchronously if the queue name is unknown or FIFO validation fails. Always handle the result.

Node / TypeScript
Python
Rust

try {
  const receipt = await iii.trigger({
    function_id: 'orders::process-payment',
    payload: { orderId: 'ord_789', amount: 149.99 },
    action: TriggerAction.Enqueue({ queue: 'payment' }),
  })
  console.log('Enqueued:', receipt.messageReceiptId)
} catch (err) {
  if (err.enqueue_error) {
    console.error('Queue rejected job:', err.enqueue_error)
  }
}

try:
    receipt = iii.trigger({
        "function_id": "orders::process-payment",
        "payload": {"orderId": "ord_789", "amount": 149.99},
        "action": TriggerAction.Enqueue(queue="payment"),
    })
    print("Enqueued:", receipt["messageReceiptId"])
except Exception as e:
    print("Queue rejected job:", e)

match iii.trigger(TriggerRequest {
    function_id: "orders::process-payment".to_string(),
    payload: json!({ "orderId": "ord_789", "amount": 149.99 }),
    action: Some(TriggerAction::Enqueue { queue: "payment".to_string() }),
    timeout_ms: None,
}).await {
    Ok(receipt) => println!("Enqueued: {}", receipt["messageReceiptId"]),
    Err(e) => eprintln!("Queue rejected job: {}", e),
}

Common rejection reasons:

The queue name does not exist in queue_configs
A FIFO queue’s message_group_field is missing or null in the payload

Use FIFO queues for ordered processing

When processing order matters — for example, financial transactions for the same account — set type: fifo and specify message_group_field. Jobs sharing the same group value are processed strictly in order.

iii-config.yaml (excerpt)

queue_configs:
  payment:
    max_retries: 10
    concurrency: 2
    type: fifo
    message_group_field: transaction_id

The payload must contain the field named by message_group_field, and its value must be non-null.

Node / TypeScript
Python
Rust

await iii.trigger({
  function_id: 'payments::process',
  payload: { transaction_id: 'txn-abc-123', amount: 49.99, currency: 'USD' },
  action: TriggerAction.Enqueue({ queue: 'payment' }),
})

iii.trigger({
    "function_id": "payments::process",
    "payload": {
        "transaction_id": "txn-abc-123",
        "amount": 49.99,
        "currency": "USD",
    },
    "action": TriggerAction.Enqueue(queue="payment"),
})

iii.trigger(TriggerRequest {
    function_id: "payments::process".to_string(),
    payload: json!({
        "transaction_id": "txn-abc-123",
        "amount": 49.99,
        "currency": "USD",
    }),
    action: Some(TriggerAction::Enqueue { queue: "payment".to_string() }),
    timeout_ms: None,
}).await?;

Configure retries and backoff

Every named queue retries failed jobs automatically. Backoff is exponential:

delay = backoff_ms × 2^(attempt - 1)

Attempt	`backoff_ms: 1000`	`backoff_ms: 2000`
1	1 000 ms	2 000 ms
2	2 000 ms	4 000 ms
3	4 000 ms	8 000 ms
4	8 000 ms	16 000 ms
5	16 000 ms	32 000 ms

iii-config.yaml (excerpt)

queue_configs:
  email:
    max_retries: 8
    backoff_ms: 2000
    concurrency: 5
    type: standard

After all retries are exhausted, the job moves to a dead-letter queue (DLQ).

See Manage Failed Triggers for DLQ inspection and redrive.

Control concurrency

The concurrency field sets the maximum number of jobs the engine processes simultaneously from a single queue (per engine instance).

iii-config.yaml (excerpt)

queue_configs:
  default:
    concurrency: 10
    type: standard
  payment:
    concurrency: 2
    type: fifo
    message_group_field: transaction_id

Standard queues: the engine pulls up to concurrency jobs simultaneously.
FIFO queues: the engine processes one job at a time (prefetch=1) to preserve ordering, regardless of the concurrency value.

Use low concurrency to protect rate-limited APIs. Use high concurrency for embarrassingly parallel work like image resizing.

Standard vs FIFO Queues

Dimension	Standard	FIFO
Processing model	Up to `concurrency` jobs in parallel	One job at a time (prefetch=1)
Ordering	No guarantees — jobs may complete in any order	Strictly ordered within a message group
`message_group_field`	Not required	Required — must be present and non-null in every payload
Throughput	High — scales with `concurrency`	Lower — trades throughput for ordering
Use cases	Email sends, image processing, notifications	Payments, ledger entries, state machines
Retries	Retried independently, other jobs continue	Retried inline — blocks the queue until success or DLQ

Standard queue flow

Jobs are dequeued and processed concurrently. Each job is independent.

FIFO queue flow

Jobs within the same message group are processed one at a time, strictly in order.

Retry and dead-letter flow

When a job fails, the engine retries it with exponential backoff. After all retries exhaust, the job moves to the DLQ.

Real-World Scenarios

Scenario 1: HTTP API to Queue Pipeline

The most common pattern — an HTTP endpoint accepts a request, responds immediately, and offloads the actual work to a queue. This keeps API response times fast regardless of how long downstream processing takes.

iii-config.yaml

modules:
  - class: modules::queue::QueueModule
    config:
      queue_configs:
        payment:
          max_retries: 10
          concurrency: 2
          type: fifo
          message_group_field: orderId
        email:
          max_retries: 5
          concurrency: 10
          type: standard
          backoff_ms: 2000
      adapter:
        class: modules::queue::BuiltinQueueAdapter
        config:
          store_method: file_based
          file_path: ./data/queue_store

Node / TypeScript
Python
Rust

import { registerWorker, TriggerAction, Logger } from 'iii-sdk'

const iii = registerWorker(process.env.III_URL ?? 'ws://localhost:49134')

iii.registerFunction({ id: 'orders::create' }, async (req) => {
  const logger = new Logger()
  const order = { id: crypto.randomUUID(), ...req.body }

  await iii.trigger({
    function_id: 'orders::process-payment',
    payload: { orderId: order.id, amount: order.total, currency: 'USD' },
    action: TriggerAction.Enqueue({ queue: 'payment' }),
  })

  await iii.trigger({
    function_id: 'emails::confirmation',
    payload: { email: order.email, orderId: order.id },
    action: TriggerAction.Enqueue({ queue: 'email' }),
  })

  await iii.trigger({
    function_id: 'analytics::track',
    payload: { event: 'order_created', orderId: order.id },
    action: TriggerAction.Void(),
  })

  logger.info('Order created', { orderId: order.id })
  return { status_code: 201, body: { orderId: order.id } }
})

iii.registerTrigger({
  type: 'http',
  function_id: 'orders::create',
  config: { api_path: '/orders', http_method: 'POST' },
})

import os
import uuid

from iii import Logger, TriggerAction, register_worker

iii = register_worker(os.environ.get("III_URL", "ws://localhost:49134"))


def create_order(req):
    logger = Logger()
    order = {"id": str(uuid.uuid4()), **req.get("body", {})}

    iii.trigger({
        "function_id": "orders::process-payment",
        "payload": {"orderId": order["id"], "amount": order["total"], "currency": "USD"},
        "action": TriggerAction.Enqueue(queue="payment"),
    })

    iii.trigger({
        "function_id": "emails::confirmation",
        "payload": {"email": order["email"], "orderId": order["id"]},
        "action": TriggerAction.Enqueue(queue="email"),
    })

    iii.trigger({
        "function_id": "analytics::track",
        "payload": {"event": "order_created", "orderId": order["id"]},
        "action": TriggerAction.Void(),
    })

    logger.info("Order created", {"orderId": order["id"]})
    return {"status_code": 201, "body": {"orderId": order["id"]}}


fn = iii.register_function("orders::create", create_order)

iii.register_trigger({
    "type": "http",
    "function_id": fn.id,
    "config": {"api_path": "/orders", "http_method": "POST"},
})

use iii_sdk::{
    register_worker, InitOptions, Logger, RegisterFunction,
    RegisterTriggerInput, TriggerAction, TriggerRequest,
};
use serde_json::{json, Value};

let iii = register_worker("ws://localhost:49134", InitOptions::default());

let iii_clone = iii.clone();
let reg = RegisterFunction::new_async("orders::create", move |req: Value| {
    let iii = iii_clone.clone();
    async move {
        let logger = Logger::new();
        let order_id = uuid::Uuid::new_v4().to_string();

        iii.trigger(TriggerRequest {
            function_id: "orders::process-payment".into(),
            payload: json!({ "orderId": order_id, "amount": req["body"]["total"], "currency": "USD" }),
            action: Some(TriggerAction::Enqueue { queue: "payment".into() }),
            timeout_ms: None,
        }).await?;

        iii.trigger(TriggerRequest {
            function_id: "emails::confirmation".into(),
            payload: json!({ "email": req["body"]["email"], "orderId": order_id }),
            action: Some(TriggerAction::Enqueue { queue: "email".into() }),
            timeout_ms: None,
        }).await?;

        iii.trigger(TriggerRequest {
            function_id: "analytics::track".into(),
            payload: json!({ "event": "order_created", "orderId": order_id }),
            action: Some(TriggerAction::Void),
            timeout_ms: None,
        }).await?;

        logger.info("Order created", Some(json!({ "orderId": order_id })));
        Ok(json!({ "status_code": 201, "body": { "orderId": order_id } }))
    }
});
iii.register_function(reg);

iii.register_trigger(RegisterTriggerInput {
    trigger_type: "http".into(),
    function_id: "orders::create".into(),
    config: json!({ "api_path": "/orders", "http_method": "POST" }),
    metadata: None,
})?;

This example uses all three trigger actions: Enqueue for payment (reliable, ordered) and email (reliable, parallel), and Void for analytics (best-effort).

Scenario 2: Event Fan-Out with Topic Queues

An order system publishes order.created events. Multiple independent services — email notifications, inventory updates, and analytics — each need to process every order. Topic-based queues fan out each message to all subscribers with independent retries per function.

Node / TypeScript
Python
Rust

import { registerWorker, TriggerAction } from 'iii-sdk'

const iii = registerWorker(process.env.III_URL ?? 'ws://localhost:49134')

iii.registerFunction({ id: 'notify::email' }, async (data) => {
  await sendEmail(data.email, `Your order ${data.orderId} is confirmed!`)
  return {}
})

iii.registerFunction({ id: 'inventory::reserve' }, async (data) => {
  for (const item of data.items) {
    await reserveStock(item.sku, item.quantity)
  }
  return {}
})

iii.registerFunction({ id: 'analytics::track' }, async (data) => {
  await trackEvent('order_created', { orderId: data.orderId, total: data.total })
  return {}
})

iii.registerTrigger({
  type: 'queue',
  function_id: 'notify::email',
  config: { topic: 'order.created' },
})

iii.registerTrigger({
  type: 'queue',
  function_id: 'inventory::reserve',
  config: { topic: 'order.created' },
})

iii.registerTrigger({
  type: 'queue',
  function_id: 'analytics::track',
  config: { topic: 'order.created' },
})

iii.registerFunction({ id: 'orders::create' }, async (req) => {
  const order = { id: crypto.randomUUID(), ...req.body }

  await iii.trigger({
    function_id: 'enqueue',
    payload: { topic: 'order.created', data: order },
    action: TriggerAction.Void(),
  })

  return { status_code: 201, body: { orderId: order.id } }
})

from iii import TriggerAction, register_worker

iii = register_worker("ws://localhost:49134")


def send_email_notification(data):
    send_email(data["email"], f"Your order {data['orderId']} is confirmed!")
    return {}


def reserve_inventory(data):
    for item in data["items"]:
        reserve_stock(item["sku"], item["quantity"])
    return {}


def track_analytics(data):
    track_event("order_created", {"orderId": data["orderId"], "total": data["total"]})
    return {}


iii.register_function({"id": "notify::email"}, send_email_notification)
iii.register_function({"id": "inventory::reserve"}, reserve_inventory)
iii.register_function({"id": "analytics::track"}, track_analytics)

for fid in ["notify::email", "inventory::reserve", "analytics::track"]:
    iii.register_trigger({
        "type": "queue",
        "function_id": fid,
        "config": {"topic": "order.created"},
    })


def create_order(req):
    import uuid
    order = {"id": str(uuid.uuid4()), **req.get("body", {})}

    iii.trigger({
        "function_id": "enqueue",
        "payload": {"topic": "order.created", "data": order},
        "action": TriggerAction.Void(),
    })

    return {"status_code": 201, "body": {"orderId": order["id"]}}


iii.register_function({"id": "orders::create"}, create_order)

use iii_sdk::{
    register_worker, InitOptions, RegisterFunction,
    RegisterTriggerInput, TriggerAction, TriggerRequest,
};
use serde_json::{json, Value};

let iii = register_worker("ws://localhost:49134", InitOptions::default());

iii.register_function(RegisterFunction::new_async(
    "notify::email",
    |data: Value| async move {
        send_email(data["email"].as_str().unwrap_or(""), &format!("Your order {} is confirmed!", data["orderId"])).await?;
        Ok(json!({}))
    },
));

iii.register_function(RegisterFunction::new_async(
    "inventory::reserve",
    |data: Value| async move {
        for item in data["items"].as_array().unwrap_or(&vec![]) {
            reserve_stock(item["sku"].as_str().unwrap_or(""), item["quantity"].as_u64().unwrap_or(0)).await?;
        }
        Ok(json!({}))
    },
));

iii.register_function(RegisterFunction::new_async(
    "analytics::track",
    |data: Value| async move {
        track_event("order_created", &json!({ "orderId": data["orderId"], "total": data["total"] })).await?;
        Ok(json!({}))
    },
));

for fid in &["notify::email", "inventory::reserve", "analytics::track"] {
    iii.register_trigger(RegisterTriggerInput {
        trigger_type: "queue".into(),
        function_id: fid.to_string(),
        config: json!({ "topic": "order.created" }),
        metadata: None,
    })?;
}

let iii_clone = iii.clone();
iii.register_function(RegisterFunction::new_async("orders::create", move |req: Value| {
    let iii = iii_clone.clone();
    async move {
        let order_id = uuid::Uuid::new_v4().to_string();

        iii.trigger(TriggerRequest {
            function_id: "enqueue".into(),
            payload: json!({ "topic": "order.created", "data": { "id": order_id, "items": req["body"]["items"] } }),
            action: Some(TriggerAction::Void),
            timeout_ms: None,
        }).await?;

        Ok(json!({ "status_code": 201, "body": { "orderId": order_id } }))
    }
}));

All three functions receive every order.created event independently. If inventory::reserve fails and retries, it does not affect notify::email or analytics::track.

Scenario 3: Financial Transaction Ledger (FIFO)

Transactions for the same account must be applied in order to prevent balance inconsistencies. Different accounts can process in parallel.

iii-config.yaml (excerpt)

queue_configs:
  ledger:
    max_retries: 15
    concurrency: 1
    type: fifo
    message_group_field: account_id
    backoff_ms: 500

Node / TypeScript
Python
Rust

import { registerWorker, TriggerAction } from 'iii-sdk'

const iii = registerWorker(process.env.III_URL ?? 'ws://localhost:49134')

iii.registerFunction({ id: 'transactions::submit' }, async (req) => {
  const { account_id, type, amount } = req.body

  const receipt = await iii.trigger({
    function_id: 'ledger::apply',
    payload: { account_id, type, amount },
    action: TriggerAction.Enqueue({ queue: 'ledger' }),
  })

  return { status_code: 202, body: { receiptId: receipt.messageReceiptId } }
})

iii.registerFunction({ id: 'ledger::apply' }, async (txn) => {
  const { account_id, type, amount } = txn
  if (type === 'deposit') {
    await db.query('UPDATE accounts SET balance = balance + $1 WHERE id = $2', [amount, account_id])
  } else if (type === 'withdraw') {
    const { rows } = await db.query('SELECT balance FROM accounts WHERE id = $1', [account_id])
    if (rows[0].balance < amount) {
      throw new Error('Insufficient funds')
    }
    await db.query('UPDATE accounts SET balance = balance - $1 WHERE id = $2', [amount, account_id])
  }
  return { applied: true }
})

from iii import TriggerAction, register_worker

iii = register_worker("ws://localhost:49134")


def submit_transaction(req):
    account_id = req["body"]["account_id"]
    txn_type = req["body"]["type"]
    amount = req["body"]["amount"]

    receipt = iii.trigger({
        "function_id": "ledger::apply",
        "payload": {"account_id": account_id, "type": txn_type, "amount": amount},
        "action": TriggerAction.Enqueue(queue="ledger"),
    })

    return {"status_code": 202, "body": {"receiptId": receipt["messageReceiptId"]}}


def apply_transaction(txn):
    account_id = txn["account_id"]
    if txn["type"] == "deposit":
        db.execute(
            "UPDATE accounts SET balance = balance + %s WHERE id = %s",
            (txn["amount"], account_id),
        )
    elif txn["type"] == "withdraw":
        balance = db.query("SELECT balance FROM accounts WHERE id = %s", (account_id,))
        if balance < txn["amount"]:
            raise ValueError("Insufficient funds")
        db.execute(
            "UPDATE accounts SET balance = balance - %s WHERE id = %s",
            (txn["amount"], account_id),
        )
    return {"applied": True}


iii.register_function("transactions::submit", submit_transaction)
iii.register_function("ledger::apply", apply_transaction)

use iii_sdk::{
    register_worker, InitOptions, RegisterFunction,
    TriggerAction, TriggerRequest,
};
use serde_json::{json, Value};

let iii = register_worker("ws://localhost:49134", InitOptions::default());

let iii_clone = iii.clone();
let reg = RegisterFunction::new_async("transactions::submit", move |req: Value| {
    let iii = iii_clone.clone();
    async move {
        let receipt = iii.trigger(TriggerRequest {
            function_id: "ledger::apply".into(),
            payload: json!({
                "account_id": req["body"]["account_id"],
                "type": req["body"]["type"],
                "amount": req["body"]["amount"],
            }),
            action: Some(TriggerAction::Enqueue { queue: "ledger".into() }),
            timeout_ms: None,
        }).await?;

        Ok(json!({
            "status_code": 202,
            "body": { "receiptId": receipt["messageReceiptId"] },
        }))
    }
});
iii.register_function(reg);

Because the ledger queue is FIFO with message_group_field: account_id, the deposit for acct_A always completes before the withdrawal. Without FIFO ordering, the withdrawal could execute first and fail with “Insufficient funds” even though the deposit was submitted first.

Scenario 4: Bulk Email with Rate Limiting

A marketing system sends thousands of emails. The SMTP provider has a rate limit. A standard queue with low concurrency prevents overloading the provider while retrying transient failures.

iii-config.yaml (excerpt)

queue_configs:
  bulk-email:
    max_retries: 5
    concurrency: 3
    type: standard
    backoff_ms: 5000

Node / TypeScript
Python
Rust

import { registerWorker, TriggerAction } from 'iii-sdk'

const iii = registerWorker(process.env.III_URL ?? 'ws://localhost:49134')

iii.registerFunction({ id: 'campaigns::launch' }, async (campaign) => {
  for (const recipient of campaign.recipients) {
    await iii.trigger({
      function_id: 'emails::send',
      payload: {
        to: recipient.email,
        subject: campaign.subject,
        body: campaign.body,
      },
      action: TriggerAction.Enqueue({ queue: 'bulk-email' }),
    })
  }

  return { enqueued: campaign.recipients.length }
})

iii.registerFunction({ id: 'emails::send' }, async (email) => {
  const response = await fetch('https://smtp-provider.example/send', {
    method: 'POST',
    body: JSON.stringify(email),
    headers: { 'Content-Type': 'application/json' },
  })

  if (!response.ok) {
    throw new Error(`SMTP error: ${response.status}`)
  }

  return { sent: true }
})

import requests
from iii import TriggerAction, register_worker

iii = register_worker("ws://localhost:49134")


def launch_campaign(campaign):
    for recipient in campaign["recipients"]:
        iii.trigger({
            "function_id": "emails::send",
            "payload": {
                "to": recipient["email"],
                "subject": campaign["subject"],
                "body": campaign["body"],
            },
            "action": TriggerAction.Enqueue(queue="bulk-email"),
        })

    return {"enqueued": len(campaign["recipients"])}


def send_email(email):
    response = requests.post(
        "https://smtp-provider.example/send", json=email
    )
    response.raise_for_status()
    return {"sent": True}


iii.register_function("campaigns::launch", launch_campaign)
iii.register_function("emails::send", send_email)

use iii_sdk::{
    register_worker, InitOptions, RegisterFunction,
    TriggerAction, TriggerRequest,
};
use serde_json::{json, Value};

let iii = register_worker("ws://localhost:49134", InitOptions::default());

let iii_clone = iii.clone();
let reg = RegisterFunction::new_async("campaigns::launch", move |campaign: Value| {
    let iii = iii_clone.clone();
    async move {
        let recipients = campaign["recipients"].as_array().unwrap();
        for recipient in recipients {
            iii.trigger(TriggerRequest {
                function_id: "emails::send".into(),
                payload: json!({
                    "to": recipient["email"],
                    "subject": campaign["subject"],
                    "body": campaign["body"],
                }),
                action: Some(TriggerAction::Enqueue { queue: "bulk-email".into() }),
                timeout_ms: None,
            }).await?;
        }
        Ok(json!({ "enqueued": recipients.len() }))
    }
});
iii.register_function(reg);

With concurrency: 3, at most three emails are in-flight at any time. Failed sends retry with exponential backoff (5s, 10s, 20s, 40s, 80s), protecting the SMTP provider from overload.

Choosing an Adapter

The queue adapter determines where messages are stored and how they are distributed.

Scenario	Recommended Adapter	Why
Local development	`BuiltinQueueAdapter` (`in_memory`)	Zero dependencies, fast iteration
Single-instance production	`BuiltinQueueAdapter` (`file_based`)	Durable across restarts, no external infra
Multi-instance production	`RabbitMQAdapter`	Distributes messages across engine instances

Regardless of which adapter you choose, retry semantics, concurrency enforcement, and FIFO ordering behave identically — the engine owns these behaviors, not the adapter.

See the Queue module reference for adapter configuration and the adapter comparison table for a feature matrix.

When using the RabbitMQ adapter, iii creates exchanges and queues using a predictable naming convention. For a queue named payment, the main queue is iii.__fn_queue::payment, the retry queue is iii.__fn_queue::payment::retry.queue, and the DLQ is iii.__fn_queue::payment::dlq.queue. See Dead Letter Queues for the full resource map.

Queue Config Reference

Field	Type	Default	Description
`max_retries`	`u32`	`3`	Maximum delivery attempts before routing to DLQ
`concurrency`	`u32`	`10`	Maximum concurrent workers for this queue (standard only)
`type`	`string`	`"standard"`	`"standard"` for concurrent processing; `"fifo"` for ordered processing
`message_group_field`	`string`	—	Required for FIFO — the JSON field in the payload used for ordering groups (must be non-null)
`backoff_ms`	`u64`	`1000`	Base retry backoff in milliseconds. Applied exponentially: `backoff_ms × 2^(attempt - 1)`
`poll_interval_ms`	`u64`	`100`	Worker poll interval in milliseconds

For the full module configuration including adapter settings, see the Queue module reference.

Next Steps

Trigger Actions

Understand synchronous, Void, and Enqueue invocation modes

Dead Letter Queues

Handle and redrive failed queue messages

Queue Module Reference

Full configuration reference for queues and adapters

Conditions

Filter queue messages with condition functions

Getting Started

Primitives & Concepts

How-to Guides

Architecture

Modules

Console

Advanced

Examples

Use Queues

Goal

What Are Queues

When to Use Queues

Two Queue Modes

Topic-Based Queues

Named Queues

Standard vs FIFO Queues

Standard queue flow

FIFO queue flow

Retry and dead-letter flow

Real-World Scenarios

Scenario 1: HTTP API to Queue Pipeline

Scenario 2: Event Fan-Out with Topic Queues

Scenario 3: Financial Transaction Ledger (FIFO)

Scenario 4: Bulk Email with Rate Limiting

Choosing an Adapter

Queue Config Reference

Next Steps

Trigger Actions

Dead Letter Queues

Queue Module Reference

Conditions

Getting Started

Primitives & Concepts

How-to Guides

Architecture

Modules

Console

Advanced

Examples

​Goal

​What Are Queues

​When to Use Queues

​Two Queue Modes

​Topic-Based Queues

​Named Queues

​Standard vs FIFO Queues

​Standard queue flow

​FIFO queue flow

​Retry and dead-letter flow

​Real-World Scenarios

​Scenario 1: HTTP API to Queue Pipeline

​Scenario 2: Event Fan-Out with Topic Queues

​Scenario 3: Financial Transaction Ledger (FIFO)

​Scenario 4: Bulk Email with Rate Limiting

​Choosing an Adapter

​Queue Config Reference

​Next Steps

Trigger Actions

Dead Letter Queues

Queue Module Reference

Conditions

Goal

What Are Queues

When to Use Queues

Two Queue Modes

Topic-Based Queues

Named Queues

Standard vs FIFO Queues

Standard queue flow

FIFO queue flow

Retry and dead-letter flow

Real-World Scenarios

Scenario 1: HTTP API to Queue Pipeline

Scenario 2: Event Fan-Out with Topic Queues

Scenario 3: Financial Transaction Ledger (FIFO)

Scenario 4: Bulk Email with Rate Limiting

Choosing an Adapter

Queue Config Reference

Next Steps