OpexFlow is **at-least-once**: a record is acknowledged to its source only after it is stored
(push queue, storage, event history) or handed to a destination that confirmed it. When it
cannot be stored, the source is told to keep it (HTTP 503 + `Retry-After`, JetStream nak) or,
for sources that cannot be re-read, the loss is counted and alerted. Nothing disappears
silently.

## `on_full`: what a full queue does

The durable push queue (`push_queue.db`) is bounded by `connector.push_queue.max_quota_mb`.
`on_full` chooses what happens at that quota:

```yaml
connector:
  push_queue:
enabled: true
max_quota_mb: 2000
on_full: block        # evict_oldest (default) | block
```

| | `evict_oldest` (default) | `block` |
| --- | --- | --- |
| Intended for | machine telemetry: the newest sample matters most | business flows (payments, orders, events): nothing may be dropped |
| At the quota | the oldest rows are deleted to make room (counted in `opex_connector_push_queue_gaps_total`, `push_status.evicted_total`, logged, alerted) | the new record is **refused**; nothing is deleted or dropped (`push_status.blocked_total`) |
| Sources | never notice | are backpressured, see below |
| Needs | nothing | `push_queue.enabled: true` and `max_quota_mb > 0` (checked at startup) |

`block` reserves each record's bytes against the quota before the record reaches the writer,
so a burst cannot overshoot it. A record larger than the whole quota is admitted when the
queue is empty, so it can never wedge the queue. Changing `on_full` needs a restart.

### How each source is backpressured

Only the sinks a record can actually reach count: the check walks the deployed flow from the
source node to its `sink` nodes. A webhook whose flow ends in `storage` is never refused
because the telemetry push queue is full. A `subflow` on the way makes the check
conservative (every sink of the flow).

| Source | While a sink it reaches is full under `block` |
| --- | --- |
| Signed webhook (`POST /api/hooks/{name}`) | after the signature check: `503` + `Retry-After`, body `{"ok":false,"result":"retry","sink":"push","retry_after_secs":N}`; nothing is accepted, the sender (payment gateway, ERP) retries; counted in `engine.webhooks[].backpressured` |
| Ingest API (`POST /api/ingest/{source}/{id}`) | the same `503` + `Retry-After` |
| NATS JetStream | the message is **nak'd** with the consumer backoff (2 s doubling to 60 s) before any node runs, so no side effect repeats; JetStream redelivers it (no `max_deliver` is set) |
| Polled devices (Modbus, OPC-UA, ...) | cannot be re-read: the poll task is held up to 10 s for the queue to drain, then the sample is dropped, counted (`engine.stats.backpressure_lost`) and alerted |
| MQTT, schedule ticks | MQTT acks on receipt and a tick is not replayed by design: the refusal is counted (`engine.stats.backpressured`) and visible in the schedule's last result; use NATS JetStream when MQTT data must not be lost |
| Backfill | waits and retries; never stops |

`respond: on_receipt` webhooks have already answered `200` when the flow finishes. A record
that ends with a transient failure (a sink refused or timed out) is parked in the
[dead-letter queue](/docs/operations#dead-letters-correct-and-replay) as `accepted (200) but not
handled` for inspection and replay, instead of vanishing.

### Alerts

Every bounded queue feeds one alert state machine (channels: `telemetry.alerts`):

| Level | When | Alert |
| --- | --- | --- |
| `high` | 80 % or more of the quota | warning: the destination is slow or down |
| `full` | at the quota, or a record was refused/evicted since the last check | critical: "refusing new records" (`block`) or "evicting the oldest records, data gap" (`evict_oldest`); repeated every 10 minutes while it lasts |
| `ok` | back below 80 % | recovery notice |

`GET /api/status` shows the state: `push_queue.{on_full, quota_bytes, fill_pct, blocking,
blocked_total, evicted_total, level}` and the same under `push_status`; every sink in
`sinks[]` has `dropped` (lost) and `blocked` (refused, not lost).

## Per-sink queue quotas

Each destination has its own bound, so one slow sink cannot starve the others:

| Sink | Queue | Quota | `on_full` |
| --- | --- | --- | --- |
| `push` | disk (`push_queue.db`) | `push_queue.max_quota_mb` | `push_queue.on_full` |
| `clickhouse`, `postgres` | memory (batch buffer + rows retained for retry) | `sink_batch.max_backlog_rows` (default `max(sink_batch_rows x 20, 10000)`) | `sink_batch.on_full` |
| `nats` | memory channel, 4096 | fixed | refuses (the publish returns failure, the source is told to retry) |
| `mqtt` | client queue, 1024 | fixed | refuses |

```yaml
clickhouse:
  sink_table: events
  sink_batch:
sink_batch_rows: 1000
max_backlog_rows: 50000
on_full: block        # refuse new rows at 50000 instead of dropping the oldest
```

The ClickHouse and Postgres backlogs are **memory only**: rows are lost if the process
crashes before they are inserted. `block` makes the loss at the quota impossible, not the loss
at a crash; for data that must survive a crash, deliver it through NATS JetStream or the push
queue.

## Audit: where a record can be dropped

| Place | Can it drop? | Now |
| --- | --- | --- |
| Push queue at the disk quota | yes (`evict_oldest`) | `block` refuses instead; evictions counted, logged, alerted |
| Volatile push buffer (no `push_queue`, 16 MB) | yes: at its quota and on restart | counted, logged, alerted; enable `push_queue` (startup warns) |
| Push writer channel (8192) | yes if the local disk stalls | counted (`dropped_total`), alerted |
| Push queue write error | yes | counted, alerted; acked callers get the error |
| Push dead-letter table (50 000 rows) | oldest rejected payloads are trimmed | logged |
| Webhook `on_receipt` after the `200` | a transient failure had no retrier | parked in the DLQ |
| Webhook / ingest at a full `block` queue | no | `503` + `Retry-After` |
| NATS JetStream ingest, per-device buffer (1024) | no: the oldest entry is returned to JetStream | counted |
| NATS ingest `latest` mode | yes, by design: keeps only the newest value per subject | counted |
| MQTT ingest per-subscription buffer (1024) | yes: oldest dropped, and MQTT is acked on receipt | counted; use NATS JetStream for no-loss |
| NATS sink after its publish retries | yes | counted (`failed`), alerted |
| NATS / MQTT sink queue full | the call returns failure | the source retries; counted (`dropped`) |
| ClickHouse / Postgres sink backlog | yes (`evict_oldest`); memory only | `block` refuses; counted, alerted; lost on a crash |
| DataHub broadcast (256) | slow WebSocket clients skip messages | live view only, not a delivery path |
| Pipeline: no matching source | dropped by design, settled `Ack` | counted (`engine.stats.unrouted`) |
| Pipeline: unwired port, node failure | the record goes to the dead-letter queue or the `error` edge | counted, alerted |
| Pipeline dead-letter queue (100 000 rows) | oldest resolved rows are trimmed first | logged |
| Deadband / aggregation (`preprocess`) | by design: suppressed values (push only) | counted (`suppressed_total`) |
| Schedule ticks | missed ticks are not replayed, by design | visible in `/api/schedules` |

## Recommendations

- **Business flows**: `push_queue.on_full: block` with a quota sized for the longest outage
  you must ride out (`records/s x 1 KB x seconds`), `respond: on_done` on webhooks where the
  caller can wait (the caller then sees the real verdict), NATS JetStream as the transport,
  and `telemetry.alerts` configured so the `high` alert reaches a person.
- **Machine telemetry**: keep `evict_oldest`; a gap is better than a stalled line.
- A mixed connector can use `block`: the check is per source, and pollers are only held for
  10 s, so telemetry never stalls longer than that.