--- question: How are module jobs scheduled? --- # Module jobs on JetStream message schedules ## Status Accepted. Step 1 (SDK modules) and step 2 (engine modules) implemented; the orchestration ports and Dagster removal follow. ## Date 2026-10-01 ## Context Scheduled and event-driven work runs in Dagster: one process imports every module through a full Engine, discovers definitions over its own gRPC code locations, and keeps its own run storage. With NATS, modules already register through discovery, expose work over request/reply, and persist runs in the document service. The 21 zen orchestrations only use ops, jobs, sensors as schedulers ("daily, skip if a run is in progress"), a few cron schedules and tag concurrency limits — not assets, partitions or IO managers. NATS 2.12+ JetStream ships a message scheduler (ADR-51): a message with a `Nats-Schedule` header (cron, `@every`, `@at`; `Nats-Schedule-Time-Zone` since 2.14) is republished by the stream to `Nats-Schedule-Target` on that schedule. ## Decision A module declares jobs and runs them in its own process; JetStream schedules and distributes the triggers. No central scheduler and no Dagster. API (SDK, mirrored by core modules in step 2): - `jobs = (JobDescriptor(name, description, triggers=(Cron(...), Every(...), OnEvent(...)), max_concurrency=1, timeout=..., max_attempts=1),)` and `self.expose_job(name, async_handler)`, or the `@job(...)` method decorator, which declares and binds in one place. Handlers receive a `JobContext` (`run_id`, `job`, `attempt`, `trigger`, `payload`, `log()`, `cancelled`). - `engine.modules[m].get_job(name).trigger(payload)` returns a `JobRun` handle (`status()`, `wait()`, `cancel()`). `ABIClient.get_job(module_id, name, project=...)` reaches any module's job by name, engine modules included. Transport: - One stream per discovery project, `ABI_JOBS_`, subjects `abi.jobs..>`, `allow_msg_schedules` and `allow_msg_ttl`, created idempotently by any job host (the bus convention). - Module and job names are hashed into subjects (module ids contain dots): triggers on `abi.jobs.

.trigger..`, schedules on `abi.jobs.

.schedule...`. On start, a host ensures one schedule message per trigger (rollup replaces it) and purges stale indexes, so schedules follow the code that is running. It publishes a schedule only when it is missing or its headers changed (see "Review hardening (2026-10-06)"). - One durable pull consumer per job, shared by replicas: one replica takes each trigger. `max_ack_pending = max_concurrency` (no overlap by default), `in_progress()` heartbeats while running, backoff on failure, `term()` once attempts are exhausted. Attempts are counted on the run record, not by `max_deliver`, which is unlimited (see "Review hardening (2026-10-06)"). - `OnEvent` bridges core-NATS event subjects (`evt..>`) into the trigger subject through a queue group, with `Nats-Msg-Id` = event id. Runs: one document per trigger message (`job_runs_` in the module namespace, key = job + stream sequence), create-only then CAS, holding status (`QUEUED`, `RUNNING`, `SUCCEEDED`, `FAILED`, `TIMED_OUT`, `CANCELLED`), attempts, trigger, payload, bounded logs, result and error. Cancellation is cooperative (`ctx.cancelled`) through a core-NATS cancel subject the running host listens on. Discovery: `ModuleDescriptor.jobs` (tag 6) lists job descriptors so callers and Nexus can see them; replicas must declare identical jobs. Engine modules (step 2, option a): core `BaseModule` has the same API (`naas_abi_core.module.jobs`; `JobsMixin` is shared with the SDK). Handlers may be sync: they run in a worker thread, where blocking engine service calls are safe. In NATS mode with `nats.jobs.enabled` (default true), `Engine.load` starts one SDK `JobHost` per module with jobs after `on_initialized`, on a dedicated event loop (never the loop serving the engine's NATS endpoints), using the `engine` service identity and the module id as document namespace. `shutdown()` stops the hosts first. Without NATS, declared jobs are logged and not run. Kernel jobs use the same host under a reserved owner id: dataset maintenance (`naas_abi_core.dataset`: `dataset_compaction` daily 02:00 UTC, `dataset_catalog_monitor` hourly) replaces the Dagster DatasetCompaction app, which Dagster now skips when its engine hosts jobs. It runs on the engine's own dataset service, not the NATS facade (2026-10-06): a facade call ended after the 10 s RPC default while compaction went on (see the RPC failure-semantics ADR). Agent memory (`naas_abi_core.agent_memory`: `agent_memory_migrate` on demand, `agent_memory_prune` daily 03:00 UTC) and the jobs a service adapter offers through `job_owners(services)` (`naas_abi_core.event_archive`, `naas_abi_core.activity_log`) are kernel owners too; the System app's Jobs tab lists every owner `Engine.job_owners` returns. Engine modules are not in discovery, so their jobs are reached by name (`ABIClient.get_job`). Engine modules later move into their own containers unchanged: same subjects, same consumers, same run records. ### Triggering a module's own jobs, run retention, event filters (2026-10-04) Added so modules can leave Dagster entirely: - **A module's own triggers.** Engine modules had no NATS client (the engine's job host owns the jobs connection), so they could not trigger their own jobs. The runtime now binds each module to its job host once it started (engine: `EngineJobLoader`; SDK: `run_module`), and `JobsMixin` exposes `self.trigger_job(name, payload=None, idempotency_key=None)` for sync code (an engine request thread; it blocks until the trigger is stored) and `await self.atrigger_job(...)` for async code. Both run on the host's loop and raise `JobsNotHosted` before the host starts or after it stops (outside NATS mode, always). Only the module's own declared jobs; other modules' jobs stay `ABIClient.get_job` / `ModuleProxy.get_job`. - **Idempotency.** `idempotency_key` (also on `JobProxy.trigger`) sets `Nats-Msg-Id: :`: within the stream's duplicate window (2 minutes) a repeated key is the same trigger and returns the same `JobRun`. ```python class ABIModule(BaseModule): @job(triggers=(Every("30s"),)) # polls stay as a safety net def runs(self, ctx: JobContext) -> None: if not self.service.has_queued_runs(): ctx.skip("queue empty") return self.service.execute_next() def request_run(self, record_id: str) -> None: # an API request thread self.service.queue(record_id) self.trigger_job("runs", {"record": record_id}, idempotency_key=record_id) ``` - **Skipped runs and retention.** `ctx.skip(reason)` records the run as `SKIPPED` (terminal, acked, never an error) with `skip_reason`. Job hosts prune finished run records (`JobRetention`, engine config `nats.jobs.retention`): skipped runs after 1 hour, other finished runs after 7 days, and at most 1,000 finished runs per job, oldest first; at most 200 deletions per pass, every 10 minutes and once at start. QUEUED, RUNNING and RETRYING runs are never touched; replicas pruning together just skip what another deleted. The Nexus System Jobs tab hides SKIPPED runs unless asked (a "Skipped" filter; `include_skipped`), keeps them out of health and failures, and shows when a job last had nothing to do. - **Event filters.** `OnEvent(..., filter={...})` keeps irrelevant events from triggering a job: an EventBridge-style dict on the event's fields (dotted paths; scalar, list, or `eq`/`ne`/`in`/`gt`/`gte`/`lt`/`lte`/`prefix`/ `suffix`/`contains`/`exists`), checked by the event bridge before it publishes a trigger, so a filtered event writes no run at all. It is validated when declared and travels with the descriptor (`JobTrigger.filter`, JSON). `naas_abi_sdk/event_filter.py` mirrors the event service's matcher (the SDK stays core-free). Example: `OnEvent(event_type=OBJECT_PUT, filter={"prefix": {"prefix": "naas/"}})`. ## Consequences - Modules own their job code and run it where they run; adding or removing a module needs no orchestrator restart. - Delivery is at-least-once: a handler must tolerate a repeat after a crash or a missed ack. Missed cron ticks fire once on broker recovery, not per tick; DST transitions can skip or repeat a cron tick (ADR-51). - `OnEvent` is at-most-once from event to trigger: events published while no host runs are lost (events are not on a stream yet). SDK `event.append` RPCs do not broadcast and cannot trigger jobs. - The broker must be NATS >= 2.14 for time zones; `allow_msg_schedules` cannot be turned off on a stream once enabled. - Every process that loads the engine in NATS mode hosts jobs (the consumers are shared, so each trigger still runs once). Set `nats.jobs.enabled: false` on one-off engines (CLI) that should not take work. - Stopping engine sync handlers: each run gets its own thread. On timeout or cancel, `ctx.cancelled` is set; after `nats.jobs.interrupt_grace_seconds` (default 5) `JobInterrupted` (a `BaseException`) is raised in that thread. The run keeps its concurrency slot and in-progress message until the thread exits, so a timed-out run never overlaps the next one. The interrupt fires at the next Python bytecode: a blocking C call (a socket read without a timeout) delays it, and it can land inside library cleanup code, so handlers should still check `ctx.cancelled` and set I/O timeouts. Only process isolation (the module's own container) can kill code stuck in C. - `close()` (shutdown) is not a user cancel: interrupted runs are recorded as failed ("Interrupted by host shutdown") and nak'ed, so they are retried while attempts remain. With `max_attempts=1` a run cut by a crash or shutdown is not run again. After a crash nothing finishes its record either, so the hosts' upkeep (every 10 minutes, next to retention) fails a RUNNING run on its last attempt once its heartbeats (`heartbeat_at`, every 10 s) have stopped for 5 minutes (`reap_lost_runs`: "Lost: no heartbeat since ..."). A run with attempts left is left to its redelivery, unless its trigger expired (see "Review hardening (2026-10-06)"). - Run records are bounded by `JobRetention`; the Nexus System app's Jobs tab reads them (runs, failures, skipped runs on demand). ## Review hardening (2026-10-04) - Event bridges deduplicate only when the source supplies `Nats-Msg-Id`. The derived ID includes the target module/job subject and source subject; raw events without a stable ID omit deduplication. Manual idempotency keys are also scoped to the target module/job. Identical payloads can be distinct events. - Delivery acknowledgement renewal runs independently of log persistence. Progress writes log failures and retry on the next heartbeat interval; even a blocked write cannot stop renewal. Renewal failure stops the handler and records a failure/retry rather than silently continuing execution. - Completion is saved before ack, term, or nak. Transient completion-write failures retry at the heartbeat interval, capped at one second, retaining the result and renewing delivery while the host remains active. Shutdown or lost renewal leaves an unsuccessful write unacknowledged. CAS exhaustion raises rather than reporting a successful save. - Persisted progress heartbeats continue until completion is durable, so another replica's lost-run reaper cannot mistake a delayed completion write for a crashed host. Progress updates apply only to RUNNING records, checking again after CAS conflicts; they cannot modify a completed record or recreate one removed by retention. - Redelivery of a saved terminal run retires delivery without executing its handler or replacing its result. This recovers a lost acknowledgement; it is still not exactly-once execution across a crash before completion is saved. ## Review hardening (2026-10-06) Fixes from the PR #1299 review (items 1, 2, 10, 17 and 18). - **Schedules survive restarts.** NATS restarts an `@every` interval when its schedule message is replaced, so republishing every schedule on start meant an `Every("1h")` job never fired if its module restarted, redeployed or scaled more often than hourly. A host now reads the current schedule (`get_last_msg`) and publishes only when it is missing or one of `Nats-Schedule`, `Nats-Schedule-Time-Zone`, `Nats-Schedule-Target`, `Nats-Schedule-TTL` or `Abi-Job-Trigger` differs. The publish carries `Nats-Expected-Last-Subject-Sequence` set to the sequence it read (0 when missing), so of two hosts starting together only one writes; the other reads again and finds it current. Stale indexes are still purged. - **Attempts are counted on the run, not by the broker.** The consumer's `max_deliver` is unlimited. The run record's `attempt` counts handler starts: a delivery for a run with attempts left starts attempt `attempt + 1`, and a delivery after the last attempt is terminated without running (the upkeep fails that run once its heartbeats stop, as before). When a delivery cannot read or record its run (document store or claim check unreachable, CAS exhausted), the host logs an error and nak's it with backoff. That uses no attempt, so a store outage when a trigger arrives no longer drops it, even with `max_attempts=1`. Retries stop when the trigger expires (`Nats-TTL`: at most an hour for schedule ticks, 24 hours for events, 7 days for manual triggers). Existing consumers are updated in place on start. - **Expired triggers end their run.** JetStream never redelivers a trigger whose `Nats-TTL` has passed. Before a nak, the host checks whether the trigger (stored at `fired_at`, living `Nats-TTL`) expires before the backoff ends. If so, it records FAILED ("Trigger expired before it could be retried; ") and terminates the delivery. A RETRYING run also records `retry_at`. The upkeep fails a RETRYING run past `retry_at`, or a RUNNING run with attempts left and no heartbeat for 5 minutes, once its trigger is no longer in the stream (`get_msg` on the sequence stored as `sequence`, or taken from the run id; another job's message at that sequence counts as gone). Runs on the reaping host are never touched. - **A recreated jobs stream does not skip new triggers.** Run ids are `:`, and sequences restart at 1 when `ABI_JOBS_` is recreated while run records survive. A record now belongs to a delivery only when its `fired_at` equals the trigger's stream timestamp; otherwise the delivery is a new trigger and replaces the record. A record without `fired_at` (written before the field existed) counts as a match, so completed work is not run again after the upgrade. - **A run that cannot save its completion holds its slot.** `_save_completion` keeps renewing the delivery and retries until the store takes the result, or until shutdown or a failed renewal. This is deliberate: acking before the result is saved could lose it, and nak'ing would run finished work again. The cost is that the run keeps its concurrency slot for the whole outage; with `max_concurrency=1` the job is blocked until the store recovers. To make that visible, retries log a warning and, once the outage lasts `completion_error_after_seconds` (60 s by default), an error naming the run. - **A rollout's consumers exist before it serves.** A process with a rollout id starts its host only after the cutover (discovery ADR, "Descriptors across generations"), and a consumer delivers only triggers stored after it was created (`DeliverPolicy.NEW`). A trigger for a job only the new rollout declares, sent in between, was never delivered. `JobHost.prepare` now runs before the instance can serve: it creates each missing consumer with the same configuration `start` uses, and does nothing else. It does not fetch, publish or purge schedules, bridge events, or update an existing consumer, so the serving generation keeps its limits until `start`. Triggers stored meanwhile wait for the host. If the rollout never cuts over, its new consumers stay, like those of removed jobs: their triggers expire with their `Nats-TTL`, or run once a later generation declares the job.