---
question: How does one job start the next without missing a step?
kind: guide
prereq: You need ABI 3.x, a NATS project (the default of `abi new project` on 3.x) and a module listed under `modules:` in `config.yaml`. The steps use [Schedule a job](/v2/capabilities/build/orchestrations/schedule-a-job) and [React to an event](/v2/capabilities/build/orchestrations/react-to-an-event).
next:
  - /v2/capabilities/build/orchestrations/react-to-an-event
  - /v2/capabilities/build/pipelines/monitor-runs
---

> **You are reading the v2.x docs (engine modules). New projects use v3.** [Read this page in v3](/capabilities/build/orchestrations/chain-jobs-with-events) · [Migrate from 2.x to 3.0](/capabilities/upgrade/migrate-to-3)

# 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.

Jobs are ABI 3.x with NATS. On ABI 2.x, and on 3.x without NATS, chain steps with Dagster jobs and sensors in a module orchestration: see [Dagster](/v2/capabilities/build/applications/dagster).

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, as service classes. Undeclared ones are refused. The engine's services are synchronous, so the handlers on this page are plain methods: each runs in a worker thread.

`ReportLoaded` is an event type of the module's own, a `LogProcess` class with `key` and `rows` fields, declared in an ontology and generated as on [Record an event](/v2/capabilities/build/platform/record-an-event).

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

from naas_abi_core.module.jobs import Cron, Every, JobContext, OnEvent, job
from naas_abi_core.module.Module import BaseModule, ModuleDependencies
from naas_abi_core.services.email.EmailService import EmailService
from naas_abi_core.services.event.EventService import EventService
from naas_abi_core.services.object_storage.ObjectStorageService import ObjectStorageService
from naas_abi_core.services.object_storage.ontologies.modules.ObjectStorageEventOntology import (
    ObjectPut,
)
from reports.ontologies.modules.ReportEvents import ReportLoaded

REPORTS = {"prefix": {"prefix": "reports/"}, "key": {"suffix": ".csv"}}


class ABIModule(BaseModule):
    dependencies: ModuleDependencies = ModuleDependencies(
        modules=[],
        services=[ObjectStorageService, EventService, EmailService],
    )
```

`OnEvent` takes the event type as an IRI: the class's `_class_uri`.

## 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)
    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 = self.build_report(day)  # your code
        out = io.StringIO()
        csv.writer(out).writerows(rows)
        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](/v2/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=ObjectPut._class_uri, filter=REPORTS), Every("15m")),
        max_concurrency=1,
        timeout=timedelta(minutes=10),
    )
    def load_report(self, ctx: JobContext) -> dict:
        """Load CSV reports written under reports/ (ObjectPut)."""
        events = self.engine.services.events
        puts = events.query_for_consumer("reports.load_report", ObjectPut, limit=20, filter=REPORTS)
        if not puts:
            ctx.skip("no new report")
            return {}
        storage = self.engine.services.object_storage
        for put in puts:
            content = storage.get_object(put.prefix, put.key)
            rows = self.load_rows(put.key, content)  # your code: an upsert keyed by the file
            events.publish(ReportLoaded(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 class, as `ObjectPut` objects, 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 `events.seek_consumer_to_end("reports.load_report", ObjectPut)` 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 declares: any IRI under a domain you control, generated as a `LogProcess` class. `events.publish(ReportLoaded(...))` writes it to the event log and broadcasts it, like the kernel's own events; its fields become the event's JSON, 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=ReportLoaded._class_uri, filter={"rows": {"gt": 0}}),),
        max_attempts=3,
    )
    def send_notice(self, ctx: JobContext) -> None:
        """Email the team when a non-empty report is loaded (ReportLoaded)."""
        loaded = ctx.payload["data"]
        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", ReportLoaded)`.

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.
self.trigger_job("send_notice", {"key": key, "rows": rows}, idempotency_key=f"notice-{key}")

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

`trigger_job` stores the trigger in JetStream and returns once it is stored. If the follower is busy or the engine restarts, the trigger waits, up to seven days. From an `async` handler, `await ... atrigger_job(...)` instead.

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

## Or run the chain as one workflow

A [durable workflow](/v2/capabilities/build/workflows/write-a-durable-workflow) can own the whole sequence: each step is a `ctx.step` or a child run of another module's workflow, started with `ctx.call` (one) or `ctx.map_call` (one per item).

```python title="src/reports/__init__.py"
from naas_abi_core.module.workflows import SyncFlowContext, workflow


class ABIModule(BaseModule):
    ...

    def export(self) -> str:
        """Write yesterday's report and return its key."""
        ...  # what export_report does, returning the object's key

    def load(self, key: str) -> int:
        """Load one report and return its row count."""
        ...  # what load_report does for one file

    @workflow(triggers=(Cron("0 0 2 * * *", input={}),), max_attempts=3)
    def nightly(self, ctx: SyncFlowContext, inp: dict) -> dict:
        key = ctx.step("export", self.export)
        rows = ctx.step("load", self.load, key)
        ctx.call("notices", "send_notice", {"key": key, "rows": rows})
        return {"key": key, "rows": rows}
```

The steps are plain methods of their own, not the `@job` handlers above: a step function takes its own arguments, not a `JobContext`. `send_notice` here is a `@workflow` of a `notices` module with no trigger of its own, not the job of this page: `ctx.call` starts another module's workflow as a child run, by module id and workflow name.

| | Events | `ctx.call` / `map_call` |
|---|---|---|
| Who knows the next step | Nobody: listeners subscribe | The parent names each child |
| History | One run per job | One parent run, its steps linking to each child run |
| A retry | Re-runs that job | Skips the steps and children that finished |
| A missing follower | An event is lost if no replica listens | The child's trigger waits in JetStream for its module |
| Cost | Nothing waits | The parent holds its slot while a child runs |

Choose events when other modules should be free to listen, or when steps run far apart. Choose a workflow when the steps form one task whose progress and failures you want in one place. A workflow that calls itself is refused (`SelfCall`), and the children should be safe to run twice, like any job.

## 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](/v2/capabilities/build/modules/module-jobs#write-handlers-that-can-run-twice). The event log and its cursors are on [Events](/v2/capabilities/build/platform/events-overview).