Upgrading from 0.24.x to 0.25.x

0.25 makes every function observable. A trigger type becomes a function that knows who observes it, and a trigger becomes an observer created with worker.observe(source, target, options?). The old trigger APIs keep working in 0.25, so most projects upgrade without code changes. The exceptions:

Rust builds that treat warnings as errors (-D warnings): put #[allow(deprecated)] on the calls you keep.
Code that calls unregisterTriggerType: engines before 0.25 ignored this call. The 0.25 engine removes the trigger type. If your code expects the type to stay, remove the call. See unregisterTriggerType now takes effect.

Read the behavior changes first, then move to the new APIs at your own pace.

The http, cron, queue, state, stream, pubsub, and database workers still publish trigger types in 0.25. Keep using registerTrigger for them; see Providers not migrated yet.

What changes for existing workers

The old APIs keep working. 0.24 SDKs run against a 0.25 engine unchanged (except that an unregisterTriggerType now takes effect), and their functions are observable. 0.25 SDKs run against a 0.24 engine too: observe fails with UNSUPPORTED, observe hooks never run, and calls are not dispatched locally.
registerTrigger and registerTriggerType are deprecated in every SDK and will be removed in a future release. They keep working through the trigger type’s provider or, once the provider moved to a function, through the engine’s legacy shim.
One runtime warning per process per API. The first call logs, through the SDK’s logger (each SDK names its own API):
registerTrigger is deprecated and will be removed in a future release. Use worker.observe(source, target) instead. Migration guide: https://iii.dev/docs/upgrading/from-0-24-x
registerTriggerType logs the equivalent message pointing to registerFunction with an observe descriptor. The engine logs deprecated registrations at debug only.
Language markers. TypeScript marks the APIs @deprecated, Python emits a DeprecationWarning at the call site, Go marks them Deprecated: (reported by linters such as staticcheck), and Rust marks the methods #[deprecated(since = "0.25.0")].

Rust crates built with #![deny(warnings)] or -D warnings fail to compile on calls to register_trigger, register_trigger_type, or unregister_trigger_type. Put #[allow(deprecated)] on the calls you keep for now (for example bindings to the http or cron worker), or migrate them.

Behavior changes

Local dispatch

A worker that calls a function it registered itself now runs the handler in-process after the engine confirms the registration. The call does not go through the engine. Results, error shapes, and timeouts are the same. The differences:

No message size limit on that path, because nothing is serialized. Through the engine, a message can be at most 64 MiB (16 MiB for browsers and SDKs older than 0.25), and a larger one fails with payload_too_large. A local call never fails with this error. If a test expects payload_too_large on a call to a function in the same worker, send the call with dispatch: "remote".
No engine-side span or call metrics for the local call. The SDK still records its own execute span, and reports the number of local invocations every 10 seconds (localinvocationstats), which the engine adds to iii.invocations.total with dispatch="local".
Value semantics. The handler receives a copy of the payload and the caller a copy of the result, as over the wire.
Rust: a local call runs on the caller’s task, and a sync handler (RegisterFunction::new) runs inline on it, where the call’s timeout cannot interrupt it. For a handler that can block, use an async handler (RegisterFunction::new_async) or set local_dispatch(false) on it.
Recursion: a local call nested more than 64 local calls deep goes through the engine. In the browser while offline, local calls run up to 256 levels deep, and a deeper one fails at once with LOCAL_DEPTH_EXCEEDED.
Python: a sync handler called locally runs on a thread of its own and honors the caller’s timeout. Whatever a local handler raises, BaseExceptions such as KeyboardInterrupt and SystemExit included, reaches the caller as an InvocationError with code invocation_failed and the original exception as its __cause__.
Disconnects: once confirmed, local dispatch keeps serving calls to the worker’s own functions while the connection is down.

Local dispatch is off for a session with an RBAC middleware_function_id, or one that may not invoke the function, because a local call would bypass those checks. Enqueue calls always go through the engine. To keep the old behavior for a function, a call, or the whole engine:

Scope	Node / browser	Python	Rust	Go
One function	`{ localDispatch: false }`	`local_dispatch=False`	`RegisterFunction::new(f).local_dispatch(false)`	`RegisterFunctionOptions{LocalDispatch: &off}`
One call	`dispatch: "remote"`	`"dispatch": "remote"`	`TriggerRequest::new(id, payload).dispatch(Dispatch::Remote)`	`TriggerRequest{Dispatch: iii.DispatchRemote}`
Whole engine	`observers.local_dispatch: false` in the engine config

`unregisterTriggerType` now takes effect

Every SDK has always sent unregistertriggertype, but engines before 0.25 ignored it. The 0.25 engine removes the trigger type when the sending worker is its provider (and ignores it otherwise). Its triggers then move to another provider of the type, are taken by a function that declares it as legacy_trigger_type, or wait until a provider registers again. If your worker calls unregisterTriggerType and expected the type to stay, remove the call.

Smaller changes

Legacy shim results. For a registerTrigger that the legacy shim turns into an observer, the engine sends one TriggerRegistrationResult, also on success. The SDKs ignore the success result.
RegisterFunction frames omit request_format and response_format when they are absent. Before, they sent null.
Loop guard. Every invocation carries an observe depth. When the depth of an event is more than observers.max_observe_depth (8 by default), the engine drops it. Workers on 0.24 SDKs do not send the depth. For their calls, the engine uses the largest depth of the invocations still in flight on that worker.
Observe hooks that hang. An onObserverAdded hook still running after 30 seconds fails the observer with OWNER_TIMEOUT, and the next hook of the same function may start while it runs.
Rust SDK:
- public protocol structs gained fields and Message gained variants, so struct literals of them and exhaustive matches on Message need updating (prefer the constructors, such as RegisterTriggerInput::new and TriggerRequest::new);
- the connection state turns Reconnecting as soon as the link drops;
- a handler that panics now returns an invocation_failed error to the caller. Before, the call timed out;
- frames omit an invocation_id of None. Before, they sent null.
Browser SDK:
- trigger now sends metadata, traceparent, and baggage, and handlers receive the metadata as their second argument;
- the unregistertriggertype frame no longer includes description;
- while the page is offline, functions it hosts run locally up to 256 nested calls, and a deeper call fails with LOCAL_DEPTH_EXCEEDED.

Transition overview

0.24	0.25
`registerTrigger({ type, function_id, config, metadata })`	`observe({ function_id: <source function>, args, filters }, <target>, { metadata })`
trigger `type`	the source function, for example `engine::on-functions-changed`
trigger `config`	`args` (owner-delivered sources) or `filters` (auto-delivered sources)
`condition_function_id`	`filters` for conditions you can state over the event; a condition that needs code stays on `registerTrigger` or moves into the target
trigger `function_id`	the target (second argument)
trigger `metadata`	`options.metadata`, delivered as the activation’s metadata sidecar
`trigger.unregister()`	`handle.dispose()`
a `TriggerRegistrationResult` error	the handle’s `error` state; `ready()` rejects with the code
`registerTriggerType({ id, description }, handler)`	`registerFunction(id, handler, { description, observe: { delivery: "owner", args_format, legacy_trigger_type } })`
`trigger_request_format` (Python, Rust, Go)	`observe.args_format`
`call_request_format` (Python, Rust, Go)	the function’s `request_format`
`TriggerHandler.registerTrigger(config)`	`observe.onObserverAdded(obs)`
`TriggerHandler.unregisterTrigger(config)`	`observe.onObserverRemoved(obs)`
`config.id`, `config.function_id`, `config.config`	`obs.id`, `obs.target`, `obs.args` (`obs.legacy.config` for old consumers)
`config.metadata`	not visible to the owner: it stays with the engine and reaches the target
firing: `trigger({ function_id: binding.function_id, payload })` per binding	`obs.emit(event)` (fire-and-forget) or `obs.call(payload)` (request/response) per observer
`unregisterTriggerType(...)`	`functionRef.unregister()`
`TriggerTypeRef.registerTrigger` / `.registerFunction` helpers	`registerFunction` plus `observe`
Rust `TriggerHandler`, `RegisterTriggerType`, `RegisterTriggerInput`	`ObserveHandler`, `RegisterFunction::observe(Observe::owner()..)`, `Source::function(..)`
`engine::registered-triggers::list` (trigger instances)	`engine::observers::list`

Move `registerTrigger` to `worker.observe`

Do this for triggers whose source is a function: any regular function, the engine-hosted event sources, or a provider that moved to an observable function. With deliver: "input" the target receives the same payload its trigger handler received, so the handler stays as it is. Register the target before you observe it. This is the best practice in most cases, but it is not required. If the target is not registered yet, the observer stays pending (TARGET_UNAVAILABLE) until the target registers, and events that occur in that time do not reach it.

Engine triggers

The engine’s own trigger types have source functions in 0.25:

Trigger type	Observe
`engine::functions-available`	`engine::on-functions-changed`
`engine::workers-available`	`engine::on-workers-changed`
`configuration`	`configuration::on-updated`, filter on `input.id` and `input.event_type`
`log`	`observability::on-log`, filter on `input.severity_number`
`trace`	`observability::on-span`

Node / TypeScript
Python
Rust
Go

Before (0.24)

worker.registerTrigger({
  type: "engine::functions-available",
  function_id: "catalog::refresh",
  config: {},
});

After (0.25)

const refresh = worker.observe(
  { function_id: "engine::on-functions-changed", deliver: "input" },
  "catalog::refresh",
);
await refresh.ready();

Before (0.24)

worker.register_trigger({
    "type": "engine::functions-available",
    "function_id": "catalog::refresh",
    "config": {},
})

After (0.25)

refresh = worker.observe(
    {"function_id": "engine::on-functions-changed", "deliver": "input"},
    "catalog::refresh",
)
refresh.ready()  # or: await refresh.ready_async()

Before (0.24)

use iii_sdk::protocol::RegisterTriggerInput;
use serde_json::json;

worker.register_trigger(RegisterTriggerInput::new(
    "engine::functions-available",
    "catalog::refresh",
    json!({}),
))?;

After (0.25)

use iii_sdk::{Deliver, Source};

let refresh = worker.observe(
    Source::function("engine::on-functions-changed").deliver(Deliver::Input),
    "catalog::refresh",
);
refresh.ready().await?;

Before (0.24)

worker.RegisterTrigger("catalog-refresh", "engine::functions-available", "catalog::refresh", nil)

After (0.25)

refresh, err := worker.Observe(iii.Source{
    FunctionID: "engine::on-functions-changed",
    Deliver:    iii.DeliverInput,
}, "catalog::refresh")
if err == nil {
    err = refresh.Ready(ctx)
}

Triggers with config and metadata

For a provider that moved to an owner-delivered function, the trigger’s config becomes the observer’s args and its metadata the observer’s metadata. The example uses the timer provider from the next section.

Node / TypeScript
Python
Rust
Go

Before (0.24)

const trigger = worker.registerTrigger({
  type: "timer",
  function_id: "reports::refresh",
  config: { every_ms: 60_000 },
  metadata: { team: "reports" },
});
// later
trigger.unregister();

After (0.25)

const observer = worker.observe(
  { function_id: "timer::on-tick", args: { every_ms: 60_000 }, deliver: "input" },
  "reports::refresh",
  { metadata: { team: "reports" } },
);
// later
await observer.dispose();

Before (0.24)

trigger = worker.register_trigger({
    "type": "timer",
    "function_id": "reports::refresh",
    "config": {"every_ms": 60000},
    "metadata": {"team": "reports"},
})
# later
trigger.unregister()

After (0.25)

observer = worker.observe(
    {"function_id": "timer::on-tick", "args": {"every_ms": 60000}, "deliver": "input"},
    "reports::refresh",
    {"metadata": {"team": "reports"}},
)
# later
observer.dispose()

Before (0.24)

use iii_sdk::protocol::RegisterTriggerInput;
use serde_json::json;

let trigger = worker.register_trigger(
    RegisterTriggerInput::new("timer", "reports::refresh", json!({ "every_ms": 60_000 }))
        .with_metadata(json!({ "team": "reports" })),
)?;
// later
trigger.unregister();

After (0.25)

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

let observer = worker.observe_with(
    Source::function("timer::on-tick")
        .args(json!({ "every_ms": 60_000 }))
        .deliver(Deliver::Input),
    "reports::refresh",
    ObserverOptions::default().metadata(json!({ "team": "reports" })),
);
// later
observer.dispose().await?;

Before (0.24)

worker.RegisterTrigger("reports-refresh", "timer", "reports::refresh",
    json.RawMessage(`{"every_ms":60000}`), json.RawMessage(`{"team":"reports"}`))

After (0.25)

observer, err := worker.Observe(iii.Source{
    FunctionID: "timer::on-tick",
    Args:       iii.Args{"every_ms": 60000},
    Deliver:    iii.DeliverInput,
}, "reports::refresh", iii.ObserverOptions{Metadata: json.RawMessage(`{"team":"reports"}`)})
if err != nil {
    log.Fatal(err)
}
// later
observer.Dispose(ctx)

Filters cover conditions you can state over the event. For example, a condition_function_id that let only gold customers through becomes filters: { "input.customer_tier": { "$eq": "gold" } }, evaluated without an extra function call; see the filter language. A condition that needs code (a lookup, an auth check) has no filter form: keep that trigger on registerTrigger, which works in 0.25, or move the check into the target function.

Browser workers use the same observe call as Node. A browser handler knows its invocation only until its first await; wrap later calls in worker.withInvocation(inv, fn) so they carry the caller and the loop guard’s depth. While the page is offline, its own functions run locally up to 256 nested calls, then fail with LOCAL_DEPTH_EXCEEDED. See Caller and causality.

Move a trigger type to an observable function

A trigger type becomes a function with an observe descriptor. Its owner keeps the same job: set something up per observer, and hand each event to the observers it belongs to. The engine now stores the bindings, validates their args, applies their filters and policy, and replays them when your worker reconnects.

Pick a function id for the event, by convention <worker>::on-<event>.
Set delivery: "owner" and describe the config your trigger type accepted as observe.args_format. If the trigger type declared schemas (Python, Rust, Go), move trigger_request_format to observe.args_format and call_request_format to the function’s request_format.
Move registerTrigger(config) to onObserverAdded(obs) and unregisterTrigger(config) to onObserverRemoved(obs). Read the config from obs.args. Before, it was config.config.
Replace trigger({ function_id: binding.function_id, payload }) with obs.emit(event), or with obs.call(payload) when the result matters (an HTTP response, a queue acknowledgement).
Declare legacy_trigger_type with the old type id, and stop registering the trigger type, so old consumers keep working (below).

The example moves a timer trigger type, which fires each binding every every_ms milliseconds.

Node / TypeScript
Python
Rust
Go

Before (0.24)

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

const logger = new Logger();
const bindings = new Map<string, ReturnType<typeof setInterval>>(); // trigger id → its timer

worker.registerTriggerType<{ every_ms: number }>(
  { id: "timer", description: "Fires every every_ms milliseconds" },
  {
    async registerTrigger({ id, function_id, config }) {
      const handle = setInterval(() => {
        worker
          .trigger({ function_id, payload: { every_ms: config.every_ms, fired_at_ms: Date.now() } })
          .catch((err) => logger.error("timer trigger failed", err));
      }, config.every_ms);
      bindings.set(id, handle);
    },
    async unregisterTrigger({ id }) {
      clearInterval(bindings.get(id));
      bindings.delete(id);
    },
  },
);

After (0.25)

import { ObserveRejection, type Observer } from "iii-sdk";

type TimerArgs = { every_ms: number };
const timers = new Map<string, ReturnType<typeof setInterval>>(); // observer id → its timer

// An observer made from an old registerTrigger has its config in obs.legacy.
const argsOf = (obs: Observer) => (obs.legacy ? obs.legacy.config : obs.args) as TimerArgs;

worker.registerFunction(
  "timer::on-tick",
  async (tick: { every_ms: number; fired_at_ms: number }) => tick,
  {
    description: "Fires every `every_ms` milliseconds. Observe it with args { every_ms }.",
    observe: {
      delivery: "owner",
      args_format: {
        type: "object",
        additionalProperties: false,
        required: ["every_ms"],
        properties: { every_ms: { type: "integer", minimum: 100 } },
      },
      legacy_trigger_type: "timer",
      onObserverAdded(obs) {
        const { every_ms } = argsOf(obs);
        if (!Number.isInteger(every_ms) || every_ms < 100) {
          // args are checked by the engine, legacy configs are not
          throw new ObserveRejection("INVALID_TIMER_CONFIG", "every_ms must be an integer of at least 100");
        }
        const handle = setInterval(() => obs.emit({ every_ms, fired_at_ms: Date.now() }), every_ms);
        timers.set(obs.id, handle);
      },
      onObserverRemoved(obs) {
        clearInterval(timers.get(obs.id));
        timers.delete(obs.id);
      },
    },
  },
);

Before (0.24)

import threading
import time

from iii.trigger import TriggerConfig, TriggerHandler

bindings: dict[str, threading.Event] = {}  # trigger id → stop flag

class TimerHandler(TriggerHandler):
    async def register_trigger(self, config: TriggerConfig) -> None:
        every_ms = config.config["every_ms"]
        stop = bindings[config.id] = threading.Event()

        def run() -> None:
            while not stop.wait(every_ms / 1000):
                worker.trigger({
                    "function_id": config.function_id,
                    "payload": {"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)},
                })

        threading.Thread(target=run, daemon=True).start()

    async def unregister_trigger(self, config: TriggerConfig) -> None:
        stop = bindings.pop(config.id, None)
        if stop:
            stop.set()

worker.register_trigger_type(
    {"id": "timer", "description": "Fires every every_ms milliseconds"},
    TimerHandler(),
)

After (0.25)

import threading
import time

from iii import ObserveHooks, ObserveRejection

timers: dict[str, threading.Event] = {}  # observer id → stop flag

def args_of(obs) -> dict:
    # An observer made from an old register_trigger has its config in obs.legacy.
    return obs.legacy["config"] if obs.legacy else obs.args

class TimerHooks(ObserveHooks):
    delivery = "owner"
    args_format = {
        "type": "object",
        "additionalProperties": False,
        "required": ["every_ms"],
        "properties": {"every_ms": {"type": "integer", "minimum": 100}},
    }
    legacy_trigger_type = "timer"

    def on_observer_added(self, obs):
        every_ms = args_of(obs).get("every_ms")
        if not isinstance(every_ms, int) or every_ms < 100:
            # args are checked by the engine, legacy configs are not
            raise ObserveRejection("INVALID_TIMER_CONFIG", "every_ms must be an integer of at least 100")
        stop = timers[obs.id] = threading.Event()

        def run() -> None:
            while not stop.wait(every_ms / 1000):
                obs.emit({"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)})

        threading.Thread(target=run, daemon=True).start()

    def on_observer_removed(self, obs):
        stop = timers.pop(obs.id, None)
        if stop:
            stop.set()

def on_tick(tick: dict) -> dict:
    return tick

worker.register_function("timer::on-tick", on_tick, observe=TimerHooks())

Before (0.24)

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::protocol::TriggerRequest;
use iii_sdk::trigger::{TriggerConfig, TriggerHandler};
use iii_sdk::{Error, IIIClient, RegisterTriggerType};
use serde_json::json;

#[derive(Clone)]
struct TimerHandler {
    worker: IIIClient,
    tasks: Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>, // trigger id → its task
}

#[async_trait::async_trait]
impl TriggerHandler for TimerHandler {
    async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> {
        let every_ms = config.config["every_ms"].as_u64().unwrap_or(1_000);
        let (worker, function_id) = (self.worker.clone(), config.function_id.clone());
        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(every_ms));
            loop {
                interval.tick().await;
                let fired_at_ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
                let tick = json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms });
                let _ = worker.trigger(TriggerRequest::new(function_id.clone(), tick)).await;
            }
        });
        self.tasks.lock().unwrap().insert(config.id, task);
        Ok(())
    }

    async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> {
        if let Some(task) = self.tasks.lock().unwrap().remove(&config.id) {
            task.abort();
        }
        Ok(())
    }
}

worker.register_trigger_type(RegisterTriggerType::new(
    "timer",
    "Fires every every_ms milliseconds",
    TimerHandler { worker: worker.clone(), tasks: Default::default() },
));

After (0.25)

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::{Error, Observe, ObserveHandler, ObserveRejection, Observer, RegisterFunction};
use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
use serde::{Deserialize, Serialize};
use serde_json::json;

#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct TimerArgs {
    every_ms: u64,
}

#[derive(Serialize, Deserialize, JsonSchema)]
struct Tick {
    every_ms: u64,
    fired_at_ms: u64,
}

#[derive(Clone, Default)]
struct Timers(Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>); // observer id → its task

#[async_trait::async_trait]
impl ObserveHandler for Timers {
    async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
        // An observer made from an old register_trigger has its config in legacy().
        let TimerArgs { every_ms } = match obs.legacy() {
            Some(legacy) => serde_json::from_value(legacy.config.clone())
                .map_err(|e| ObserveRejection::new("INVALID_TIMER_CONFIG", e.to_string()))?,
            None => obs.args()?,
        };
        let observer = obs.clone();
        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_millis(every_ms));
            loop {
                interval.tick().await;
                let fired_at_ms = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
                let _ = observer.emit(&json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms }));
            }
        });
        self.0.lock().unwrap().insert(obs.id().to_string(), task);
        Ok(())
    }

    async fn on_observer_removed(&self, obs: &Observer) {
        if let Some(task) = self.0.lock().unwrap().remove(obs.id()) {
            task.abort();
        }
    }
}

worker.register_function(
    "timer::on-tick",
    RegisterFunction::new(|tick: Tick| Ok::<_, Error>(tick))
        .description("Fires every `every_ms` milliseconds. Observe it with args { every_ms }.")
        .observe(
            Observe::owner()
                .args_format::<TimerArgs>()
                .legacy_trigger_type("timer")
                .handler(Timers::default()),
        ),
);

Before (0.24)

type timerHandler struct {
    worker *iii.Client
    mu     sync.Mutex
    stops  map[string]chan struct{} // trigger id → stop
}

func (h *timerHandler) RegisterTrigger(ctx context.Context, cfg iii.TriggerConfig) error {
    var config struct {
        EveryMs int64 `json:"every_ms"`
    }
    if err := json.Unmarshal(cfg.Config, &config); err != nil {
        return err
    }
    stop := make(chan struct{})
    h.mu.Lock()
    h.stops[cfg.ID] = stop
    h.mu.Unlock()
    go func() {
        ticker := time.NewTicker(time.Duration(config.EveryMs) * time.Millisecond)
        defer ticker.Stop()
        for {
            select {
            case <-stop:
                return
            case now := <-ticker.C:
                data, _ := json.Marshal(map[string]int64{"every_ms": config.EveryMs, "fired_at_ms": now.UnixMilli()})
                h.worker.Trigger(context.Background(), iii.TriggerRequest{FunctionID: cfg.FunctionID, Data: data, Namespace: cfg.Namespace})
            }
        }
    }()
    return nil
}

func (h *timerHandler) UnregisterTrigger(ctx context.Context, cfg iii.TriggerConfig) error {
    h.mu.Lock()
    defer h.mu.Unlock()
    if stop, ok := h.stops[cfg.ID]; ok {
        close(stop)
        delete(h.stops, cfg.ID)
    }
    return nil
}

worker.RegisterTriggerType("timer", "Fires every every_ms milliseconds",
    &timerHandler{worker: worker, stops: map[string]chan struct{}{}})

After (0.25)

var (
    mu    sync.Mutex
    stops = map[string]chan struct{}{} // observer id → stop
)

func everyMs(obs *iii.Observer) (int64, error) {
    var args struct {
        EveryMs int64 `json:"every_ms"`
    }
    var err error
    if obs.Legacy != nil {
        err = json.Unmarshal(obs.Legacy.Config, &args) // an old RegisterTrigger: its config is the args
    } else {
        err = obs.DecodeArgs(&args)
    }
    if err == nil && args.EveryMs < 100 {
        // args are checked by the engine, legacy configs are not
        err = iii.ObserveRejection("INVALID_TIMER_CONFIG", "every_ms must be at least 100")
    }
    return args.EveryMs, err
}

func registerTimer(worker *iii.Client) error {
    return worker.RegisterFunction("timer::on-tick", func(ctx context.Context, data json.RawMessage) (any, error) {
        return data, nil
    }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
        Delivery:          iii.DeliveryOwner,
        ArgsFormat:        json.RawMessage(`{"type":"object","additionalProperties":false,"required":["every_ms"],"properties":{"every_ms":{"type":"integer","minimum":100}}}`),
        LegacyTriggerType: "timer",
        OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
            every, err := everyMs(obs)
            if err != nil {
                return err
            }
            stop := make(chan struct{})
            mu.Lock()
            stops[obs.ID] = stop
            mu.Unlock()
            go func() {
                ticker := time.NewTicker(time.Duration(every) * time.Millisecond)
                defer ticker.Stop()
                for {
                    select {
                    case <-stop:
                        return
                    case now := <-ticker.C:
                        obs.Emit(context.Background(), map[string]int64{"every_ms": every, "fired_at_ms": now.UnixMilli()})
                    }
                }
            }()
            return nil
        },
        OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
            mu.Lock()
            defer mu.Unlock()
            if stop, ok := stops[obs.ID]; ok {
                close(stop)
                delete(stops, obs.ID)
            }
        },
    }})
}

The “after” versions keep one timer per observer, like the original. To share one timer per interval between all its observers, index them by interval and emit from the function’s own handler, as the file watcher example does per path.

Keep old consumers working

legacy_trigger_type: "timer" lets workers that still call registerTrigger({ type: "timer" }) keep working with no change, as long as no worker registers the timer trigger type the old way:

The engine turns such a trigger into an observer of your function, owned by the worker that registered it. Its id is the trigger id, it has no args and no filters, its metadata is the trigger’s, and it uses deliver: "input", so the bound function receives the event you emit, exactly as it received the fired payload before.
Your onObserverAdded hook sees obs.legacy = { trigger_type, config }. The engine does not check the config against args_format, so validate it as your old TriggerHandler did. Accept or reject it like any observer; the old worker receives one TriggerRegistrationResult with the outcome.
unregisterTrigger disposes the observer.
A worker that registers the trigger type the old way always wins for new registrations. Remove the registerTriggerType call in the same release that adds the function.

See Accept old triggers for the lookup rules.

Providers not migrated yet

In their latest releases, the http, cron, queue (durable:subscriber), state, pubsub (subscribe), and database workers still register trigger types. The stream worker, which ships with the engine, does too. Some of these workers also ship a source function, starting with the release listed:

Worker	Source function	Since
http	`http::on-request`	0.22.8
cron	`cron::on-tick`	0.21.27
queue	`queue::on-message`	0.21.25
state	`state::on-changed`	0.22.22
database	`database::on-row-changed`	0.5.23

None of them declares legacy_trigger_type, and none will: 0.26 removes registerTrigger, trigger types, and the legacy shim. A 0.26 engine drops trigger registrations and logs a warning in its own log; the worker gets no error, and the bound function never runs. On 0.25:

your registerTrigger calls to these workers keep working; the deprecation warning is informational, and Rust code needs #[allow(deprecated)] on those calls;
their documentation, including the config shapes at workers.iii.dev, stays valid;
to move a binding to worker.observe, upgrade the worker to a release that ships its source function, and observe that function with the args its documentation lists (they are not the trigger config: an http route takes { method, path }, with a leading slash on path). Remove the registerTrigger call in the same deploy, so each route, schedule, or subscriber has one binding.

Move every binding before you upgrade the engine to 0.26, or move them all in that upgrade. The 0.26 engine, these workers’ 0.26 releases, and the workers that bind to them upgrade together; see Upgrading from 0.25.x. What you can observe in 0.25: any regular function, the engine-hosted event sources, and providers that already moved to observable functions.

Migration checklist

Upgrade the engine and the SDKs; nothing else is required to keep running
Check functions that call themselves for local dispatch differences (large payloads, blocking Rust sync handlers, tests that expect engine spans)
Remove unregisterTriggerType calls that expected the type to stay registered
Add #[allow(deprecated)] to Rust trigger calls you keep, if warnings are errors
Move triggers on engine events (engine::functions-available, engine::workers-available, configuration, log, trace) to worker.observe
Move your own trigger types to observable functions with legacy_trigger_type
Keep registerTrigger for the http, cron, queue, state, stream, pubsub, and database workers on 0.25, and move those bindings to worker.observe before or with the 0.26 upgrade (see Providers not migrated yet)

Result

The project runs on 0.25 with its triggers unchanged. Functions you observe deliver to their targets through the engine’s observers, calls a worker makes to its own functions run in-process, and trigger types you moved keep serving the workers that still use registerTrigger.

For the observer model, see Observers. For the provider side, see Observable functions. For filter syntax, see the filter language reference.

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Upgrading from 0.24.x to 0.25.x

What changes for existing workers

Behavior changes

Local dispatch

`unregisterTriggerType` now takes effect

Smaller changes

Transition overview

Move `registerTrigger` to `worker.observe`

Engine triggers

Triggers with config and metadata

Move a trigger type to an observable function

Keep old consumers working

Providers not migrated yet

Migration checklist

Result

​What changes for existing workers

​Behavior changes

​Local dispatch

​unregisterTriggerType now takes effect

​Smaller changes

​Transition overview

​Move registerTrigger to worker.observe

​Engine triggers

​Triggers with config and metadata

​Move a trigger type to an observable function

​Keep old consumers working

​Providers not migrated yet

​Migration checklist

​Result

What changes for existing workers

Behavior changes

Local dispatch

`unregisterTriggerType` now takes effect

Smaller changes

Transition overview

Move `registerTrigger` to `worker.observe`

Engine triggers

Triggers with config and metadata

Move a trigger type to an observable function

Keep old consumers working

Providers not migrated yet

Migration checklist

Result