> ## 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}
{
"job_id": "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",
"size_bytes": 184320,
"row_count": 1204,
"format": "json",
"hmac": "9f1c…",
"integrity": { "algorithm": "HMAC-SHA256", "signature": "9f1c…" },
"protection": { "encrypted": true, "applied_by": "adapter", "mode": "adapter-managed" }
},
"execution_time_ms": 8421,
"completed_at": "2026-07-13T18:04:11.204Z"
}
```
Fetcher copies submitted metadata into the event, so `source` and a correlation identifier remain application fields. On `job.failed`, `metadata.error` is reserved for Fetcher's sanitized error details and replaces a caller-supplied value for that key.
`size_bytes` and `row_count` describe the plaintext result, before encryption. `path` is the object key of the stored result.
## `job.failed`
***
This event means the run stopped. Extraction is fail-fast, so the first datasource that fails ends the whole job. A `job.failed` payload has no `result` or `completed_at`, but it does not prove that no object exists: the Worker writes the encrypted object before it persists `completed`, and a terminal-status persistence failure can produce `job.failed` after storage succeeded.
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}
{
"job_id": "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 `completed_at`.
`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` | `2.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.
Schema `2.0.0` is a breaking payload version. Update your consumer for the snake\_case Fetcher-owned fields shown above. The `job.completed` and `job.failed` routing keys and the `ce-id` format do not change.
### 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.
* Every 30 seconds, a repairer scans terminal jobs whose pending-event marker remains set. That includes a failed marker-clear after a successful publish, so it can re-emit an event that already reached the broker.
* `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 HMAC-protected artifacts use 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 them without 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 envelope signs the timestamp, signature version, tenant, exchange, routing key, and message body. Current v2 events emit `job_id`, while the envelope extractor recognizes `jobId`; do not rely on a separately extracted job-ID field.
**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.