Observers

Observe a function

observe(a, b) means: when a runs, b runs.

Node / TypeScript
Python
Rust
Go

worker.observe("orders::create", "alerts::new-order");

worker.observe("orders::create", "alerts::new-order")

worker.observe("orders::create", "alerts::new-order");

worker.ObserveFunction("orders::create", "alerts::new-order")

An observer connects two functions in one direction. After each successful call of the observed function (a), iii runs the target (b) with the call’s input, its result and its caller. The examples on this page react to orders::create, a function that another worker registers.

Node / TypeScript
Python
Rust
Go

import { registerWorker, type Observed } from "iii-sdk";
import { Logger } from "@iii-dev/helpers/observability";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url);
const logger = new Logger();

type Order = { id: string; email: string; total: number };

// The target: a regular function. It receives the Observed payload of each call.
worker.registerFunction("alerts::new-order", async (event: Observed<Order>) => {
  logger.info(`order ${event.input?.id} totals ${event.input?.total}`);
  return null;
});

// When orders::create runs, run alerts::new-order.
const newOrders = worker.observe("orders::create", "alerts::new-order");
await newOrders.ready(); // resolves once the observer is active

import os
from iii import InitOptions, register_worker
from iii_helpers.observability import Logger

worker = register_worker(os.environ.get("III_URL"), InitOptions(worker_name="alerts"))
logger = Logger()

# The target: a regular function. It receives the Observed payload of each call.
def new_order(event: dict) -> None:
    order = event.get("input", {})
    logger.info(f"order {order.get('id')} totals {order.get('total')}")

worker.register_function("alerts::new-order", new_order)

# When orders::create runs, run alerts::new-order.
new_orders = worker.observe("orders::create", "alerts::new-order")
new_orders.ready()  # blocks until active; use `await new_orders.ready_async()` in asyncio code

use iii_helpers::observability::Logger;
use iii_sdk::{Error, InitOptions, Observed, RegisterFunction, register_worker};
use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
struct Order {
    id: String,
    email: String,
    total: f64,
}

let url = std::env::var("III_URL").expect("III_URL must be set");
let worker = register_worker(&url, InitOptions::default());

// The target: a regular function. It receives the Observed payload of each call.
worker.register_function(
    "alerts::new-order",
    RegisterFunction::new(|event: Observed<Order>| {
        if let Some(order) = &event.input {
            Logger::new().info(&format!("order {} totals {}", order.id, order.total), None);
        }
        Ok::<_, Error>(())
    }),
);

// When orders::create runs, run alerts::new-order.
let new_orders = worker.observe("orders::create", "alerts::new-order");
new_orders.ready().await?; // resolves once the observer is active

import "github.com/iii-hq/iii/sdk/packages/go/iii/observability"

type Order struct {
    ID    string  `json:"id"`
    Email string  `json:"email"`
    Total float64 `json:"total"`
}

var logger observability.Logger

worker := iii.RegisterWorker(os.Getenv("III_URL"))

// The target: a regular function. It receives the Observed payload of each call.
worker.RegisterFunction("alerts::new-order", func(ctx context.Context, data json.RawMessage) (any, error) {
    var event iii.Observed
    if err := json.Unmarshal(data, &event); err != nil {
        return nil, err
    }
    logger.Info(ctx, "new order", event.Input)
    return nil, nil
})

// When orders::create runs, run alerts::new-order.
newOrders, err := worker.ObserveFunction("orders::create", "alerts::new-order")
if err != nil {
    log.Fatal(err) // the request could not be built (a missing id)
}
if err := newOrders.Ready(ctx); err != nil {
    log.Fatal(err) // an *iii.ObserveRejectionError carries the engine's code
}

What can be observed

The observed function can be any function on the engine: one that your team wrote, or one a worker hosts to expose something that happens, such as an HTTP request (http::on-request), a queue message (queue::on-message) or a row change (database::on-row-changed). The target is any registered function. Any worker can observe any function, private ones included, unless the function is registered with observable: false or RBAC restricts it (see Authorization). By default, an observer is accepted as soon as it is added, and its target runs on every call its capture fires on and its filters match. The worker that hosts the function can override each part of that with a hook: onObserverAdded sets something up for the observer, or rejects it; onObservationMatch decides which calls reach which observer; and onObservationDeliver delivers the calls itself and decides what the caller gets back (see Hooks). If unsure, refer to that worker’s documentation.

Defaults

The target runs after each successful call of the observed function.
The target receives the Observed payload of the call, with its input, its output, and its caller.
The caller of the observed function gets its result without waiting for the target. Each call runs the target at most once, and a failure of the target does not change the call.
The observer stays pending until both functions are registered, in any order. The handle that observe returns reports this in its state property, and its ready() resolves once the observer is active. Calls that happen before then can be missed.
You call observe once at startup, like registerFunction. It returns immediately, even before the connection is up: the SDK sends the observer when it connects and replays it with the same id after every reconnect.

This page is about running your function when another one runs. To decide what happens when the functions your worker hosts are observed, see Observable functions. Coming from 0.25 triggers? registerTrigger becomes observe: see Upgrading from 0.25.x to 0.26.x.

Args and filters

An observer can carry args and filters. args go to the worker that hosts the function, which reads them to set something up for the observer (a route, a database trigger, a queue consumer) and to filter which calls the observer receives. This is custom logic: each function defines the args its use case needs. filters are standard: the iii engine applies them to each call, the same way for every function.

Node / TypeScript
Python
Rust
Go

worker.observe(
  {
    function_id: "database::on-row-changed",
    args: { db: "primary", table: "orders", ops: ["insert", "update"] }, // for the database worker
    filters: { "caller.parent.worker_name": { $ne: "order-projector" } }, // for iii
  },
  "orders::project",
);

worker.observe(
    {
        "function_id": "database::on-row-changed",
        "args": {"db": "primary", "table": "orders", "ops": ["insert", "update"]},  # for the database worker
        "filters": {"caller.parent.worker_name": {"$ne": "order-projector"}},  # for iii
    },
    "orders::project",
)

use iii_sdk::Source;
use serde_json::json;

worker.observe(
    Source::function("database::on-row-changed")
        // for the database worker
        .args(json!({ "db": "primary", "table": "orders", "ops": ["insert", "update"] }))
        // for iii
        .filters(json!({ "caller.parent.worker_name": { "$ne": "order-projector" } })),
    "orders::project",
);

worker.Observe(iii.Source{
    FunctionID: "database::on-row-changed",
    // for the database worker
    Args: iii.Args{"db": "primary", "table": "orders", "ops": []string{"insert", "update"}},
    // for iii
    Filters: iii.Filters{"caller.parent.worker_name": iii.Op{"$ne": "order-projector"}},
}, "orders::project")

Args

The engine checks args against the args_format of the function, a JSON Schema the function’s worker declares when it registers the function. The observer fails with INVALID_ARGS when the args do not match, or when the function has no args_format and the args are not empty. The worker can also refuse an observer whose args it cannot serve, with a code of its own (for example UNKNOWN_DB or HTTP_ROUTE_CONFLICT). Each function documents its args: see its worker’s page on workers.iii.dev.

Filters

A filter is a JSON object. Each key is a path in the call that starts with input, output, error, caller, or metadata. Each value is an operator object, for example { "$eq": "EUR" }. A literal value such as "EUR" is not allowed.

Node / TypeScript
Python
Rust
Go

worker.registerFunction("alerts::big-order", async (event: Observed<Order>) => {
  logger.info(
    `order ${event.input?.id} totals ${event.input?.total}, placed by ${event.caller?.worker_name}`,
  );
  return null;
});

// After each successful call of orders::create with total >= 500, run alerts::big-order.
await worker
  .observe(
    { function_id: "orders::create", filters: { "input.total": { $gte: 500 } } },
    "alerts::big-order",
  )
  .ready();

def big_order(event: dict) -> None:
    order = event.get("input", {})
    caller = event.get("caller", {})
    logger.info(f"order {order.get('id')} totals {order.get('total')}, placed by {caller.get('worker_name')}")

worker.register_function("alerts::big-order", big_order)

# After each successful call of orders::create with total >= 500, run alerts::big-order.
worker.observe(
    {"function_id": "orders::create", "filters": {"input.total": {"$gte": 500}}},
    "alerts::big-order",
).ready()

use iii_sdk::Source;
use serde_json::json;

worker.register_function(
    "alerts::big-order",
    RegisterFunction::new(|event: Observed<Order>| {
        if let Some(order) = &event.input {
            Logger::new().info(&format!("order {} totals {}", order.id, order.total), None);
        }
        Ok::<_, Error>(())
    }),
);

// After each successful call of orders::create with total >= 500, run alerts::big-order.
worker
    .observe(
        Source::function("orders::create").filters(json!({ "input.total": { "$gte": 500 } })),
        "alerts::big-order",
    )
    .ready()
    .await?;

worker.RegisterFunction("alerts::big-order", func(ctx context.Context, data json.RawMessage) (any, error) {
    var event iii.Observed
    if err := json.Unmarshal(data, &event); err != nil {
        return nil, err
    }
    logger.Info(ctx, "big order", event.Input)
    return nil, nil
})

// After each successful call of orders::create with total >= 500, run alerts::big-order.
bigOrders, err := worker.Observe(iii.Source{
    FunctionID: "orders::create",
    Filters:    iii.Filters{"input.total": iii.Op{"$gte": 500}},
}, "alerts::big-order")
if err != nil {
    log.Fatal(err)
}
if err := bigOrders.Ready(ctx); err != nil {
    log.Fatal(err)
}

Operators

The operators of one operator object must all match: { "input.total": { "$gte": 100, "$lt": 500 } } is a range. See the filter language reference for the full rules and the limits.

Comparison

$eq, $ne, $in, $nin, $gt, $gte, $lt, $lte. Equality is JSON deep equality, with no coercion between types. Ordering applies within one type only: numbers compare numerically, strings by code point.

{
  "input.currency": { "$eq": "EUR" },
  "caller.worker_name": { "$ne": "importer" },
  "input.country": { "$in": ["DE", "FR"] },
  "input.status": { "$nin": ["draft", "cancelled"] },
  "input.total": { "$gt": 0, "$lte": 10000 },
  "input.quantity": { "$gte": 2 },
  "input.created_at": { "$lt": "2025-01-01" }
}

Existence and `null`

$exists takes a boolean and matches on the presence of the path. null is a value: { "$exists": true } matches null, and { "$eq": null } matches only null, and will not match a missing path. Only $ne, $nin, and $exists: false match a missing path.

{
  "error": { "$exists": true },
  "input.coupon": { "$exists": false },
  "input.parent_id": { "$eq": null }
}

Strings

$prefix matches a string that starts with the operand.

{ "input.path": { "$prefix": "/api/" } }

Arrays

A path does not read inside an array. When input.items is an array of objects, the path input.items.sku is a missing path: it does not read the sku of each item. To match on the elements, use $any. It matches when at least one element matches its operand. For scalar elements, the operand is an operator object. For object elements, the operand is a filter whose paths start at @, the current element.

{
  "input.tags": { "$any": { "$eq": "vip" } },
  "input.items": { "$any": { "@.sku": { "$eq": "A-1" }, "@.qty": { "$gte": 10 } } }
}

Logical

$and and $or take a non-empty array of filters. $not takes one filter.

{
  "$and": [{ "input.total": { "$gte": 500 } }, { "input.currency": { "$eq": "EUR" } }],
  "$or": [{ "input.priority": { "$eq": "high" } }, { "input.total": { "$gte": 10000 } }],
  "$not": { "input.path": { "$prefix": "/health" } }
}

$and, $or and $not are clauses of a filter. They cannot go inside an operator object on a path.

Validation

When the observer is created, the SDK and the engine check the filter. If it is not valid, the observer fails with INVALID_FILTER. The engine also checks the input.* and output.* paths against the request_format and response_format of the function. If a closed schema does not have a path, the observer fails with INVALID_FILTER_PATH. If the function is not registered yet, the engine does this check when the function registers.

What the target receives

capture sets what the target receives, and with it after which calls the target runs.

The payload

The target receives the Observed payload of the call. It always has the ids, status, and the timings. It has input, output, error, caller, and metadata when capture includes them and the call has them: output only after a successful call, error only after a failed one. A field that is not captured, or that the call does not have, is absent, never null. This differs from 0.25, which sent "output": null to an observer that captured output when a successful call reported no output.

Field	Meaning
`observer_id`, `activation_id`	The observer, and this activation of it.
`function_id`, `namespace`	The observed function.
`invocation_id`	The observed call.
`status`	`success` or `error`.
`input`, `output`, `error`	The observed payload, its result, or its `{ code, message, stacktrace? }` error.
`caller`	Who made the observed call. See Caller.
`metadata`	The metadata of the observed call.
`started_at_ms`, `duration_ms`	When the observed invocation started, and how long it ran.

Node / TypeScript
Python
Rust
Go

worker.registerFunction("emails::send-receipt", async ({ input }: Observed<Order>) => {
  await sendReceipt(input.email, input);
  return null;
});

// After each successful call of orders::create, run emails::send-receipt with the order.
worker.observe({ function_id: "orders::create", capture: ["input"] }, "emails::send-receipt");

def send_receipt(observed: dict) -> None:
    order = observed["input"]
    mailer.send_receipt(order["email"], order)

worker.register_function("emails::send-receipt", send_receipt)

# After each successful call of orders::create, run emails::send-receipt with the order.
worker.observe({"function_id": "orders::create", "capture": ["input"]}, "emails::send-receipt")

use iii_sdk::{Capture, Observed, Source};

worker.register_function(
    "emails::send-receipt",
    RegisterFunction::new(|observed: Observed<Order>| {
        let order = observed.input.expect("captured");
        send_receipt(&order.email, &order);
        Ok::<_, Error>(())
    }),
);

// After each successful call of orders::create, run emails::send-receipt with the order.
worker.observe(
    Source::function("orders::create").capture([Capture::Input]),
    "emails::send-receipt",
);

worker.RegisterFunction("emails::send-receipt", func(ctx context.Context, data json.RawMessage) (any, error) {
    var observed iii.Observed
    if err := json.Unmarshal(data, &observed); err != nil {
        return nil, err
    }
    var order Order
    if err := json.Unmarshal(observed.Input, &order); err != nil {
        return nil, err
    }
    return nil, sendReceipt(ctx, order.Email, order)
})

// After each successful call of orders::create, run emails::send-receipt with the order.
worker.Observe(iii.Source{
    FunctionID: "orders::create",
    Capture:    []iii.Capture{iii.CaptureInput},
}, "emails::send-receipt")

`capture`

capture lists the fields the payload carries: any of input, output, error, caller, and metadata. The default is ["input", "output", "caller"]. metadata is opt-in: the call’s metadata reaches the target only when capture names it. Earlier 0.26 release candidates of this change added metadata to the default; 0.26 does not. To route calls by their metadata (for example a for_observer key), use the hosting worker’s onObservationMatch or onObservationDeliver hook, which receives the call’s metadata whatever the observers capture. A filter can also read metadata.* without metadata in capture. The fields also decide after which calls the target runs:

input and output run it after a successful call.
error runs it after a failed call.
caller and metadata add to the payload only. A capture must name at least one of input, output, and error; one with only caller or metadata fails with INVALID_ARGS. The exception is capture: [], which runs the target after successful calls with no payload fields.
[] runs the target after a successful call, with the ids, status, and the timings only: no payload moves to it. An observer that only counts calls uses it.

`capture`	The target runs after
default, `["input"]`, `["output", "caller"]`	a successful call
`["error"]`, `["error", "caller"]`	a failed call
`["input", "error"]`, `["output", "error"]`	each call, successful or failed
`[]`	a successful call (no payload)

The default does not include error, so a default observer runs after successful calls only. A call the engine cancels (for example, a call it routes whose caller stopped waiting) does not run the target, for any capture. The engine counts it as a dropped call with the reason CANCELLED in the iii.observer.dropped.total metric. A call the hosting worker runs in its own process is not cancelled when its caller stops waiting: its handler runs to the end, and its outcome reaches the observers like any other. Once capture has decided which outcomes run the target, a filter can read any field of the call, including the fields that capture leaves out of the payload. To keep observed calls small, capture only what the target reads. See Observed call size. To run the target after failed calls with their error, capture error:

Node / TypeScript
Python
Rust
Go

// After each failed call: the error, and who made the call.
worker.observe(
  { function_id: "orders::create", capture: ["error", "caller"] },
  "alerts::order-failed",
);

# After each failed call: the error, and who made the call.
worker.observe(
    {"function_id": "orders::create", "capture": ["error", "caller"]},
    "alerts::order-failed",
)

use iii_sdk::{Capture, Source};

// After each failed call: the error, and who made the call.
worker.observe(
    Source::function("orders::create").capture([Capture::Error, Capture::Caller]),
    "alerts::order-failed",
);

// After each failed call: the error, and who made the call.
worker.Observe(iii.Source{
    FunctionID: "orders::create",
    Capture:    []iii.Capture{iii.CaptureError, iii.CaptureCaller},
}, "alerts::order-failed")

To get the input of every call, successful or failed, capture input and error: the target runs after both, and the payload’s status says which.

Failed calls with their input

To run the target after failed calls only, and still get each call’s input, capture input and error and keep only the calls that have an error with a filter:

Node / TypeScript
Python
Rust
Go

// After each failed call only: its input and its error.
worker.observe(
  {
    function_id: "orders::create",
    capture: ["input", "error"],
    filters: { error: { $exists: true } },
  },
  "alerts::order-failed",
);

# After each failed call only: its input and its error.
worker.observe(
    {
        "function_id": "orders::create",
        "capture": ["input", "error"],
        "filters": {"error": {"$exists": True}},
    },
    "alerts::order-failed",
)

use iii_sdk::{Capture, Source};
use serde_json::json;

// After each failed call only: its input and its error.
worker.observe(
    Source::function("orders::create")
        .capture([Capture::Input, Capture::Error])
        .filters(json!({ "error": { "$exists": true } })),
    "alerts::order-failed",
);

// After each failed call only: its input and its error.
worker.Observe(iii.Source{
    FunctionID: "orders::create",
    Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError},
    Filters:    iii.Filters{"error": iii.Op{"$exists": true}},
}, "alerts::order-failed")

capture: ["input", "error"] fires on both outcomes, and the filter drops the successful calls before they reach the observing worker. The engine evaluates it for the calls it routes and starts no target for a successful one. For the calls a worker on a 0.26 SDK runs in-process, that SDK evaluates it and does not report a successful call to the engine at all. After each failed call the target gets status: "error", the input, and the error. Capturing error alone would fire on failures only, but its payload would not carry the input. The observing worker needs a 0.26 SDK. An observer from a worker without the observe.capture feature keeps the 0.25 semantics and fires on successful calls only, so with this filter it never fires (see Compatibility).

Metadata

The observer’s metadata option, set in the policy, reaches the target function as its second argument on every activation.

The payload’s metadata field is the metadata of the observed call, present only when capture names metadata: the default leaves it out. The observer’s metadata option is the target function’s second argument.

The target function’s return value

The iii engine does not use the return values of target functions. However, an observable function can use onObservationDeliver to get that return value from the target function and use it, for example to answer an HTTP request or to acknowledge a queue message. The function’s documentation says what its target functions must return. Likewise, observers of a target function receive the return value as the output of its call. See Chain reactions.

Policy and metadata

An observer can carry a policy. The policy sets limits for the observer itself. An activation is one run of the target.

Option	Default	Meaning
`id`	a random UUID	The observer id. An `observe` with an id that a live handle of this worker, or another worker’s observer, already holds fails with `OBSERVER_ID_CONFLICT`. The id is free again once that handle is disposed. With an id of your own, a restarted worker takes back its `suspended` observer (see Observer states).
`max_fires`	unlimited	The observer becomes `disposed` after this number of activations (reason `MAX_FIRES`).
`expires_in_ms`	never	The observer becomes `disposed` after this time (reason `EXPIRED`).
`max_in_flight`	`observers.default_max_in_flight` (64)	The maximum number of activations that run at the same time. More activations wait in a queue of `observers.max_queue_per_observer` (1024). When the queue is full, or when the queues of all observers hold 64 MiB of observed calls, the engine drops new activations (`OVERFLOW`). With `1`, activations run in the order they arrive.
`metadata`	none	JSON that each activation gives to the target as its second argument. See Metadata.

Node / TypeScript
Python
Rust
Go

worker.observe(
  {
    function_id: "payments::charge",
    filters: { "input.amount": { $gte: 500 } },
    capture: ["error", "caller"],
  },
  "alerts::payment-failed",
  { max_in_flight: 4, metadata: { team: "billing" } },
);

worker.observe(
    {
        "function_id": "payments::charge",
        "filters": {"input.amount": {"$gte": 500}},
        "capture": ["error", "caller"],
    },
    "alerts::payment-failed",
    {"max_in_flight": 4, "metadata": {"team": "billing"}},
)

use iii_sdk::{Capture, ObserverOptions, Source};
use serde_json::json;

worker.observe_with(
    Source::function("payments::charge")
        .filters(json!({ "input.amount": { "$gte": 500 } }))
        .capture([Capture::Error, Capture::Caller]),
    "alerts::payment-failed",
    ObserverOptions::default()
        .max_in_flight(4)
        .metadata(json!({ "team": "billing" })),
);

worker.Observe(iii.Source{
    FunctionID: "payments::charge",
    Filters:    iii.Filters{"input.amount": iii.Op{"$gte": 500}},
    Capture:    []iii.Capture{iii.CaptureError, iii.CaptureCaller},
}, "alerts::payment-failed", iii.ObserverOptions{
    MaxInFlight: 4,
    Metadata:    json.RawMessage(`{"team":"billing"}`),
})

The engine does not retry a failed activation. A worker that delivers calls to the observers itself can retry on its own terms, which its documentation states. The engine counts each dropped call in the iii.observer.dropped.total metric, by function and reason.

Observer handle

observe returns a handle with the observer’s id and state, and three methods:

ready() resolves when the observer is active. It rejects with an ObserveRejection that carries the reason’s code when the observer turns error or disposed first.
onStateChange(listener) reports each state change with its reason, and returns a function that unsubscribes. Listeners run outside the connection’s read path, so a listener can call the worker.
dispose() removes the observer. The SDK does not replay a disposed observer after a reconnect. On an observer the engine refused, dispose() has nothing to remove and makes no request to the engine.

Node / TypeScript
Python
Rust
Go

const newOrders = worker.observe("orders::create", "alerts::new-order");

newOrders.onStateChange((state, reason) => {
  logger.info(`observer ${newOrders.id}: ${state}`, { reason });
});
await newOrders.ready();

// later: remove the observer
await newOrders.dispose();

new_orders = worker.observe("orders::create", "alerts::new-order")

def log_state(state, reason):
    logger.info(f"observer {new_orders.id}: {state}")

new_orders.on_state_change(log_state)
new_orders.ready()

# later: remove the observer
new_orders.dispose()  # or: await new_orders.dispose_async()

let new_orders = worker.observe("orders::create", "alerts::new-order");

let _listener = new_orders.on_state_change(|state, reason| {
    println!("observer state: {state:?} {reason:?}");
});
new_orders.ready().await?;

// later: remove the observer
new_orders.dispose().await?;

newOrders, err := worker.ObserveFunction("orders::create", "alerts::new-order")
if err != nil {
    log.Fatal(err)
}

unsubscribe := newOrders.OnStateChange(func(state iii.ObserverState, reason *iii.ObserveReason) {
    log.Printf("observer %s: %s", newOrders.ID(), state)
})
defer unsubscribe()
if err := newOrders.Ready(ctx); err != nil {
    log.Fatal(err)
}

// later: remove the observer
if err := newOrders.Dispose(ctx); err != nil {
    log.Fatal(err)
}

A target that runs in the observing worker can stop its own observer: it calls dispose() on the handle, or it calls engine::observers::dispose with the observer_id from the payload. To stop after a number of calls or after a time, set max_fires or expires_in_ms in the policy.

Observer states

State	Meaning
`pending`	Waiting. No worker hosts the function yet (`OWNER_UNAVAILABLE`), the target is not registered yet (`TARGET_UNAVAILABLE`), or the worker that hosts the function has not accepted the observer yet.
`active`	The target runs for each matching call.
`suspended`	The observing worker disconnected (`OBSERVER_DISCONNECTED`), or the target was removed after the observer was `active` (`TARGET_UNAVAILABLE`).
`error`	Rejected. The reason has the code, for example `INVALID_ARGS`, `FORBIDDEN`, or a code from the function’s worker.
`disposed`	Removed permanently, by `dispose()`, `max_fires`, `expires_in_ms`, or because the observing worker did not reconnect in time.

An observer belongs to the connection that created it. When that worker disconnects, its observers become suspended for observers.observer_grace_ms (30 s by default). If the worker reconnects in this time, it keeps its observers and their counters. If not, the observers become disposed.
A restarted worker is a new connection. With an id of your own (see Policy and metadata), the restarted worker’s observe of that id replaces its suspended observer, when both connections have the same worker name, namespace and principal: both on ports without rbac, or both RBAC sessions whose auth function gave them the same non-empty context. An observer with a random id stays suspended until the grace period ends.
Set the worker name with III_WORKER_NAME or the worker name option of the SDK; a worker that iii compose starts gets III_WORKER_NAME from it. The default name, hostname:pid, changes when the process restarts, so the restarted worker’s observe of the id fails with OBSERVER_ID_CONFLICT until the grace period ends.
When a worker reconnects, its SDK sends each of its observers to the engine again, with the same id. The engine keeps a record of a disposed observer for observers.disposed_retention_ms (1 hour by default). In this time, the engine does not create the observer again, and the handle becomes disposed, so max_fires and expires_in_ms stay in effect after a reconnect. After this time, the engine creates the observer again with new counters, as after an engine restart.
An observer with neither max_fires nor expires_in_ms is not lost when its worker is away for longer than the grace period. When the engine answers that it was disposed because the worker was away (OBSERVER_DISCONNECTED), the SDK observes it again at once, with the same id, source, target, metadata and options, and logs a line naming the id. The handle goes to pending and then active, and does not become disposed. The engine creates a new observer, so its count of fires starts again at 0, and the worker that hosts the observed function gets it as a new observer (onObserverAdded runs). An observer you are disposing is not observed again.
To know when an observer becomes disposed, and why, use onStateChange on its handle. engine::observers::list and get do not show disposed observers.
When the worker that hosts the observed function disconnects, its observers go back to pending and become active again when the function is registered again.

Engine-hosted event sources

The engine hosts event sources as functions in the default namespace, and you observe them with filters on the event. A worker in any namespace observes engine::* and configuration::* sources without naming a namespace. For observability::* sources, a worker in another namespace names namespace: "default".

Function	Event
`engine::on-functions-changed`	`{ event: "functions_changed", functions: [{ function_id, namespace, worker_name, description?, metadata?, args_format?, private?, observable?, observer_count }] }`. `args_format`, `private: true`, and `observable: false` appear when the function sets them. Changes within about 100 ms are coalesced into one event.
`engine::on-workers-changed`	`{ event, worker_id }`, `event` one of `worker_connected`, `worker_disconnected`, `worker_metadata_updated`
`engine::on-observers-changed`	`{ observer_id, function_id, target, state, reason? }`, one event for each state change of an observer.
`configuration::on-updated`	`{ type: "configuration", event_type, id, name, description, schema, metadata? }`. `event_type` is `configuration:registered`, `configuration:updated`, or `configuration:deleted`. The event carries no value: read it with `configuration::get`.
`observability::on-log`	One log record: `severity_text`, `severity_number`, `body`, `attributes`, `service_name`, `trace_id`, `span_id`, …
`observability::on-span`	`{ trace_ids: [...] }`: the traces that recorded spans in one window of about 300 ms.

Each of them is a private function whose handler returns the event. So input and output are both the event: capture: ["input"] or capture: ["output"] gives the target the event in the payload. Their calls never fail, so an observer that captures error but neither input nor output is accepted, but its target never runs. Their request_format is closed at the top level, so an input.* filter path that is not in the event fails with INVALID_FILTER_PATH. Only the engine calls them. A call from a worker fails with FORBIDDEN. The SDK exports their ids as EngineSources, for example EngineSources.WORKERS_CHANGED.

Node / TypeScript
Python
Rust
Go

// Reload the billing worker's settings when its configuration entry changes.
worker.observe(
  {
    function_id: "configuration::on-updated",
    capture: ["input"],
    filters: {
      "input.id": { $eq: "billing" },
      "input.event_type": { $eq: "configuration:updated" },
    },
  },
  "billing::reload",
);

// Alert on error logs.
worker.observe(
  {
    function_id: "observability::on-log",
    namespace: "default",
    capture: ["input"],
    filters: { "input.severity_number": { $gte: 17 } }, // OpenTelemetry ERROR and above
  },
  "alerts::log",
);

// Refresh a cache when workers change.
worker.observe({ function_id: "engine::on-workers-changed", capture: ["input"] }, "ops::refresh-workers");

worker.observe(
    {
        "function_id": "configuration::on-updated",
        "capture": ["input"],
        "filters": {
            "input.id": {"$eq": "billing"},
            "input.event_type": {"$eq": "configuration:updated"},
        },
    },
    "billing::reload",
)

# Alert on error logs.
worker.observe(
    {
        "function_id": "observability::on-log",
        "namespace": "default",
        "capture": ["input"],
        "filters": {"input.severity_number": {"$gte": 17}},  # OpenTelemetry ERROR and above
    },
    "alerts::log",
)

use iii_sdk::{Capture, Source};
use serde_json::json;

worker.observe(
    Source::function("configuration::on-updated")
        .capture([Capture::Input])
        .filters(json!({
            "input.id": { "$eq": "billing" },
            "input.event_type": { "$eq": "configuration:updated" },
        })),
    "billing::reload",
);

// Alert on error logs.
worker.observe(
    Source::function("observability::on-log")
        .namespace("default")
        .capture([Capture::Input])
        .filters(json!({ "input.severity_number": { "$gte": 17 } })),
    "alerts::log",
);

worker.Observe(iii.Source{
    FunctionID: "configuration::on-updated",
    Capture:    []iii.Capture{iii.CaptureInput},
    Filters: iii.Filters{
        "input.id":         iii.Op{"$eq": "billing"},
        "input.event_type": iii.Op{"$eq": "configuration:updated"},
    },
}, "billing::reload")

// Alert on error logs.
worker.Observe(iii.Source{
    FunctionID: "observability::on-log",
    Namespace:  "default",
    Capture:    []iii.Capture{iii.CaptureInput},
    Filters:    iii.Filters{"input.severity_number": iii.Op{"$gte": 17}},
}, "alerts::log")

Chain reactions

A call of a target is a normal call, so another observer can observe the target. In a chain, each observer observes the target of the observer before it, and capture: ["output"] hands it the previous step’s return value as the payload’s output.

Node / TypeScript
Python
Rust
Go

// Inventory worker: after each successful call of orders::create, reserve the items.
inventoryWorker.registerFunction("inventory::reserve", async ({ input }: Observed<Order>) =>
  reserveItems(input),
);
inventoryWorker.observe({ function_id: "orders::create", capture: ["input"] }, "inventory::reserve");

// Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
shippingWorker.registerFunction("shipping::schedule", async ({ output }: Observed<unknown, Reservation>) =>
  scheduleShipment(output),
);
shippingWorker.observe(
  { function_id: "inventory::reserve", capture: ["output"] },
  "shipping::schedule",
);

# Inventory worker: after each successful call of orders::create, reserve the items.
inventory_worker.register_function("inventory::reserve", lambda observed: reserve_items(observed["input"]))
inventory_worker.observe({"function_id": "orders::create", "capture": ["input"]}, "inventory::reserve")

# Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
shipping_worker.register_function("shipping::schedule", lambda observed: schedule_shipment(observed["output"]))
shipping_worker.observe({"function_id": "inventory::reserve", "capture": ["output"]}, "shipping::schedule")

use iii_sdk::{Capture, Source};

// Inventory worker: after each successful call of orders::create, reserve the items.
inventory_worker.register_function("inventory::reserve", RegisterFunction::new(reserve_items));
inventory_worker.observe(
    Source::function("orders::create").capture([Capture::Input]),
    "inventory::reserve",
);

// Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
shipping_worker.register_function("shipping::schedule", RegisterFunction::new(schedule_shipment));
shipping_worker.observe(
    Source::function("inventory::reserve").capture([Capture::Output]),
    "shipping::schedule",
);

// Inventory worker: after each successful call of orders::create, reserve the items.
inventoryWorker.RegisterFunction("inventory::reserve", reserveItems)
inventoryWorker.Observe(iii.Source{
    FunctionID: "orders::create",
    Capture:    []iii.Capture{iii.CaptureInput},
}, "inventory::reserve")

// Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
shippingWorker.RegisterFunction("shipping::schedule", scheduleShipment)
shippingWorker.Observe(iii.Source{
    FunctionID: "inventory::reserve",
    Capture:    []iii.Capture{iii.CaptureOutput},
}, "shipping::schedule")

In Rust and Go, reserve_items / reserveItems and schedule_shipment / scheduleShipment take the Observed payload and read its input or output. To add a step, observe the last function in the chain. To branch, add two observers to the same function. Each step adds 1 to the depth of the call: see Loop guard.

Inspect observers

The engine lists observers through three built-in functions:

Function	Input	Returns
`engine::observers::list`	`{ function_id?, namespace?, worker? }`	`{ observers: [...] }`: id, observed function, args, normalized filters, target, `capture`, `fires_on` (the outcomes that fire the observer: `success`, `error`, or both), observing worker, state, reason, `fires` (the number of activations), drops and activation failures by reason, policy.
`engine::observers::get`	`{ id }`	One observer, or `OBSERVER_NOT_FOUND`.
`engine::observers::dispose`	`{ id }`	Disposes an observer. Only the observing worker, or a caller without RBAC restrictions, may do it.

The output does not include the observer’s metadata option. With RBAC, the output holds only the observers of functions the session’s grants cover.

iii call engine::observers::list function_id=orders::create

To react to observers as they change, observe engine::on-observers-changed.

Errors

When an observer cannot be created, it becomes error, and ready() fails with one of these codes:

Code	When
`UNSUPPORTED`	The engine is older than 0.25 (the SDK gives this code), or `capture` has a value that the engine does not support.
`INVALID_ARGS`	`args` do not match the `args_format` of the function, or the function takes no `args`. Also: an empty or too-long observer id, an empty `function_id`, or `max_fires`, `expires_in_ms`, or `max_in_flight` set to `0`, or a non-empty `capture` that names none of `input`, `output`, and `error` (not checked for an observer that gets the 0.25 semantics: see Legacy `on` and `deliver`).
`INVALID_FILTER`	The filter is not valid, uses a literal value in place of an operator object, or is larger than a limit.
`INVALID_FILTER_PATH`	A filter path is not in the function’s closed `request_format` or `response_format`.
`NOT_OBSERVABLE`	The function is registered with `observable: false`.
`FORBIDDEN`	RBAC does not allow the session to call the observed function or the target, the registration hook refused the observer, or the target is a private function.
`SELF_TARGET`	The target is the observed function.
`LIMIT_EXCEEDED`	The worker has `observers.max_per_connection` observers (1024 by default), or the function has `observers.max_per_function` observers (4096 by default).
`OBSERVER_ID_CONFLICT`	An observer with the same id and different options exists, or another worker’s observer holds the id. A restarted worker under the same name and namespace replaces its own `suspended` observer with this id.
`OWNER_TIMEOUT`	The worker that hosts the function did not accept the observer in 30 seconds.
`OBSERVE_REJECTED`	The hosting worker’s `onObserverAdded` threw an error other than `ObserveRejection`.

A worker can also reject an observer with its own code, which its documentation lists (for example HTTP_ROUTE_CONFLICT or UNKNOWN_DB).

Observed call size

The engine keeps one copy of each observed call for all observers of the function. The copy has each field that any of these observers captures or filters on. If observers capture input and output, both count: a function that takes a 40 MiB input and returns a 40 MiB result makes an observed call of about 80 MiB. The engine does not deliver an observed call in these cases:

The fields the engine keeps come to more than 64 MiB (approximate in-memory size).
The activation is larger than the target worker accepts in one message: 64 MiB, or 16 MiB for browser workers and some older SDKs. See Payload size.
For calls that run in the function’s own worker (local dispatch), that SDK’s report of the call is larger than its 8 MiB report buffer.

Each such call counts as a TOO_LARGE drop in the iii.observer.dropped.total metric. In the first two cases it also counts in the observer’s dropped counts (engine::observers::list), and the engine logs a warning naming the observer, the function, the size, and the limit, at most once a minute per observer. To keep observed calls small, capture only what the target reads, for example capture: ["input"]. For payloads larger than these limits, pass a channel reference in place of the data.

OVERFLOW is a different drop: the queues were full when the call arrived. A call of the same size can go through when the queues have space again.

Caller and causality

Every invocation carries a caller: the worker that made the call and, when the call was made from inside a handler, the function and invocation it was made from.

{
  "worker_id": "3f1c...",
  "worker_name": "billing",
  "function_id": "billing::close-invoice",
  "invocation_id": "c3f...",
  "parent": { "worker_name": "api", "function_id": "api::checkout", "invocation_id": "77..." }
}

The engine sets worker_id and worker_name from the calling connection. It replaces any values that the SDK sends. The calling SDK sends function_id, invocation_id, and parent (the caller of that handler, one level only). Use these 3 fields for routing and filters only. A filter such as { "caller.parent.worker_name": { "$ne": "search-indexer" } } stops an indexer from reacting to its own writes. When the engine activates a target, the target’s caller is the engine (worker_name: "iii-engine", no worker_id), with the observed function and invocation as function_id and invocation_id. The current invocation’s activation holds the observer_id and activation_id. The SDK propagates the caller automatically from the handler you are in, and exposes the current invocation (function_id, invocation_id, caller, activation, observe_depth) as worker.currentInvocation(), carried across await.

Loop guard

An observer’s target can call the observed function again, directly or through other observers. This makes a loop. To stop loops, every invocation carries an observe_depth, which increases by 1 on each activation. Activations deeper than observers.max_observe_depth (8 by default) are dropped and counted as LOOP_GUARD, and an observer whose target is the observed function itself is rejected at creation with SELF_TARGET.

Workers on SDKs older than 0.25 do not send the depth. For their calls, the engine uses the largest depth of the invocations still in flight on that worker.

Local dispatch

When a worker calls a function that it registered, the SDK runs the handler in the same process. The call does not go through the engine. If the function has no observers, the SDK sends nothing to the engine. If the function has observers, the SDK applies their filters and sends the matching calls to the engine in batches. Observers get local calls and engine calls in the same way.

The hosting SDK reports the observed values of local calls to the engine. Values JSON cannot hold, such as a number no double can represent, arrive as null, and a lone surrogate in a string or key arrives as U+FFFD. Filters read the values the same way. Reports are at most once: a report lost with a dropped connection is not sent again.

A call runs locally when all of these hold:

This worker registered the function with a handler. HTTP-invoked functions always go through the engine. Local dispatch is on for the function.
The engine confirmed local dispatch for the registration. This requires an engine at 0.25 or later with observers.local_dispatch on. If the worker connects through RBAC, its session must have no middleware_function_id (unless the worker hosts the function privately), and the session must be allowed to call the function.
The call resolves to the namespace the function is registered in.
The action is Void or a normal call. An Enqueue call always goes through the engine.
The call does not set dispatch: "remote".
The call is fewer than 64 local calls deep. Deeper calls go through the engine.

A local call returns the same result, error shape, and timeout as a remote one. The differences are in Upgrading from 0.24.x. To keep a function on the engine path:

Node / TypeScript
Python
Rust
Go

// per function
worker.registerFunction("reports::build", buildReport, { localDispatch: false });

// per call
await worker.call({ function_id: "reports::build", payload, dispatch: "remote" });

# per function
worker.register_function("reports::build", build_report, local_dispatch=False)

# per call
worker.call({"function_id": "reports::build", "payload": payload, "dispatch": "remote"})

use iii_sdk::protocol::CallRequest;
use iii_sdk::{Dispatch, RegisterFunction};

// per function
worker.register_function(
    "reports::build",
    RegisterFunction::new_async(build_report).local_dispatch(false),
);

// per call
worker
    .call(CallRequest::new("reports::build", payload).dispatch(Dispatch::Remote))
    .await?;

// per function
off := false
worker.RegisterFunction("reports::build", buildReport, iii.FunctionOptions{LocalDispatch: &off})

// per call
worker.Call(ctx, iii.CallRequest{FunctionID: "reports::build", Payload: payload, Dispatch: iii.DispatchRemote})

To turn it off for the whole engine, set observers.local_dispatch: false in the engine configuration.

Authorization

If your worker connects with RBAC, it can observe a function only when its grants (expose_functions, namespaces) cover both the observed function and the target. Otherwise observe fails with FORBIDDEN. A private function is observable by any session whose grants cover it. When the observed function or the target registers again later, the engine does the check again. To change or deny observers in the RBAC configuration, see Worker manager.

Compatibility

With an engine older than 0.25, observe fails with UNSUPPORTED.
With a 0.25 engine, observers work. A function registered with args_format, private: true, or observable: false, or that sets onObservationMatch or onObservationDeliver, needs a 0.26 engine: the SDK does not register it on a 0.25 engine, so its observers stay pending. The event functions of workers such as http and queue are such functions.
Functions hosted by workers on older SDKs are observable. The engine evaluates the filters of every routed call, whichever SDK hosts the function, and the hosting SDK reports its own local calls.
Workers on SDKs older than 0.25 cannot observe.
on and deliver in an observer’s source are deprecated. A 0.26 engine translates them to their 0.25 behavior and logs a warning, and an observer from a worker on a 0.25 SDK or a 0.26 release candidate SDK always gets the 0.25 semantics. Replace them with capture: see Legacy on and deliver. Support will be removed in a future release.
A 0.26 SDK connected to a 0.25 engine or a 0.26 release candidate engine sends each observer with the legacy on its capture implies, and logs once that this is deprecated. Upgrade the engine. See the compatibility matrix.

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Observers

Observe a function

What can be observed

Defaults

Args and filters

Args

Filters

Operators

Comparison

Existence and `null`

Strings

Arrays

Logical

Validation

What the target receives

The payload

`capture`

Failed calls with their input

Metadata

The target function’s return value

Policy and metadata

Observer handle

Observer states

Engine-hosted event sources

Chain reactions

Inspect observers

Errors

Observed call size

Caller and causality

Loop guard

Local dispatch

Authorization

Compatibility

​Observe a function

​What can be observed

​Defaults

​Args and filters

​Args

​Filters

​Operators

Comparison

Existence and null

Strings

Arrays

Logical

​Validation

​What the target receives

​The payload

​capture

​Failed calls with their input

​Metadata

​The target function’s return value

​Policy and metadata

​Observer handle

​Observer states

​Engine-hosted event sources

​Chain reactions

​Inspect observers

​Errors

​Observed call size

​Caller and causality

​Loop guard

​Local dispatch

​Authorization

​Compatibility

Observe a function

What can be observed

Defaults

Args and filters

Args

Filters

Operators

Existence and `null`

Validation

What the target receives

The payload

`capture`

Failed calls with their input

Metadata

The target function’s return value

Policy and metadata

Observer handle

Observer states

Engine-hosted event sources

Chain reactions

Inspect observers

Errors

Observed call size

Caller and causality

Loop guard

Local dispatch

Authorization

Compatibility