> ## Documentation Index
> Fetch the complete documentation index at: https://iii.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Upgrading from 0.24.x to 0.25.x

> Nothing breaks: triggers keep working. Review the behavior changes, then move registerTrigger to worker.observe and registerTriggerType to observable functions where the source allows it.

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](#unregistertriggertype-now-takes-effect).

Read the [behavior changes](#behavior-changes) first, then move to the new APIs at your own pace.

> [!NOTE]
>
> 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](#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](#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](#keep-old-consumers-working).

* **One runtime warning per process per API.** The first call logs, through the SDK's logger (each
  SDK names its own API):

  ```text theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
  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")]`.

> [!WARNING]
>
> 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, `BaseException`s 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](https://iii.dev/docs/0-25-0/using-iii/configuration.md#observers) | | | |

### `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](#keep-old-consumers-working) 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 `match`es 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](https://iii.dev/docs/0-25-0/using-iii/observers.md#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` |

> [tab] Node / TypeScript
>
> **Before (0.24)**
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.registerTrigger({
>   type: "engine::functions-available",
>   function_id: "catalog::refresh",
>   config: {},
> });
> ```
>
> **After (0.25)**
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> const refresh = worker.observe(
>   { function_id: "engine::on-functions-changed", deliver: "input" },
>   "catalog::refresh",
> );
> await refresh.ready();
> ```

> [tab] Python
>
> **Before (0.24)**
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.register_trigger({
>     "type": "engine::functions-available",
>     "function_id": "catalog::refresh",
>     "config": {},
> })
> ```
>
> **After (0.25)**
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> refresh = worker.observe(
>     {"function_id": "engine::on-functions-changed", "deliver": "input"},
>     "catalog::refresh",
> )
> refresh.ready()  # or: await refresh.ready_async()
> ```

> [tab] Rust
>
> **Before (0.24)**
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::protocol::RegisterTriggerInput;
> use serde_json::json;
>
> worker.register_trigger(RegisterTriggerInput::new(
>     "engine::functions-available",
>     "catalog::refresh",
>     json!({}),
> ))?;
> ```
>
> **After (0.25)**
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Deliver, Source};
>
> let refresh = worker.observe(
>     Source::function("engine::on-functions-changed").deliver(Deliver::Input),
>     "catalog::refresh",
> );
> refresh.ready().await?;
> ```

> [tab] Go
>
> **Before (0.24)**
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterTrigger("catalog-refresh", "engine::functions-available", "catalog::refresh", nil)
> ```
>
> **After (0.25)**
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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.

> [tab] Node / TypeScript
>
> **Before (0.24)**
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> const trigger = worker.registerTrigger({
>   type: "timer",
>   function_id: "reports::refresh",
>   config: { every_ms: 60_000 },
>   metadata: { team: "reports" },
> });
> // later
> trigger.unregister();
> ```
>
> **After (0.25)**
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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();
> ```

> [tab] Python
>
> **Before (0.24)**
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> trigger = worker.register_trigger({
>     "type": "timer",
>     "function_id": "reports::refresh",
>     "config": {"every_ms": 60000},
>     "metadata": {"team": "reports"},
> })
> # later
> trigger.unregister()
> ```
>
> **After (0.25)**
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> observer = worker.observe(
>     {"function_id": "timer::on-tick", "args": {"every_ms": 60000}, "deliver": "input"},
>     "reports::refresh",
>     {"metadata": {"team": "reports"}},
> )
> # later
> observer.dispose()
> ```

> [tab] Rust
>
> **Before (0.24)**
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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)**
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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?;
> ```

> [tab] Go
>
> **Before (0.24)**
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterTrigger("reports-refresh", "timer", "reports::refresh",
>     json.RawMessage(`{"every_ms":60000}`), json.RawMessage(`{"team":"reports"}`))
> ```
>
> **After (0.25)**
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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](https://iii.dev/docs/0-25-0/reference/filters.md). 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.

> [!NOTE]
>
> 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](https://iii.dev/docs/0-25-0/using-iii/observers.md#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.

1. Pick a function id for the event, by convention `<worker>::on-<event>`.
2. 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`.
3. Move `registerTrigger(config)` to `onObserverAdded(obs)` and `unregisterTrigger(config)` to
   `onObserverRemoved(obs)`. Read the config from `obs.args`. Before, it was `config.config`.
4. 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).
5. Declare `legacy_trigger_type` with the old type id, and stop registering the trigger type, so old
   consumers keep working ([below](#keep-old-consumers-working)).

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

> [tab] Node / TypeScript
>
> **Before (0.24)**
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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)**
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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);
>       },
>     },
>   },
> );
> ```

> [tab] Python
>
> **Before (0.24)**
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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)**
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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())
> ```

> [tab] Rust
>
> **Before (0.24)**
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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)**
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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()),
>         ),
> );
> ```

> [tab] Go
>
> **Before (0.24)**
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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)**
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> 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](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md#example-a-file-watcher-that-emits)
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](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md#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](https://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](https://iii.dev/docs/next/upgrading/from-0-25-x.md).

What you can observe in 0.25: any regular function, the
[engine-hosted event sources](https://iii.dev/docs/0-25-0/using-iii/observers.md#engine-hosted-event-sources), and providers
that already moved to [observable functions](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md).

## Migration checklist

* [ ] Upgrade the engine and the SDKs; nothing else is required to keep running
* [ ] Check functions that call themselves for [local dispatch](#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](#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`.

> [!NOTE]
>
> For the observer model, see [Observers](https://iii.dev/docs/0-25-0/using-iii/observers.md). For the provider side, see
> [Observable functions](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md). For filter syntax, see the
> [filter language reference](https://iii.dev/docs/0-25-0/reference/filters.md).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.