Observers

Every function in iii is observable. An observer watches a function (the source) and invokes another function (the target) for each invocation or event that passes its filters. Any worker can observe any function it is allowed to invoke, and the function’s worker needs no code for it.

iii.observe(source, target, options?)

source is what is observed: the function_id, its args (only for functions whose owner routes events, see Auto and owner delivery), declarative filters, and what to capture.
target is the id of a registered function. There is no inline handler: register the target first.
options holds the observer’s id, its delivery policy, and metadata.

iii.observe replaces registerTrigger for sources that are functions. See Upgrading from 0.24.x to move existing triggers and trigger types.

The http, cron, queue, state, stream, pubsub, and database workers still publish trigger types in 0.25. Keep binding to them with registerTrigger until they ship source functions; the deprecation warning it logs is informational. You can observe any regular function, the engine-hosted event sources, and providers that already moved to observable functions.

Observe a function

The example below sends large orders to an alert function. orders::create is a regular function registered by another worker; it needs no changes to be observed. The target receives one Observed envelope per matching invocation.

Node / TypeScript
Python
Rust
Go

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

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

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

// The target: a registered function, like any other.
worker.registerFunction("alerts::big-order", async (event: Observed<Order>) => {
  console.log(`order ${event.input?.id} totals ${event.input?.total}`);
  return null;
});

const bigOrders = worker.observe(
  { function_id: "orders::create", filters: { "input.total": { $gte: 500 } } },
  "alerts::big-order",
);
await bigOrders.ready(); // resolves once the observer is active

import os
from iii import InitOptions, register_worker

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

# The target: a registered function, like any other.
def big_order(event: dict) -> None:
    order = event.get("input", {})
    print(f"order {order.get('id')} totals {order.get('total')}")

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

big_orders = worker.observe(
    {"function_id": "orders::create", "filters": {"input.total": {"$gte": 500}}},
    "alerts::big-order",
)
big_orders.ready()  # blocks until active; use `await big_orders.ready_async()` in asyncio code

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

#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
struct Order {
    id: 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 registered function, like any other.
worker.register_function(
    "alerts::big-order",
    RegisterFunction::new(|event: Observed<Order>| {
        if let Some(order) = &event.input {
            println!("order {} totals {}", order.id, order.total);
        }
        Ok::<_, Error>(())
    }),
);

let big_orders = worker.observe(
    Source::function("orders::create").filters(json!({ "input.total": { "$gte": 500 } })),
    "alerts::big-order",
);
big_orders.ready().await?; // resolves once the observer is active

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

// The target: a registered function, like any other.
client.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
    }
    log.Printf("big order: %s", event.Input)
    return nil, nil
})

bigOrders, err := client.Observe(iii.Source{
    FunctionID: "orders::create",
    Filters:    iii.Filters{"input.total": iii.Op{"$gte": 500}},
}, "alerts::big-order")
if err != nil {
    log.Fatal(err) // the request could not be built (missing ids, args that are not JSON)
}
if err := bigOrders.Ready(ctx); err != nil {
    log.Fatal(err) // an *iii.ObserveRejectionError carries the engine's code
}

observe returns a handle at once, also before the connection is up. The SDK sends the observer when it connects and replays it with the same id after every reconnect, so you call it once at startup, like registerFunction. The browser SDK (iii-browser-sdk) has the same observe API as Node.

The source

Field	Default	Meaning
`function_id`	required	The observed function. Every function is observable unless it opted out (`NOT_OBSERVABLE`).
`namespace`	the worker’s namespace	The namespace of the observed function. `engine::` and `configuration::` resolve to `default`, as for a `trigger` call.
`args`	none	Owner-delivered functions only: what the function’s owner should set up for this observer. Validated by the engine.
`filters`	`{}` (every event)	Which events reach the target. See Filters.
`on`	`success`	Auto delivery only: `success`, `error`, or `complete` (both).
`capture`	`["input", "output", "caller"]`	What the envelope carries: any of `input`, `output`, `error`, `caller`, `metadata`. `[]` sends ids, status, and timings only.
`deliver`	`envelope`	What the target’s first argument is: the `envelope`, or the observed `input` or `output` alone.

A filter may read any field, whatever the observer captures: capture only decides what travels to the target.

Auto and owner delivery

How events reach observers depends on the observed function. Its owner chooses when it registers the function; observers use the same observe call in both cases.

	Auto (the default)	Owner
An event is	every invocation of the function	what the function’s owner hands to an observer
`args`	not accepted (`INVALID_ARGS`)	validated against the function’s `args_format`, then read by the owner
Filters evaluated by	the engine (routed calls) or the hosting SDK (local calls)	the owner’s SDK, each time it hands an event to the observer
When the target runs	after the observed invocation completes, per `on`	when the owner decides, usually inside the function’s own handler
The target’s result	discarded	discarded for `obs.emit`; returned to the owner for `obs.call`
Examples	any regular function, the engine-hosted event sources	sources that need setup per observer: a schedule, a route, a table to watch

Owner-delivered functions document their args. For example, timer::on-tick, the example function you write in Observable functions (it is not built in), takes { every_ms }:

worker.observe({ function_id: "timer::on-tick", args: { every_ms: 60_000 } }, "reports::refresh");

To write such a function yourself, see Creating Workers / Observable functions.

What the target receives

With the default deliver: 'envelope', the target’s first argument is the Observed envelope. Fields left out of capture are absent.

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

With deliver: 'input' or deliver: 'output', the target receives that value alone, which suits a target written as a regular function of the observed payload. An observer of an owner-delivered function that the owner reaches with obs.call receives the owner’s payload instead (for example an HTTP request), and its return value goes back to the owner. The observer’s metadata option arrives as the activation’s metadata sidecar, the same second argument (metadata in Python, MetadataFromContext(ctx) in Go) that trigger metadata used. The target’s return value is discarded for auto delivery and obs.emit.

For local calls and obs.emit, the hosting SDK reports the observed values 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.

Event size

The engine keeps one copy of an observed invocation for all observers of its function, with every field any of them captures or filters on. Capturing input and output counts both: a function that echoes a 40 MiB input makes an event of about 80 MiB. An event too large on its own is not delivered, however idle the engine is:

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 local calls and obs.emit, the hosting SDK’s report of the event is larger than its 8 MiB report buffer.

Each such event 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. An owner’s obs.call whose activation the target cannot receive fails with payload_too_large. OVERFLOW is different: the queues were full when the event arrived, and an event of the same size can get through once they drain. To keep events small, capture only what the target reads, for example capture: ["input"]. With deliver: "input" or deliver: "output", set a matching capture too: captured fields are kept even when the target receives one value. For payloads larger than these limits, pass a channel reference instead of the data.

Filters

Filters are declarative JSON objects over the event: input, output, error, caller, and metadata. Each path takes an operator object; there is no literal shorthand.

{
  "input.total": { "$gte": 500 },
  "input.currency": { "$in": ["EUR", "USD"] },
  "caller.worker_name": { "$ne": "importer" }
}

The engine validates a filter when the observer is created (INVALID_FILTER) and checks its input.* and output.* paths against the function’s request_format and response_format (INVALID_FILTER_PATH for a path a closed schema does not have). The same evaluator runs in the engine and in every SDK, so a filter matches the same events wherever it runs. See the filter language reference for the operators, the rules for absent and null values, and the limits.

Policy and metadata

The third argument configures the observer itself:

Option	Default	Meaning
`id`	a random UUID	The observer id. Replays with the same id are idempotent; a different spec is `OBSERVER_ID_CONFLICT`.
`max_fires`	unlimited	The observer is disposed after this many activations (`MAX_FIRES`).
`expires_in_ms`	never	The observer is disposed after this long (`EXPIRED`).
`max_in_flight`	`observers.default_max_in_flight` (64)	Activations running at once. Further ones wait in a queue of `observers.max_queue_per_observer` (1024); beyond it, or once the queues of all observers hold 64 MiB of events, they are dropped (`OVERFLOW`). With `1`, activations run in the order they arrive.
`metadata`	none	JSON delivered as the metadata sidecar of every activation.

Node / TypeScript
Python
Rust
Go

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

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

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

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

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

Delivery is at most once. An observer failure never affects the observed invocation, and every dropped event is counted per function and reason in the iii.observer.dropped.total metric. An event waiting in several observers’ queues is held once, however many observers it matched.

Observer states

State	Meaning
`pending`	Waiting: no worker hosts the function yet (`OWNER_UNAVAILABLE`), the target is not registered yet (`TARGET_UNAVAILABLE`), or the owner has not acknowledged the observer.
`active`	Delivering.
`suspended`	The observing worker disconnected (`OBSERVER_DISCONNECTED`), or the target went away after the observer was active (`TARGET_UNAVAILABLE`).
`error`	Rejected. The reason carries the code (`INVALID_ARGS`, `FORBIDDEN`, an owner’s own code, …).
`disposed`	Gone for good: `dispose()`, `max_fires`, expiry, or the observing worker did not come back in time.

ready() resolves when the observer is active and rejects with the reason’s code when it turns error or disposed first (ObserveRejection in Node, Python, and Rust; *iii.ObserveRejectionError in Go).
onStateChange(listener) (on_state_change in Python and Rust, OnStateChange in Go) reports every transition with its reason. Listeners never run on the connection’s read path, so they may call the client.
dispose() removes the observer. A disposed observer is not replayed.
An observer belongs to the connection that created it. When that worker disconnects, its observers are suspended for observers.observer_grace_ms (30 s by default); a worker that reconnects in time keeps them, with their counters. After that they are disposed, and a worker that reconnects later gets that answer when its SDK replays them, so max_fires and expires_in_ms still hold. The engine keeps the answer for observers.disposed_retention_ms (1 hour by default). After that, a returning worker’s replay creates its observers again, as after an engine restart.
When the observed function’s worker disconnects, its observers go back to pending and become active again when the function is registered 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..." }
}

worker_id and worker_name are stamped by the engine from the calling connection and cannot be forged. function_id, invocation_id, and parent (the caller of that handler, one level only) come from the calling SDK, so use them for routing and filtering, not for authorization. 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 (auto delivery and obs.emit), 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. For obs.call, the caller is the owner’s worker and handler. In both cases the current invocation’s activation holds the observer_id and activation_id. The SDKs propagate the caller automatically from the handler you are in. Each SDK also exposes the current invocation (function_id, invocation_id, caller, activation, observe_depth):

SDK	Current invocation	Notes
Node	`worker.currentInvocation()`	Carried across `await` with `AsyncLocalStorage`.
Browser	`worker.currentInvocation()`	Known until the handler’s first `await`. Read it first and wrap later calls in `worker.withInvocation(inv, fn)`.
Python	`worker.current_invocation()`	A `contextvars` context, carried into sync handlers’ threads.
Rust	`iii_sdk::current_invocation()`	A task-local. Work moved to `tokio::spawn` loses it; wrap the future in `iii_sdk::scope(ctx, fut)`.
Go	`iii.InvocationFromContext(ctx)`	Pass the handler’s `ctx` to `client.Trigger` so outgoing calls carry the caller.

In the browser, calls made after an await outside withInvocation carry the depth of the deepest invocation running on the page, so with several invocations running the depth is over-counted: an observer chain may stop early at the loop guard, and some concurrent local calls go through the engine (with the same results). While the page is offline, functions it hosts still run locally, up to 256 nested local calls; past that a call fails at once with LOCAL_DEPTH_EXCEEDED.

Loop guard

An observer whose target calls the observed function again (directly or through other observers) would loop forever. Every invocation carries an observe_depth, which grows by one 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 propagate the depth; through them the engine estimates it from their activations still in flight.

Local dispatch

A worker that calls a function it registered itself runs the handler in-process, without a round trip through the engine. When nobody observes the function, such a call puts nothing on the wire. When observers match, the SDK evaluates their filters itself and reports the matching invocations to the engine in batches, so observers see local and routed calls alike. A call runs locally when all of these hold:

The function was registered by this worker with a handler (not an HTTP-invoked function), and local dispatch was not turned off for it.
The engine confirmed the registration for local dispatch (a 0.25 engine, observers.local_dispatch on, and a session without RBAC middleware_function_id that may invoke the function).
The call resolves to the namespace the function is registered in.
The action is not Enqueue (a Void call runs locally, fire-and-forget).
The call does not ask for the remote path.

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.trigger({ function_id: "reports::build", payload, dispatch: "remote" });

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

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

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

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

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

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

// per call
client.Trigger(ctx, iii.TriggerRequest{FunctionID: "reports::build", Data: payload, Dispatch: iii.DispatchRemote})

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

Engine-hosted event sources

The engine hosts event sources as auto-delivered functions in the default namespace. Observe them like any function, with filters on the event. They replace the engine’s own trigger types, which keep firing in 0.25.

Function	Replaces the trigger type	Event (`input`)
`engine::on-functions-changed`	`engine::functions-available`	`{ event: "functions_changed", functions: [{ function_id, namespace, worker_name, description?, metadata? }] }`. Changes within about 100 ms are coalesced into one event.
`engine::on-workers-changed`	`engine::workers-available`	`{ event, worker_id }`, `event` one of `worker_connected`, `worker_disconnected`, `worker_metadata_updated`
`configuration::on-updated`	`configuration`	`{ type: "configuration", event_type, id, name, description, schema, old_value, new_value, metadata? }`. `event_type` is `configuration:registered`, `configuration:updated`, or `configuration:deleted`.
`observability::on-log`	`log`	One log record: `severity_text`, `severity_number`, `body`, `attributes`, `service_name`, `trace_id`, `span_id`, …
`observability::on-span`	`trace`	`{ trace_ids: [...] }`: the traces that recorded spans in one window of about 300 ms.

These events have no output, so use deliver: 'input' to hand the event itself to the target. Their request_format is closed at the top level, so a misspelled input.* filter path fails with INVALID_FILTER_PATH. Calling one of them directly fails with FORBIDDEN: their events are the engine’s to emit.

Node / TypeScript
Python
Rust
Go

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

// Alert on error logs.
worker.observe(
  {
    function_id: "observability::on-log",
    deliver: "input",
    filters: { "input.severity_text": { $eq: "ERROR" } },
  },
  "alerts::log",
);

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

worker.observe(
    {
        "function_id": "configuration::on-updated",
        "deliver": "input",
        "filters": {
            "input.id": {"$eq": "iii-stream"},
            "input.event_type": {"$eq": "configuration:updated"},
        },
    },
    "stream::reload",
)

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

worker.observe(
    Source::function("configuration::on-updated")
        .deliver(Deliver::Input)
        .filters(json!({
            "input.id": { "$eq": "iii-stream" },
            "input.event_type": { "$eq": "configuration:updated" },
        })),
    "stream::reload",
);

client.Observe(iii.Source{
    FunctionID: "configuration::on-updated",
    Deliver:    iii.DeliverInput,
    Filters: iii.Filters{
        "input.id":         iii.Op{"$eq": "iii-stream"},
        "input.event_type": iii.Op{"$eq": "configuration:updated"},
    },
}, "stream::reload")

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, delivery, args, normalized filters, target, observing worker, state, reason, fires, 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 never includes an observer’s metadata. Under RBAC, a session sees only the observers of functions it may invoke.

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

Errors

An observer that cannot be created turns error, and ready() rejects with one of these codes:

Code	When
`UNSUPPORTED`	The engine is older than 0.25 (raised by the SDK), or `on`, `capture`, or `deliver` has an unknown value.
`INVALID_ARGS`	`args` fail the function’s `args_format`, or the function takes none (every auto function).
`INVALID_FILTER`	The filter is malformed, uses a literal instead of an operator object, or exceeds a limit.
`INVALID_FILTER_PATH`	A filter path is not in the function’s closed `request_format` or `response_format`.
`NOT_OBSERVABLE`	The function opted out (`observe.allow: false`).
`FORBIDDEN`	RBAC: the session may not invoke the observed function or the target, or the registration hook denied it.
`SELF_TARGET`	The target is the observed function.
`LIMIT_EXCEEDED`	`observers.max_per_connection` or `observers.max_per_function` is reached.
`OBSERVER_ID_CONFLICT`	An observer with the same id but a different spec exists.
`OWNER_TIMEOUT`	The function’s owner did not acknowledge the observer in time (30 s by default).
`OBSERVE_REJECTED`	The owner’s `onObserverAdded` hook threw a plain error. An owner can also reject with its own code.

Authorization

To observe a function, a session must be allowed to invoke both the observed function and the target (RBAC expose_functions). The check is repeated when the target registers again on another connection. The RBAC block of a worker-manager listener accepts an on_observer_registration_function_id hook, with the same contract as on_trigger_registration_function_id: it may rewrite function_id, target_function_id, args, filters, and metadata, or deny the observer (FORBIDDEN). See Worker manager.

Compatibility

Against an engine older than 0.25, observe fails with UNSUPPORTED and nothing runs locally.
Functions hosted by workers on older SDKs are still observable: their invocations are routed through the engine, which evaluates the filters.
Workers on older SDKs cannot observe. They keep using registerTrigger.

Getting Started

Upgrading

Using iii

Understanding iii

Creating Workers

Observers

Observe a function

The source

Auto and owner delivery

What the target receives

Event size

Filters

Policy and metadata

Observer states

Caller and causality

Loop guard

Local dispatch

Engine-hosted event sources

Inspect observers

Errors

Authorization

Compatibility

​Observe a function

​The source

​Auto and owner delivery

​What the target receives

​Event size

​Filters

​Policy and metadata

​Observer states

​Caller and causality

​Loop guard

​Local dispatch

​Engine-hosted event sources

​Inspect observers

​Errors

​Authorization

​Compatibility

Observe a function

The source

Auto and owner delivery

What the target receives

Event size

Filters

Policy and metadata

Observer states

Caller and causality

Loop guard

Local dispatch

Engine-hosted event sources

Inspect observers

Errors

Authorization

Compatibility