> ## Documentation Index
> Fetch the complete documentation index at: https://docs.lerian.studio/llms.txt
> Use this file to discover all available pages before exploring further.
# Job events
> Consume the job.completed and job.failed notifications a Fetcher Worker publishes — payload, CloudEvents envelope, deduplication, and result verification.
Every extraction job ends in one terminal event. The Worker publishes `job.completed` when the result reaches storage, and `job.failed` when the run stops. Subscribe to those events instead of polling `GET /v1/fetcher/{id}`.
This page covers the consumer side: what arrives, what it means, and what to do with it.
## Where the events arrive
***
The Worker publishes to a durable **topic** exchange. `RABBITMQ_JOB_EVENTS_EXCHANGE` names it, and the shipped value is `fetcher.job.events`.
| Event | Routing key |
| ------------------------- | --------------- |
| Job finished successfully | `job.completed` |
| Job stopped with an error | `job.failed` |
Bind your own queue to the routing keys you care about. The local infrastructure definition binds two queues as an example: `fetcher.job.completed.queue` and `fetcher.job.failed.queue`.
Job events are a product contract, not an optional feature. The Worker refuses to start without streaming enabled and without an exchange name. See [Deployment](/en/fetcher/fetcher-deployment) for the operator side.
## `job.completed`
***
This event means one thing: the extraction ran to the end, and the encrypted result now sits in object storage.
The Worker publishes it after two earlier steps. First it writes the result object, then it records the terminal status on the job. Only then does it emit the event.
```json theme={null}
{
"jobId": "6a5b0f8c-2f1e-4a41-9f0e-3d2c1b0a9e88",
"status": "completed",
"metadata": {
"source": "payments",
"correlationId": "settlement-2026-07-13"
},
"result": {
"path": "external-data/6a5b0f8c-2f1e-4a41-9f0e-3d2c1b0a9e88.json",
"sizeBytes": 184320,
"rowCount": 1204,
"format": "json",
"hmac": "9f1c…",
"integrity": { "algorithm": "HMAC-SHA256", "signature": "9f1c…" },
"protection": { "encrypted": true, "appliedBy": "adapter", "mode": "adapter-managed" }
},
"executionTimeMs": 8421,
"completedAt": "2026-07-13T18:04:11.204Z"
}
```
`metadata` carries the metadata you sent on the job, so `source` and any correlation identifier come back to you unchanged.
`sizeBytes` and `rowCount` describe the plaintext result, before encryption. `path` is the object key of the stored result.
## `job.failed`
***
This event means the run stopped and Fetcher stored no result. Extraction is fail-fast, so the first datasource that fails ends the whole job. A partial result never reaches storage.
The event fires for any failure along the path: an unresolvable connection name, a datasource error, a schema mismatch, or a storage write that did not complete.
```json theme={null}
{
"jobId": "8f2c1a9e-4b30-4c02-9a1b-2d5e6f7a1c33",
"status": "failed",
"metadata": {
"source": "payments",
"error": { "message": "failed to connect to datasource" }
}
}
```
A failed event carries no `result` block and no `completedAt`.
`metadata.error.message` passes through a redaction step first. Fetcher replaces four leak shapes with `[redacted]`: connection URIs, the address operand of a Go network error, the `Addr:` operand of a MongoDB driver error, and an IPv4 literal. The rest of the text survives, so the message stays actionable. Route it to operators.
## The CloudEvents envelope
***
Every message travels in CloudEvents binary mode, version 1.0. The context attributes ride as AMQP headers.
| Header | Value |
| -------------------- | ----------------------------------------------------------- |
| `ce-specversion` | `1.0` |
| `ce-id` | `fetcher.job..` — the deduplication key |
| `ce-source` | The value of `STREAMING_CLOUDEVENTS_SOURCE` |
| `ce-type` | `studio.lerian.job.completed` or `studio.lerian.job.failed` |
| `ce-time` | RFC 3339 emission timestamp |
| `ce-subject` | The job identifier |
| `ce-resourcetype` | `job` |
| `ce-eventtype` | `completed` or `failed` |
| `ce-schemaversion` | `1.0.0` |
| `ce-datacontenttype` | `application/json` |
| `ce-tenantid` | The tenant that owns the job |
A single-tenant deployment still carries a tenant value. It emits the literal `single-tenant`, so one consumer handles both deployment shapes with the same code.
### CloudEvents source configuration
***
`STREAMING_CLOUDEVENTS_SOURCE` has no default value. The Worker requires it whenever streaming is on, and it stops at startup when the value is empty. The shipped example uses `//lerian.fetcher/worker`.
Fetcher copies the value into `ce-source` verbatim. Give each Worker deployment its own source value when several producers share one broker, and route on that header.
## Delivery contract
***
Delivery is **at-least-once**. Deduplicate on `ce-id`.
* The Worker writes the event to a durable outbox before it publishes. A broker outage delays the event, it does not lose it.
* A repairer scans for terminal events that never published and re-emits them every 30 seconds.
* `ce-id` stays identical across every re-emission of the same job and status. Nothing else is stable enough to key on.
* Order is not a guarantee. Two jobs can complete in one order and arrive in another.
* The Worker records the terminal job status before it emits. `GET /v1/fetcher/{id}` remains the authority on job state.
Treat a repeated `ce-id` as a duplicate and acknowledge it without reprocessing. A consumer that keys on the message identifier of the broker will process the same job twice.
## Verifying what you receive
***
Two independent signatures protect the path between the Worker and you. Both use HMAC-SHA256 with the **external HMAC key**. HKDF-SHA256 derives that key from the `APP_ENC_KEY` master key. The Fetcher repository ships a small tool that prints it, so a consumer verifies signatures without ever holding the master key.
**The message.** The Worker signs each published message and stamps three headers on it: `x-message-signature`, `t` for the signing timestamp, and `signature-version`. The signed payload binds the timestamp, the signature version, the tenant, the job identifier, the exchange, and the routing key, followed by the message body. Replaying the same body under another tenant or another route fails verification.
**The result.** `result.hmac` is the keyed HMAC-SHA256 over the plaintext result JSON, computed before encryption. The `integrity` block states the same value with its algorithm. Verify it after you decrypt and before you trust the rows.
The `protection` block describes the stored bytes: `encrypted` is true, the storage adapter applied it, and the mode is `adapter-managed`. It describes the result only, never the datasource credentials.
## Next steps
***
What a job requests, and the states it moves through.
Create a job, read a job, and manage connections.
The streaming, exchange, and encryption variables behind these events.
The Manager, the Worker, and the Engine they both run over.