Observers

observe runs a function (the target) each time another function (the source) has an event:

worker.observe(source, target)

By default, an event is a successful call of the source. Some sources make their own events, and their behavior may differ. In these cases, consult the documentation of the specific worker. Any worker can observe any function, unless otherwise restricted by RBAC or by the source itself (allow: false).

source is the function to observe, plus optional filters that select which events reach the target.
target is the id of the function to run. It can be any function except the source itself. The observer stays pending until the source and the target are registered.

worker.observe replaces registerTrigger for sources that are functions. See Upgrading from 0.24.x to move existing triggers.

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 examples on this page build on each other. They all react to orders::create, a regular function registered by another worker.

React to every call

After each successful call of orders::create, run emails::send-receipt with the order. deliver: "input" makes the target receive the input of orders::create (the order) as its argument.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

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

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

// The target: a regular function.
worker.registerFunction("emails::send-receipt", async (order: Order) => {
  await sendReceipt(order.email, order);
  return null;
});

// After each successful call of orders::create, run emails::send-receipt with the order.
const receipts = worker.observe(
  { function_id: "orders::create", deliver: "input" },
  "emails::send-receipt",
);
await receipts.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="emails"))

# The target: a regular function.
def send_receipt(order: dict) -> None:
    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.
receipts = worker.observe(
    {"function_id": "orders::create", "deliver": "input"},
    "emails::send-receipt",
)
receipts.ready()  # blocks until active; use `await receipts.ready_async()` in asyncio code

use iii_sdk::{Deliver, Error, InitOptions, RegisterFunction, Source, 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.
worker.register_function(
    "emails::send-receipt",
    RegisterFunction::new(|order: Order| {
        send_receipt(&order.email, &order);
        Ok::<_, Error>(())
    }),
);

// After each successful call of orders::create, run emails::send-receipt with the order.
let receipts = worker.observe(
    Source::function("orders::create").deliver(Deliver::Input),
    "emails::send-receipt",
);
receipts.ready().await?; // resolves once the observer is active

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

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

// The target: a regular function.
worker.RegisterFunction("emails::send-receipt", func(ctx context.Context, data json.RawMessage) (any, error) {
    var order Order
    if err := json.Unmarshal(data, &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.
receipts, err := worker.Observe(iii.Source{
    FunctionID: "orders::create",
    Deliver:    iii.DeliverInput,
}, "emails::send-receipt")
if err != nil {
    log.Fatal(err) // the request could not be built (missing ids, args that are not JSON)
}
if err := receipts.Ready(ctx); err != nil {
    log.Fatal(err) // an *iii.ObserveRejectionError carries the engine's code
}

From this point, emails::send-receipt runs after each successful call of orders::create. 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 observer can miss events that occur before ready() resolves.

Narrow it with a filter

This observer runs alerts::big-order only for orders with a total of 500 or more. A filter on the source selects which events reach the target. The source does not set deliver, so the target receives the Observed envelope. With the default capture, the envelope has the order (input), the result (output), the caller (caller), and the duration of the call (duration_ms).

Node / TypeScript
Python
Rust
Go

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

const logger = new Logger();

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();

from iii_helpers.observability import Logger

logger = Logger()

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_helpers::observability::Logger;
use iii_sdk::Observed;
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
    }
    log.Printf("big order: %s", 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)
}

Both observers watch the same function, independently. The next examples add policy, chains, and engine events.

Observe errors or all calls

By default, the target runs only after a successful call of the source. To change this, set on in the source:

`on`	The target runs after
`success` (default)	a successful call
`error`	a failed call
`complete`	each call, successful or failed

A cancelled call (for example, its caller stopped waiting) does not run the target, for any value of on. The engine counts it as a dropped event with the reason CANCELLED in the iii.observer.dropped.total metric. The default capture is ["input", "output", "caller"], without error. If capture does not include error, the envelope has no error field, and the target does not get the error. To give the target the error, add error to capture. This observer runs alerts::order-failed after each failed call of orders::create. The target receives the Observed envelope with the order (input) and the error (error):

Node / TypeScript
Python
Rust
Go

worker.observe(
  { function_id: "orders::create", on: "error", capture: ["input", "error"] },
  "alerts::order-failed",
);

worker.observe(
    {"function_id": "orders::create", "on": "error", "capture": ["input", "error"]},
    "alerts::order-failed",
)

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

worker.observe(
    Source::function("orders::create")
        .on(On::Error)
        .capture([Capture::Input, Capture::Error]),
    "alerts::order-failed",
);

worker.Observe(iii.Source{
    FunctionID: "orders::create",
    On:         iii.OnError,
    Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError},
}, "alerts::order-failed")

To run the target after each call, use on: "complete". The envelope’s status is success or error.

Target arguments

With the default deliver: "envelope", the target’s first argument is the Observed envelope. The envelope always has the ids, status, and the timings. It has input, output, error, caller, and metadata only when capture includes them. output is present only after a successful call.

Field	Meaning
`observer_id`, `activation_id`	The observer, and this activation of it.
`function_id`, `namespace`	The observed function.
`invocation_id`	The observed call. For an event that a source makes, see below.
`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.

When a source makes its own events, the invocation_id depends on where the source sends the event:

From the handler of the source function: the id of that call. Events sent during the same call have the same invocation_id.
From other code: a new id for each event; for example when a source calls another function that then performs the emit. The envelope’s caller then has the function_id and invocation_id of that function.

Engine-hosted event sources give each event a new id. With deliver: "input" or deliver: "output", the target’s first argument is that value alone. If the value is absent (for example, output after a failed call), the argument is null. The observer’s metadata option is the target’s second argument (the metadata parameter in Python, MetadataFromContext(ctx) in Go). It is different from the metadata field of the envelope, which is the metadata of the observed call. iii discards the target’s return value, unless the source documents otherwise. Observers of the target can see it as the output of the target’s call. See Chain reactions.

Source options

All fields are optional except function_id:

Field	Default	Meaning
`function_id`	required	The function to observe.
`namespace`	`default` for `engine::` and `configuration::` functions; the worker’s namespace for all others	The namespace of the function.
`filters`	`{}` (every event)	Which events reach the target. See Filters.
`on`	`success`	Which calls run the target: `success`, `error`, or `complete` (both). See Observe errors or all calls.
`capture`	`["input", "output", "caller"]`	Which fields the envelope has: any of `input`, `output`, `error`, `caller`, `metadata`. With `[]`, the envelope has only the ids, `status`, and the timings.
`deliver`	`envelope`	What the target’s first argument is: the `envelope`, or the observed `input` or `output` alone.
`args`	none	Arguments for the source, only for sources that document them. See Source `args`.

A filter can read any field, including those not defined in capture. capture sets only which fields the target function receives.

Source `args`

Some sources take args, for example a path for a file watcher or an interval for a schedule. The documentation of the source lists its args. For example, files::watch, the example source in Observable functions, takes { path }. It is not part of iii:

Node / TypeScript
Python
Rust
Go

// When /data/orders.csv changes, run csv::process.
worker.observe(
  { function_id: "files::watch", args: { path: "/data/orders.csv" }, deliver: "input" },
  "csv::process",
);

# When /data/orders.csv changes, run csv::process.
worker.observe(
    {"function_id": "files::watch", "args": {"path": "/data/orders.csv"}, "deliver": "input"},
    "csv::process",
)

// When /data/orders.csv changes, run csv::process.
worker.observe(
    Source::function("files::watch")
        .args(json!({ "path": "/data/orders.csv" }))
        .deliver(Deliver::Input),
    "csv::process",
);

// When /data/orders.csv changes, run csv::process.
worker.Observe(iii.Source{
    FunctionID: "files::watch",
    Args:       iii.Args{"path": "/data/orders.csv"},
    Deliver:    iii.DeliverInput,
}, "csv::process")

args tell the source which events to make (the changes to one file). Filters select from those events, for example only changes that leave the file non-empty: { "input.size_bytes": { "$gt": 0 } }. The engine checks args against the args_format of the source. The observer fails with INVALID_ARGS when the args do not match, or when the source has no args_format and the args are not empty.

Filters

A filter is a JSON object. Each key is a path in the event 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. The operators are $eq, $ne, $gt, $gte, $lt, $lte, $in, $nin, $exists, $prefix, $and, $or, $not, and $any.

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

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. The engine and each SDK have their own filter implementation. All of them pass the same set of test cases, so a filter matches the same events in all of them. See the filter language reference for the operators, the rules for absent and null values, and the limits.

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. In this example, the inventory worker reserves the items of each new order, and the shipping worker schedules a shipment for each reservation. Each worker registers its own function and observer.

Node / TypeScript
Python
Rust
Go

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

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

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

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

// 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").deliver(Deliver::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").deliver(Deliver::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", Deliver: iii.DeliverInput}, "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", Deliver: iii.DeliverOutput}, "shipping::schedule")

inventory::reserve returns the reservation. With deliver: "output", shipping::schedule receives that return value as its argument. 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 event. When the depth is more than observers.max_observe_depth (8 by default), the engine drops the event with LOOP_GUARD. See Loop guard.

Engine-hosted event sources

The engine hosts event sources as functions in the default namespace. Observe them like any function, with filters on the event. They replace the trigger types of the engine.

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.

No function runs for these events. The engine creates each event, puts it in input, and handles it as a successful call with no output. This has two results:

The calls never fail. An observer with on: "error" is accepted, but its target never runs.
There is no output. With deliver: "output", the target gets null. To give the target the event, use deliver: "input".

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 makes their events. A direct call to one of them fails with FORBIDDEN.

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_number": { $gte: 17 } }, // OpenTelemetry ERROR and above
  },
  "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",
);

worker.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")

Policy and metadata

observe takes an optional third argument, the 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. The same id with the same options again changes nothing. The same id with different options fails with `OBSERVER_ID_CONFLICT`.
`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 events, 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 Target arguments.

This observer runs alerts::payment-failed after each failed call of payments::charge with an amount of 500 or more. It runs at most 4 activations at the same time, and it gives each activation { "team": "billing" } as metadata:

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" })),
);

worker.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"}`),
})

Each event runs the target at most once. The engine does not retry. A failure of the target does not change the observed call. The engine counts each dropped event in the iii.observer.dropped.total metric, by function and reason.

Observer states

State	Meaning
`pending`	Waiting. No worker hosts the function yet (`OWNER_UNAVAILABLE`), the target is not registered yet (`TARGET_UNAVAILABLE`), or the source has not accepted the observer yet.
`active`	The target runs for each matching event.
`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 source.
`disposed`	Removed permanently, by `dispose()`, `max_fires`, `expires_in_ms`, or because the observing worker did not reconnect 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 each state change with its reason. Listeners run outside the connection’s read path, so a listener can call the worker.
dispose() (Dispose(ctx) in Go) removes the observer. A disposed observer is not replayed. 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 envelope. To stop after a number of events or after a time, set max_fires or expires_in_ms in the policy.
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.
When a worker reconnects, its SDK sends each of its observers to the engine again, with the same id. The engine remembers 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.
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 observed function’s worker disconnects, its observers go back to pending and become active again when the function is registered again.

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 does not include the observer’s metadata option. With RBAC, a session sees only the observers of functions that it is allowed to invoke.

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

Errors

When the engine cannot create an observer, the observer 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 `on`, `capture`, or `deliver` has a value that the engine does not know.
`INVALID_ARGS`	`args` do not match the `args_format` of the source, or the source 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`.
`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 refuses observers (`allow: false`).
`FORBIDDEN`	RBAC does not allow the session to invoke the observed function or the target, the registration hook refused the observer, or the target is an owner-delivered 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.
`OWNER_TIMEOUT`	The source’s worker did not accept the observer in 30 seconds.
`OBSERVE_REJECTED`	The source refused the observer. A source can also reject with its own code, documented by that source.

Event 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 returns its 40 MiB input makes an event of about 80 MiB. The engine does not deliver an event 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 source’s own worker (local dispatch), 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. OVERFLOW is different: the queues were full when the event arrived. An event of the same size can go through when the queues have space again. 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.

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 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 `worker.Trigger` so outgoing calls carry the caller.

In the browser, a call made after an await and outside withInvocation gets the depth of the deepest invocation that runs on the page. When several invocations run at the same time, this depth can be too high, and the loop guard can stop an observer chain early. While the page is offline, the functions that it hosts run locally, to a maximum of 256 nested calls. A call past that limit fails immediately with LOCAL_DEPTH_EXCEEDED.

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 invocations 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 a 0.25 engine with observers.local_dispatch on. If the worker connects through RBAC, its session must have no middleware_function_id, 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 ask for the remote path.
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.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
worker.RegisterFunction("reports::build", buildReport, iii.RegisterFunctionOptions{LocalDispatch: &off})

// per call
worker.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.

Create an observable function

To make something observable, register a function for it and call that function whenever the thing happens. Each call is an event for its observers:

// Producer: call this whenever a file changes.
worker.registerFunction("files::on-change", async (change: FileChanged) => change);

// ...in your watcher loop
await worker.trigger({ function_id: "files::on-change", payload: { path, event: "modified" } });

// Consumer, in any worker: when a file changes, process it.
worker.observe({ function_id: "files::on-change" }, "csv::process");

A function can also send different events to different observers. For example, a file watcher can watch only the paths that its observers give in args. To write this type of function, see Observable functions. You observe it in the same way, sometimes with args.

Authorization

If your worker connects with RBAC, it can observe a function only when it is allowed to call both the observed function and the target. Otherwise observe fails with FORBIDDEN. 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.
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

Using iii

Understanding iii

Creating Workers

Upgrading

Observers

Observe a function

React to every call

Narrow it with a filter

Observe errors or all calls

Target arguments

Source options

Source `args`

Filters

Chain reactions

Engine-hosted event sources

Policy and metadata

Observer states

Inspect observers

Errors

Event size

Caller and causality

Loop guard

Local dispatch

Create an observable function

Authorization

Compatibility

​Observe a function

​React to every call

​Narrow it with a filter

​Observe errors or all calls

​Target arguments

​Source options

​Source args

​Filters

​Chain reactions

​Engine-hosted event sources

​Policy and metadata

​Observer states

​Inspect observers

​Errors

​Event size

​Caller and causality

​Loop guard

​Local dispatch

​Create an observable function

​Authorization

​Compatibility

Observe a function

React to every call

Narrow it with a filter

Observe errors or all calls

Target arguments

Source options

Source `args`

Filters

Chain reactions

Engine-hosted event sources

Policy and metadata

Observer states

Inspect observers

Errors

Event size

Caller and causality

Loop guard

Local dispatch

Create an observable function

Authorization

Compatibility