---
question: How does one job start the next without missing a step?
kind: guide
prereq: You need a NATS project (the default of `abi new project`) and a remote module (`abi new module`). The steps use [Schedule a job](/capabilities/build/orchestrations/schedule-a-job) and [React to an event](/capabilities/build/orchestrations/react-to-an-event).
next:
  - /capabilities/build/orchestrations/react-to-an-event
  - /capabilities/build/pipelines/monitor-runs
---

# Chain jobs with events

A chain is several jobs, each started by what the previous one did. Nothing orchestrates them from above: a job writes something, the write is an event, and the event starts the next job. Each step stays a job of its own, with its own retries, timeout and run history, and a step can be re-run alone.

This page builds a three-step chain in a module called `reports`:

1. `export_report` runs each night and writes a CSV report to object storage. Object storage emits `ObjectPut`.
2. `load_report` starts on that `ObjectPut`, reads the file and loads it. It then publishes an event of its own, `ReportLoaded`.
3. `send_notice` starts on `ReportLoaded` and sends a notice by email.

![A scheduled export job writes a CSV to object storage. Object storage emits ObjectPut. The load job's filter keeps the reports/ puts; the event wakes the job, which reads what it has not handled from its durable cursor in the event log, then publishes ReportLoaded, which wakes the notice job.](/diagrams/abi-job-chain.svg)

## Declare the module

The module names the engine services it calls. Undeclared ones are refused.

:::note In-engine module?
The code on this page is for a remote module (`naas_abi_sdk`). An in-engine module declares its jobs the same way, from `naas_abi_core.module.jobs`, but calls the engine's core services, which differ: `ModuleDependencies` lists service classes, not names; the calls are synchronous (no `await`); the event service is `self.engine.services.events`; and its events, and the `event_class` that `query_for_consumer` takes, are typed `LogProcess` classes, not IRI strings. See [Module jobs](/capabilities/build/modules/module-jobs) and [Events](/capabilities/build/platform/events-overview).
:::

```python title="src/reports/__init__.py"
import csv
import io
from datetime import UTC, datetime, timedelta

from naas_abi_sdk import BaseModule, ModuleDependencies
from naas_abi_sdk.jobs import Cron, Every, JobContext, OnEvent, job
from naas_abi_sdk.services.event import Event

OBJECT_PUT = "http://ontology.naas.ai/abi/object_storage/ObjectPut"
REPORT_LOADED = "https://example.com/ontology/ReportLoaded"
REPORTS = {"prefix": {"prefix": "reports/"}, "key": {"suffix": ".csv"}}


class ABIModule(BaseModule):
    module_id = "reports"
    dependencies = ModuleDependencies(services=("object_storage", "event", "email"))
```

## Write on a schedule

The first job needs no event: it runs on a cron, or by hand with a `day` in its payload to redo one day.

```python title="src/reports/__init__.py"
    @job(triggers=(Cron("@daily"),), max_attempts=3)
    async def export_report(self, ctx: JobContext) -> dict:
        """Write yesterday's report to reports/daily/<day>.csv."""
        day = ctx.payload.get("day") or str(datetime.now(UTC).date() - timedelta(days=1))
        rows = await self.build_report(day)  # your code
        out = io.StringIO()
        csv.writer(out).writerows(rows)
        await self.engine.services.object_storage.put_object(
            "reports/daily", f"{day}.csv", out.getvalue().encode()
        )
        return {"day": day, "rows": len(rows)}
```

`put_object` stores the file, and the engine's object storage publishes `ObjectPut` with the `prefix`, `key` and `size_bytes` of the write. That event is the link to the next step. The key is the day, so a second run for the same day overwrites the file instead of adding one.

## Wake up, then read the log

The second job could read the object straight from the event, as on [React to an event](/capabilities/build/orchestrations/react-to-an-event). But `OnEvent` is **at most once**: an `ObjectPut` published while `load_report` is not running (a deploy, a crash, a restart) never starts it, and the report would wait until someone notices.

The event log keeps every event. So the event only wakes the job, and the job reads what it has not handled yet from the log, through a durable cursor:

```python title="src/reports/__init__.py"
    @job(
        triggers=(OnEvent(event_type=OBJECT_PUT, filter=REPORTS), Every("15m")),
        max_concurrency=1,
        timeout=timedelta(minutes=10),
    )
    async def load_report(self, ctx: JobContext) -> dict:
        """Load CSV reports written under reports/ (ObjectPut)."""
        events = self.engine.services.event
        puts = await events.query_for_consumer(
            "reports.load_report", OBJECT_PUT, limit=20, filter=REPORTS
        )
        if not puts:
            ctx.skip("no new report")
            return {}
        storage = self.engine.services.object_storage
        for put in puts:
            content = await storage.get_object(put["prefix"], put["key"])
            rows = await self.load_rows(put["key"], content)  # your code: an upsert keyed by the file
            await events.publish(
                Event(event_type=REPORT_LOADED, payload={"key": put["key"], "rows": rows})
            )
            ctx.log(f"loaded {put['key']}: {rows} rows")
        return {"loaded": [put["key"] for put in puts]}
```

- **`query_for_consumer`** returns the events of that type the consumer `reports.load_report` has not read yet, and moves its cursor past them in the same step. The cursor lives in the event log, not in the job: it survives restarts and redeploys.
- **A new cursor starts at the beginning.** The first run of a new consumer id reads the whole log for that event type: deployed into a project that already has reports, `load_report` loads every past `reports/` put, 20 at a time, and `send_notice` emails about each one. To start from now instead, call `await events.seek_consumer_to_end("reports.load_report", OBJECT_PUT)` once, before the first deploy (from a script, or a job you run by hand), not at every start: that would skip what the job missed while it was down.
- **The filter** is the same dict as the trigger's. The cursor is per consumer id and event type, not per filter, and moves past events the filter skips: keep one filter per consumer id for good.
- **`Every("15m")`** is the safety net. A report written while the module was down starts no run; the next tick finds it in the log. Without it, that report waits for the next `ObjectPut`.
- **`ctx.skip`** ends a run that found nothing as `SKIPPED`. Nexus hides those by default and keeps them for an hour.

The cursor moves when the events are read, before they are loaded. A run that crashes in the middle of a batch does not see that batch again on a retry. Keep `limit` small, and see [Make each step safe to run twice](#make-each-step-safe-to-run-twice) for work that must not be lost.

## Bursts of events

Ten reports written in a second publish ten `ObjectPut` events, and each becomes a trigger. With `max_concurrency=1` those runs never overlap. The first run takes whatever is in the log when it reads: all ten, or only the first few, since each put is appended and broadcast on its own. Later runs take the rest, or find nothing and end `SKIPPED`. Each report is loaded by one run, in order, whatever the number of events.

Without the cursor, ten triggers would be ten runs, each loading one file, and up to `max_concurrency` of them at once.

## Publish an event of your own

Step 2 announces its result with `ReportLoaded`, a type the module makes up: any IRI under a domain you control. `Event(event_type=..., payload={...})` from `naas_abi_sdk.services.event` writes it to the event log and broadcasts it, like the kernel's own events; `payload` becomes its fields, next to `_class_uri`, `_uri` and `created_at`.

The third job listens to that type. Filters work on your fields as on the kernel's:

```python title="src/reports/__init__.py"
    @job(triggers=(OnEvent(event_type=REPORT_LOADED, filter={"rows": {"gt": 0}}),), max_attempts=3)
    async def send_notice(self, ctx: JobContext) -> None:
        """Email the team when a non-empty report is loaded (ReportLoaded)."""
        loaded = ctx.payload["data"]
        await self.engine.services.email.send(
            to_email="team@example.com",
            from_email="reports@example.com",
            subject=f"Report {loaded['key']} loaded",
            text_body=f"{loaded['rows']} rows loaded.",
        )
```

The notice can afford to be at most once: a lost notice is a missing email, not a missing report. If it cannot, give it the same wake-up and cursor pattern, with `query_for_consumer("reports.send_notice", REPORT_LOADED)`.

An event decouples the steps: the job that publishes does not know who listens, and a fourth job (an archive, a second notice) can listen too without a change to step 2. Listeners in other modules work the same way.

## Or start the next job directly

When a step has exactly one follower, it can start it by name instead of publishing an event.

```python
# A job of the same module.
await self.atrigger_job("send_notice", {"key": key, "rows": rows}, idempotency_key=f"notice-{key}")

# A job of another module, listed in ModuleDependencies(modules=("notices",)).
notice = await self.engine.modules["notices"].get_job("send_notice")
await notice.trigger({"key": key, "rows": rows}, idempotency_key=f"notice-{key}")
```

- **Same module:** `atrigger_job` stores the trigger in JetStream at once. If the follower is busy or the module restarts, the trigger waits, up to seven days.
- **Another module:** `get_job` first asks discovery for a ready replica of `notices`. With none (a deploy, a crash), it raises `RPCError` with code `MODULE_UNAVAILABLE` and stores nothing. In step 2 the cursor has already moved past the report, so a retry of the run does not read it again: the notice is lost. For a follower in another module, publish an event instead, and let `notices` read it with its own cursor, as in the previous section.

The handler then reads `ctx.payload["key"]`, not `ctx.payload["data"]`. See [Run a job by hand](/capabilities/build/orchestrations/run-a-job-by-hand?via=python).

## Idempotency keys

A direct trigger can carry an `idempotency_key`. Two triggers of the same job with the same key **within two minutes** are one run: the second stores nothing and returns the first run, with `run.duplicate` set. Build the key from the work (`notice-reports/daily/2026-10-24.csv`), so a retried step 2 does not start step 3 twice.

## Make each step safe to run twice

Every run is **at least once**: after a crash or a lost acknowledgement, a handler can start again with the same payload. Retries (`max_attempts`) repeat it on purpose. Design each step so a second run changes nothing:

- **Write to a key, not to the end.** Step 1 writes `reports/daily/<day>.csv`; a second run overwrites the same object. Step 2 loads each row under a stable id (an upsert keyed by the file and the row), so loading twice adds no row.
- **Derive ids from the input.** Name outputs after the day, the file or the event, never after the time of the run or a random id.
- **Check before a side effect.** An email cannot be unsent. Before sending, record the notice under its key (the key-value store's compare-and-set, or a document) and skip when it is already there.
- **Keep steps small.** One file per item in a batch, one notice per report: a retry then repeats little.

Run records, retries, timeouts and retention are on [Module jobs](/capabilities/build/modules/module-jobs#write-handlers-that-can-run-twice). The event log and its cursors are on [Events](/capabilities/build/platform/events-overview).