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:
connector:
push_queue:
enabled: true
max_quota_mb: 2000
on_full: block # evict_oldest (default) | blockevict_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 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 |
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 oldestThe 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: blockwith a quota sized for the longest outage you must ride out (records/s x 1 KB x seconds),respond: on_doneon webhooks where the caller can wait (the caller then sees the real verdict), NATS JetStream as the transport, andtelemetry.alertsconfigured so thehighalert 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.