# Ingest Pipeline > **Related:** [Access Control](https://wavehouse.dev/access-control.md) · [API Reference](https://wavehouse.dev/api.md) · [Architecture](https://wavehouse.dev/architecture.md) · [Claude Code & AI agents](https://wavehouse.dev/claude-code.md) · [Configuration](https://wavehouse.dev/configuration.md) · [Deployment](https://wavehouse.dev/deployment.md) · [Development](https://wavehouse.dev/development.md) · [Durability & Storage](https://wavehouse.dev/durability.md) · [Getting Started](https://wavehouse.dev/getting-started.md) · [Named Pipes](https://wavehouse.dev/pipes.md) · [Behind a reverse proxy](https://wavehouse.dev/reverse-proxy.md) · [TypeScript SDK](https://wavehouse.dev/sdk.md) · [Settings Directory](https://wavehouse.dev/settings-directory.md) · [Why WaveHouse?](https://wavehouse.dev/why-wavehouse.md) > **Also:** [HTML version](https://wavehouse.dev/ingest-pipeline) · [Docs index](https://wavehouse.dev/llms.txt) --- This page is the deep-dive on `internal/ingest` — the worker that turns the stream of ingest events into batched ClickHouse inserts. The [Architecture](/architecture) page covers where it sits in the system; this page covers **how the code itself works** so contributors and reviewers can reason about (and safely change) it. It is deliberately detailed: this is a hot, concurrency-heavy path, and the goroutine / channel / timer interplay is subtle. ## Responsibilities and files | File | Contents | | --- | --- | | `worker.go` | `StartIngestWorker`, the `dispatchLoop`, `parseMsg` (+ `rejectPoison` for an envelope it cannot read), the per-tenant-table `tableBatcher`/`tableLoop`, `flushTable` (splits a batch per column list via `groupByColumns`, or hands the whole batch of a tenant with no ClickHouse connection to `parkBatch`) and `flushGroup` (bulk insert with a row-by-row isolation fallback when ClickHouse rejects the batch, or refuses a multi-row batch for its size — `chconn.Splittable`), `retryLater` (a batch ClickHouse could not take, handed back for a delayed redelivery), `insertToClickHouse` (into the batch's tenant's ClickHouse, `chconn.Pools.Target`), `handleSuccess` (acks, after `invalidate` bumps the tenant's cache namespaces — under every tenant on the same ClickHouse address and database, through the cache `internal/app` hands the worker, since they read the same tables), `sendToDLQ`/`parkOnDLQ` | | `backoff.go` | The retry backoff per ClickHouse pool: one outage backs off every table on it together, probing once per window; a failure of one table (read-only, too many parts or mutations, a missing grant) backs off that table alone — see [When ClickHouse cannot take an insert](#when-clickhouse-cannot-take-an-insert) | | `compact.go` | `EncodeCompactRow` — renders one record as a `JSONCompactEachRow` line over the table's **insertable** columns, in declaration order. Serialization only: it validates nothing and judges no value | | `sweeper.go` | The **Active Sweeper** — every minute, asks the MQ to purge the events that are both written to ClickHouse and past the SSE gap window (the purge arithmetic below lives in `internal/mq/purge.go`) | | `claims.go` | `ClaimShards` — over the external broker's shards, narrows the worker to this process's share: the membership lease, the per-tick assignment, handover by drain then release, the reset of a dead owner's shards at takeover, each shard's share of the `MaxHeld` budget of rows held, the halt that lets the worker write what it holds before a clean stop releases its shards, and the shard gauges. The behavior is in [Scaling ingest processes](/deployment#scaling-ingest-processes). | | `assign.go` | `assignUnits` — rendezvous hashing capped at ⌈units/slots⌉, the same owners in every process for the same view | | `types.go` | `EventMessage` wire format and the `BufferConsumerName` constant | The pipeline is **insert-only**. (Upgrading across the v2 envelope? [Drain the queue first](/deployment#upgrading-across-the-v2-ingest-envelope).) The wire format carries `{table_name, scope, received_timestamp, format, columns, row}`: `row` is one `JSONCompactEachRow` line — a positional JSON array — and `columns` names its positions — the table's insertable columns, in declaration order (a `MATERIALIZED` or `ALIAS` column cannot be named in an `INSERT`, so it is not part of the row's contract). (`scope` is reserved and always `""` today.) Each NATS message is its own envelope, so the names ride along per record; where they are carried once is the `INSERT` the worker emits per group. The worker parses the envelope, groups a batch by column list, and bulk-`INSERT`s each group as `INSERT INTO … (cols) FORMAT JSONCompactEachRow` — schema validation already happened at the HTTP ingest handler, before publish. Non-insert mutations go through `POST /v1/ops/query` (admin-only) or an operator-authored [pipe that writes](/pipes#pipes-that-write). ## High-level shape With the embedded broker each tenant's events are queued on a JetStream stream of its own, and one process holds one durable consumer on each tenant's stream (for `mq.backend: nats`, see [Scaling to multiple instances](#scaling-to-multiple-instances)), delivered into one handler, and fans events out to a goroutine per tenant table — the tenant is the subject's leading token. Each tenant's table batches independently and POSTs to ClickHouse over the HTTP interface (`JSONCompactEachRow`). When ClickHouse rejects a bulk insert the batch is re-inserted row by row, so a single poison row can't sink it: clean rows ack, and only the rows ClickHouse rejects again go to the dead-letter stream. When ClickHouse cannot take the insert at all — down, overloaded, read-only — nothing is dead-lettered for it: the batch goes back to the queue and is retried with backoff. The one exception is a multi-row batch refused for its size (too many partitions, the memory limit), which is first split row by row like a rejected one; if a row then fails any way but a rejection, that row and every row after it go back. A batch whose tenant has no ClickHouse connection — one no longer served, or one no pool could be opened for (such as by the connection ceiling) — skips the row-by-row retry, which no row of it could pass, and meets the dead-letter switch once, whole; a tenant no longer served has no switch to read, so its batch is parked. An envelope the worker cannot *read* — malformed JSON, an unknown row `format`, or columns and a row that don't pair — never reaches a table loop at all: `parseMsg` parks it on the same dead-letter stream, or, where the DLQ is off for the table, acks and drops it rather than redelivering a message that can never insert. With the embedded broker, a separate sweeper reclaims stream storage. ```mermaid flowchart LR API["POST /v1/ingest"] -->|"publish ingest.TENANT.TABLE"| Stream subgraph NATS["Embedded NATS JetStream (in-process)"] Stream["INGEST_TENANT stream, one per tenant
ingest.TENANT.>
LimitsPolicy + DiscardNew"] Cons["buffer-consumer
(durable, pull, one per tenant stream)"] Stream --> Cons end Cons --> D subgraph Worker["Ingest worker (one process)"] D["dispatchLoop
(route by tenant + table)"] D --> TLa["tableLoop: clicks"] D --> TLb["tableLoop: events"] D --> TLc["tableLoop: ..."] end TLa -->|"JSONCompactEachRow POST"| CH[("ClickHouse")] TLb --> CH TLc --> CH TLa -.->|"poison rows"| DLQ["DLQ_TENANT stream
dlq.TENANT.TABLE"] D -.->|"unreadable envelope"| DLQ Sweep["Active Sweeper"] -.->|"reads AckFloor, purges"| Stream Stream -.->|"live events"| Hub["hub-bridge consumer
(SSE fan-out)"] Stream -.->|"DeliverByStartTime gap-fill"| Replay["replay consumer, one per connection
(the tenant's subject)"] ``` Note the embedded broker's stream is **dual-use**: it is both the durable buffer feeding the worker and the replay buffer that SSE clients gap-fill from. That is why a custom sweeper exists there instead of plain work-queue auto-deletion; `mq.backend: nats` splits the two roles instead, with work-queue partitions and a separate history stream (see [Scaling out](#scaling-to-multiple-instances)). :::note[Omitted columns on `Nullable` columns with a default] Inserts also pin `input_format_null_as_default=1`. A positional row has one value per insertable column and no way to say "absent", so a field the record omitted rides as an explicit `null` in its slot. That setting turns the `null` back into the column's default for a **non-nullable** column, matching what omitting the key did under `JSONEachRow` — but on a `Nullable(T) DEFAULT …` column ClickHouse stores `NULL` whatever the setting says, because only an *absent* key ever took the default. So such a column now stores `NULL` where it previously took its default. Verified on ClickHouse 26.6.3. ::: :::note[ClickHouse timestamp parsing] Inserts pin `date_time_input_format=best_effort` — the server default since ClickHouse 26.5, but on older servers the `basic` default rejects the canonical RFC 3339 form's `Z` suffix ([#372](https://github.com/Wave-RF/WaveHouse/issues/372)). The ordinary spellings (zone-less date-times, 9–10-digit Unix-seconds strings) parse identically under both settings. (This is moot for anything an older build buffered: the upgrade deletes it — see [Upgrading across the v2 ingest envelope](/deployment#upgrading-across-the-v2-ingest-envelope).) Bare digit-strings of other lengths are the exception: `best_effort` reads them as ClickHouse's calendar/epoch shapes, where `basic` read a plain `DateTime` column's digit string of five or more digits as Unix seconds (shorter runs it rejected outright, where `best_effort` reads `"2026"` as a year): under `best_effort` `"20260711"` stores 2026-07-11, where `basic` stored 1970-08-23. `DateTime64` columns diverge the same way on calendar-shaped runs, and additionally whenever an epoch run's unit doesn't match the column scale (under `basic`, runs longer than 10 digits are ticks at the column's own scale; `best_effort` unit-detects 13/16/19-digit runs as ms/µs/ns). A producer relying on the old `basic` reading changes meaning as soon as this WaveHouse version is deployed — the pin, not a ClickHouse upgrade, is what flips the parse. ::: ## The journey of one event ```mermaid sequenceDiagram participant P as POST /v1/ingest participant JS as JetStream participant CB as Consume callback (the tenant's) participant D as dispatchLoop participant TL as tableLoop participant CH as ClickHouse P->>JS: publish ingest.0.clicks (EventMessage) JS->>CB: deliver (prefetch up to pullMaxMessages) CB->>D: msgChan channel send D->>D: parseMsg (validate envelope, route key = tenant + table_name) D->>TL: per-tenant-table channel send TL->>TL: add row#59; arm deadline timer on first row Note over TL: flush on size (maxBatch) OR deadline (maxWait) TL->>CH: POST JSONCompactEachRow (flush goroutine) CH-->>TL: 200 OK TL->>JS: DoubleAck each row (background, ackWg) Note over JS: consumer AckFloor advances#59; Sweeper may now purge ``` ## Goroutine topology The design rule is **single-owner state**: each piece of mutable batching state is touched by exactly one goroutine. The one shared exception is the retry backoffs (`backoff.go`), which every table loop and flush goroutine consults under their own locks: every flush takes them, and a row arriving costs one atomic load while no backoff is open and read-locked map lookups plus each backoff's mutex once one is — and a table that fails once and is never written again keeps one open. The one fan-in is at the top: each tenant's stream is delivered on a nats.go goroutine of its own, and they all send into the one `msgChan`, which is safe from all of them at once; everything from `dispatchLoop` down stays single-owner, and a full `msgChan` pauses every tenant's delivery (layer 2 below). ```mermaid flowchart TD CB["Consume callbacks
(one nats.go goroutine per tenant stream)"] -->|"msgChan (cap maxBatch*2)"| D D["dispatchLoop
1 goroutine — owns the routing map
the ONLY ctx watcher — tracked by wg"] D -->|"per-tenant-table chan (cap maxBatch)"| T1["tableLoop: clicks
owns its batch + timer
tracked by tableWg"] D --> T2["tableLoop: events
tracked by tableWg"] T1 -->|"go (at most 1 in flight)"| F1["flush goroutine
insert to ClickHouse"] F1 -->|"go"| A1["ack goroutines
DoubleAck — tracked by ackWg"] T2 -->|"go"| F2["flush goroutine"] F2 --> A2["ack goroutines (ackWg)"] ``` Three `WaitGroup`s form a strict containment hierarchy, which is what makes shutdown correct (below): - **`wg`** tracks the `dispatchLoop` goroutine. - **`tableWg`** (owned by `dispatchLoop`) tracks the per-tenant-table `tableLoop`s. - **`ackWg`** tracks the background `DoubleAck` goroutines, and the poison disposal (a DLQ publish *plus* a `DoubleAck`) that `rejectPoison` backgrounds from the dispatch loop. ## Why per table? The bug this design fixes A single shared batch across all tables couples them: a high-volume table can trip the size trigger and strand a low-volume table's rows in a batch that then waits for the time trigger, and vice-versa. Routing each table to its own `tableLoop` gives every table an **independent** size trigger and timer, so one table's traffic never delays another's. (`dispatchLoop` does no batching itself — it parses the envelope, pairs `columns` with `row`, and picks the route key; an envelope it cannot read never reaches a `tableLoop`.) ## The `tableBatcher` state machine Each `tableLoop` owns a `tableBatcher`. It has exactly two flush **triggers** — the batch reaching `maxBatch` (checked in `add`) and the `maxWait` deadline timer — plus a rule that **at most one insert runs per tenant table at a time** ("coalescing"). A flush *completing* is **not** a trigger. The `flushing` channel signals "an insert is in flight" (it is `nil` when idle — and receiving from a `nil` channel blocks forever, so the loop's `<-flushing` arm is automatically inert while idle). The `flushQueued` flag **latches** a trigger that fires while an insert is already running, so the deferred flush runs the moment the slot frees. ```mermaid stateDiagram-v2 [*] --> Idle Idle --> Accumulating: first row (arm timer) Accumulating --> Accumulating: add (under maxBatch) Accumulating --> Flushing: size or timer — flush Flushing --> Flushing: add (fresh batch) Flushing --> FlushingQueued: size or timer — latch FlushingQueued --> FlushingQueued: add Flushing --> Accumulating: done — rows remain Flushing --> Idle: done — empty FlushingQueued --> Flushing: done — deferred flush ``` Two consequences worth internalizing: - **`maxBatch` is a "try to flush" threshold, not a hard cap.** If rows keep arriving while a flush is in flight, the next batch coalesces and can exceed `maxBatch`, flushing as one larger insert when the slot frees. That is *good* for ClickHouse part pressure (bigger batches when hot) and is bounded upstream by `maxAckPending`. - **A partial leftover after a size flush waits for its own size/timer.** When 500 rows flush and 100 remain, those 100 do **not** flush just because the first insert finished — they wait for their own `maxBatch` or `maxWait`. The `flushQueued` latch is also what prevents a stranding bug: if the leftover's timer fires *during* the in-flight insert, the latch remembers it so the rows still flush when the slot frees (rather than the tick being silently lost). ### Why there are no data races When a flush starts, the batcher hands the goroutine a **private snapshot**: ```go rows := b.batch // snapshot the slice header b.batch = nil // fresh batch here; appends allocate a new backing array go func() { w.flushTable(ctx, b.table, rows) }() ``` The flush goroutine only ever touches `rows` and the worker's concurrency-safe collaborators (HTTP client, cache, `ackWg`, the backoffs); it never touches `b.batch`, `b.timer`, or `b.flushing`. Those are touched solely by the `tableLoop` goroutine. `b.batch = nil` (rather than `b.batch[:0]`) is load-bearing — reusing the array would let new appends overwrite rows the flush is still reading. The race detector (`go test -race`) guards this. ## Contexts The signal context cancels `app.Run`'s errgroup context, which cancels `workerCtx` — the worker's stop signal. Two more are immune to it by construction: `flushCtx` is `context.WithoutCancel(workerCtx)`, and the drain deadline each component builds is rooted in `context.Background()`. Each has one job. ```mermaid flowchart TB PC["signal ctx
(main, WithCancel + signal.Notify)"] -->|"SIGINT/SIGTERM cancels"| RC["run ctx
(app.Run's errgroup)"] RC --> WC["workerCtx
(child — the STOP signal)"] WC -->|"context.WithoutCancel"| FC["flushCtx
(values only; never canceled)"] WC -.->|"only dispatchLoop watches Done()"| D[dispatchLoop] FC -.->|"passed down; never watched"| TL["tableLoops + flushes"] ``` - **`workerCtx`** is the stop signal. **Only `dispatchLoop` watches it.** Every downstream goroutine stops via channel-close instead, which gives a deterministic drain with no select race that could abandon buffered rows. - **`flushCtx` = `context.WithoutCancel(workerCtx)`** carries trace values but is never canceled. A flush that has started must finish, so data already written to ClickHouse gets acked rather than redelivered. It is bounded by the HTTP client timeout (30s); shutdown bounds the *wait* for it with a deadline. - Each component that drains builds its own **shutdown-deadline context** (`App.shutdownContext` in `internal/app`, `server.shutdown_timeout`), rooted in `context.Background()` so it survives `workerCtx` being canceled. The ingest worker's and the API server's run concurrently, so the drain phase is bounded by one timeout, not their sum. The principle: **`ctx` cancellation is the stop mechanism for long-running loops; `Close()`/stop-funcs are the mechanism for resources.** ## Lifecycle and shutdown Startup: `StartIngestWorker` creates the consumer, builds the worker, and launches `dispatchLoop`. It returns a `stopFunc` closure that the app's ingest-worker component holds and calls once the run context is canceled; `app.Run` does not return until that drain has finished. It also returns a `failed` channel, which carries at most one error if the worker ends on its own — see [When the consumer dies](#when-the-consumer-dies). Shutdown drains **bottom-up through the containment hierarchy**, under the ingest-worker component's own `server.shutdown_timeout` deadline (the API server's drain takes another, concurrently): ```mermaid sequenceDiagram participant M as app: ingest-worker component participant SF as stopFunc participant D as dispatchLoop participant TL as tableLoops participant A as ack goroutines M->>M: run ctx canceled (SIGTERM via app.Run) → shutCtx (deadline) M->>SF: stop(shutCtx) SF->>D: workerCancel() → ctx.Done fires D->>TL: close every per-tenant-table channel TL->>TL: drain buffered rows, await in-flight insert, final flush TL-->>D: tableWg drains D->>A: ackWg.Wait() A-->>D: background acks drained D-->>SF: dispatchLoop returns → wg drains SF-->>M: waitOrDeadline returns nil (or deadline error) ``` Why this ordering is correct: every `ackWg.Add` happens either inside a `tableLoop`'s lifetime **or on the `dispatchLoop` goroutine itself** (`rejectPoison`, reached from `parseMsg`), and `dispatchLoop` is the goroutine that `Wait`s — so no `Add` can race the `Wait` on either path. The `tableLoop` ones all complete before `tableWg.Wait()` returns — which means `dispatchLoop` can safely `ackWg.Wait()` afterward with no `Add` racing `Wait`. The old code relied on "flush runs synchronously" for this; the hierarchy makes it structural instead. If the deadline fires first, `waitOrDeadline` returns the deadline error and the in-flight goroutines are abandoned — the process is exiting anyway, and anything un-acked is redelivered on the next boot (at-least-once). Messages still sitting in `msgChan` or the consumer's prefetch buffer at shutdown are **not** flushed; they are simply redelivered next boot. Graceful shutdown flushes the in-hand per-tenant-table batches, not the entire in-flight pipeline. ### When the consumer dies Delivery can end underneath a running worker: the durable consumer is deleted, or the MQ connection closes. The broker client reports either only through an asynchronous error callback and then stops delivering — no message ever arrives to say so, so a loop that only watches `msgChan` would wait forever while the API kept accepting events nothing writes. `mq.Consumer.Consume` therefore returns a `failed` channel next to `stop` (`mq.ErrDeliveryEnded`, wrapping the broker's reason), and `dispatchLoop` selects on it beside `ctx.Done()` and `msgChan`. On a failure it runs the same bottom-up drain as a shutdown — the rows already in hand are flushed and acked, not abandoned — and then reports the error on the worker's own `failed` channel. A consumer that cannot start at all takes the same path. The worker does not try to revive the consumer. The app's ingest-worker component returns the error from `app.Run`, which stops every other component and exits non-zero, the same way any failed component does; with the embedded broker the supervisor's restart recreates the durable consumer at boot, and everything unacked is redelivered (at-least-once). Under `mq.backend: nats` WaveHouse never creates the durable: the restarted process waits `mq.nats.topology_wait` for the operator to recreate it, then refuses to boot naming it. Passing conditions the client also reports through that callback (a missed heartbeat) are logged at `WARN` and do not end the worker; a consumer leadership change, and a pull refused because another process holds the shard's pin, are logged at `DEBUG`. With the embedded broker (`DontListen`, no external client that could delete a durable) this path is hard to reach. A tenant's queue, opened at runtime, that the consumer cannot join is not a way in: the broker keeps that queue closed to the tenant's publishes, which answer `503` until a reload or a publish joins the consumer, and the worker goes on delivering every other tenant's rows. Under `nats` it is reachable: an operator deleting one of the configured shard durables, or a connection closed for good. The end of a removed partition's delivery, or of a durable past a lowered V (see [Choosing and changing N](/deployment#choosing-and-changing-n) and [Choosing and changing V](/deployment#choosing-and-changing-v)), only stops draining it. ## When ClickHouse cannot take an insert A failed insert is classed by `chconn.Classify` (`internal/chconn/errclass.go`) before anything else happens to it, because the HTTP status does not tell a rejected row from a ClickHouse that cannot take work — a row it cannot parse is a `400` and an unknown table a `404`, but a server out of memory, and many a rejected row, are a `500` — so the exception code decides: | Class | What it covers | What the worker does | | --- | --- | --- | | `Rejected` | Any ClickHouse exception code outside the `Unavailable` and `Denied` lists (in full: `unavailableCodes` and `deniedCodes` in `internal/chconn/errclass.go` — the `Unavailable` row below is abridged): `CANNOT_PARSE_*`, `TYPE_MISMATCH`, `INCORRECT_DATA`, `UNKNOWN_TABLE`, `NO_SUCH_COLUMN_IN_TABLE`, …; also a `413` from a proxy | Row-by-row isolation; the rows rejected again go to the DLQ (or, with `dlq.enabled` off, stay unacked) | | `Unavailable` | Connection refused/reset, DNS, timeouts (the client's and `TIMEOUT_EXCEEDED`/`SOCKET_TIMEOUT`), `NETWORK_ERROR`, `TOO_MANY_SIMULTANEOUS_QUERIES`, `SERVER_OVERLOADED`, `QUOTA_EXCEEDED`, `MEMORY_LIMIT_EXCEEDED`, `TOO_MANY_PARTS`, `READONLY`, `TABLE_IS_READ_ONLY`, `NOT_ENOUGH_SPACE`, `KEEPER_EXCEPTION`, `ALL_CONNECTION_TRIES_FAILED`, lost replicas and quorum, `UNKNOWN_STATUS_OF_INSERT`, …; a `502`/`503`/`504`/`429`/`408` with no exception code | Retry with backoff; never dead-lettered. Not isolated, except a multi-row batch refused with `TOO_MANY_PARTS` or `MEMORY_LIMIT_EXCEEDED`, which is split first (below) | | `Denied` | `AUTHENTICATION_FAILED`, `ACCESS_DENIED`, `UNKNOWN_USER`, `WRONG_PASSWORD`, `REQUIRED_PASSWORD`, `IP_ADDRESS_NOT_ALLOWED`, `DATABASE_ACCESS_DENIED`, `USER_EXPIRED`; a `401`/`403`/`407` with no code | Retry with backoff — the credentials or grants are wrong, not the rows | | `Unknown` | No exception code and no recognizable transport failure (a bare `500` from something that is not ClickHouse, a TLS setup error) | Retry with backoff — nothing says a row was judged | The line is drawn at the exception code. A code means ClickHouse was up and read the request, and nearly all of its several hundred codes are verdicts on what it read, so an unlisted code — including one added by a future ClickHouse — is `Rejected`: the row is parked, not lost, and a retry storm on a row that can never insert would pin its tenant's ingest queue's ack floor (and a share of that tenant's `maxAckPending`) for every table of the tenant behind it. No code means ClickHouse never judged anything, so isolating the batch would only multiply the requests and dead-lettering it would park good rows. Schema drift — a table dropped or a column removed between publish and insert — is `Rejected` for the same reason: the row cannot insert into the table as it now is, and the DLQ keeps it rather than retrying it forever (reading parked rows back is [#237](https://github.com/Wave-RF/WaveHouse/issues/237)). `Denied` is retried rather than parked because a fix (a grant, or a corrected `clickhouse.username` or `clickhouse.password` and a reload) makes every row of the batch insert as it is. Two `Unavailable` codes can be earned by a batch's size alone, each of its rows inserting on its own (`chconn.Splittable`): `TOO_MANY_PARTS` is also what ClickHouse answers for a batch spanning more partitions than `max_partitions_per_insert_block` (100 by default), and a large batch can exceed the memory limit (`MEMORY_LIMIT_EXCEEDED`) where a single row would not. A multi-row batch refused with either is split row by row, like a rejected one — ClickHouse's own Distributed async inserts split a batch on the same codes. Each row that inserts is acked. If a row fails any way but `Rejected` — typically the first row, failing the same way, because the server itself is the problem — isolation stops there as it does for an outage (below): the batch goes back to the queue under the backoff, at the cost of one extra request. A one-row batch is not split again. A batch that is neither `Rejected` nor split is handed back to the queue with `mq.Message.NakWithDelay` (`retryLater`), counted one per row each time it is handed back by `wavehouse_ingest_retries_total{table, reason}` (`reason` is `unavailable`, `denied` or `unknown`, or `backoff` for rows turned away without a try). The delay comes from `backoff.go`, one backoff per ClickHouse pool — the target's URL, user and database, so every table of every tenant sharing that URL, user and database waits together (tenants on the same server under another database or user back off, and probe, on their own): 1 s doubling to a 30 s cap, each delay jittered down to half so the tables of one outage do not come back in step. While the window runs, a flush for any table on that pool makes no request, and a row that arrives is handed straight back rather than buffered, so an outage's backlog waits in the queue, not in the worker; when it elapses one flush probes — rows that arrive meanwhile are handed back with a delay of at least half a second, since a probe to a server dropping packets can take the whole 30 s client timeout — and any answer that is not an outage — a success, or a rejected row — closes it. A failure ClickHouse reports for **one table** — `TABLE_IS_READ_ONLY`, `TABLE_IS_PERMANENTLY_READ_ONLY`, `TOO_MANY_PARTS`, `TOO_MANY_MUTATIONS`, and `ACCESS_DENIED`, which is usually a grant missing on that table (`chconn.TableScoped`) — backs off that table alone, on its own backoff: the server answered, so its other tables keep inserting — until the failing table's waiting rows fill its tenant's `maxAckPending` (below) — and their successes do not close the failing table's backoff. The outage is logged at `WARN` when it starts and at most every 30 s while it lasts, and at `INFO` when ClickHouse takes inserts again. If ClickHouse cannot take a row in the middle of row-by-row isolation — it stops answering, or answers with any class but `Rejected` — isolation stops there: the rows already inserted stay acked, the rows already rejected stay parked, and the row that met the outage and every row after it go back to the queue. Rows waiting out an outage stay unacked in their tenant's ingest stream, so the [Active Sweeper](#the-active-sweeper) cannot purge past them and they count toward that tenant's `maxAckPending`: a long outage fills each affected tenant's stream to its `mq.max_bytes_gb` and that tenant's ingest answers `503` — backpressure, with nothing lost and nothing parked. The rows of a failure of one table count the same way, so one that lasts — a missing grant, a permanently read-only table — suspends delivery for every table of that tenant once its waiting rows reach `maxAckPending`, and the tenant's other tables stop inserting until it is fixed. A batch whose tenant has no ClickHouse connection at all is a different case and keeps its own rule (`parkBatch`, above). This paragraph describes the embedded broker; under `mq.backend: nats` a stall is per shard and a full partition refuses every tenant in it (see [Limits that differ from the embedded queue](/deployment#limits-that-differ-from-the-embedded-queue)). A retry does not keep arrival order, and it can insert a row twice. Rows handed back return after their delay mixed in with rows that arrived since, so after an outage a table's rows reach ClickHouse out of the order they were ingested. A plain `MergeTree` sorts every part by its `ORDER BY` key, so no query sees the order; it matters where insert order picks the winner — a `ReplacingMergeTree` without a version column keeps the last row inserted, and a `CollapsingMergeTree` expects a cancel row after the row it cancels. Give such a table a version column the producer sets in the record (`ReplacingMergeTree(ver)`, or `VersionedCollapsingMergeTree`) — not an insert-time `DEFAULT now64()`, which stamps a delayed retry later than the rows that overtook it. And a retried failure can follow an INSERT ClickHouse did commit — the client's 30 s timeout, a connection reset or EOF after the body was sent, `UNKNOWN_STATUS_OF_INSERT` — so delivery after one of them is at-least-once: the first attempt may have landed, and because the retried rows come back regrouped with newer ones, ClickHouse's insert deduplication does not match them. Nor does `dedupe.enabled`, which drops a repeated publish at the HTTP edge, before the queue. ## Backpressure and durability knobs Several layers throttle the pipeline, inner to outer: 1. **`batch`** flushes at `maxBatch` rows or `maxWait`. 2. **`msgChan`** (cap `maxBatch*2`) — when full, the consume callback blocks and delivery pauses. 3. **`pullMaxMessages`** — nats.go's client-side prefetch buffer in front of `msgChan`, shared by the tenants' streams (at least one message each). 4. **`maxAckPending`** — the server suspends a tenant's delivery once this many of its messages are delivered-but-unacked; no other tenant's delivery waits on it. The outermost in-memory bound, and a per-tenant one: while ClickHouse stalls, the worker can hold up to `maxAckPending` rows for every tenant served. 5. **`MaxBytes` + `DiscardNew`** on each tenant's stream (its `mq.max_bytes_gb` in the [settings directory](/settings-directory#message-queue), resized in place on reload) — when it fills (e.g. ClickHouse is down so nothing acks/purges), that tenant's new publishes are rejected and the API returns 503. Steps 4 and 5 are the embedded broker's; under `mq.backend: nats` a shard slows to what keeps its pin at its share of the process's 10,000-row budget of unsettled rows (an even share over its shards, never under 1,000, so a process with more than ten shards holds 1,000 per shard), which pauses every table in that shard, and a full partition refuses every tenant in it (see [Limits that differ from the embedded queue](/deployment#limits-that-differ-from-the-embedded-queue)). | Knob | Default | Meaning / invariant | | --- | --- | --- | | `maxBatch` | 500 | rows that trigger a flush (soft — coalescing can exceed it) | | `maxWait` | 5s | max time a row waits before its batch flushes | | `ackWait` | 60s | server redelivery timeout; **must exceed `maxWait` + flush time** or in-flight rows get redelivered → duplicate inserts | | `pullMaxMessages` | 500 | client prefetch, shared by the tenants' streams; keep `<= maxAckPending` | | `maxAckPending` | 10,000 | server cap on a tenant's unacked messages (backpressure) | `DoubleAck` is used (not fire-and-forget `Ack`) because acking is what records "this data is durably in ClickHouse." With the embedded server's `SyncAlways`, every ack is an fsync and therefore *slow*, which is exactly why acks run in the background (`ackWg`) off the insert path. ## The Active Sweeper The worker advances the consumer's `AckFloor` by acking; the sweep observes it to decide what is safe to purge. They never call each other — the consumer's `AckFloor` is their only contract. The sweeper (`internal/ingest`) owns the schedule and the window: each tick it calls `mq.Purger.PurgeAcked(buffer-consumer, cutoffs)` with each tenant's cutoff at now − its own `stream.gap_window_minutes` — a rejected tenant's as its folder last had it, or one before anything it holds if its folder has been rejected since boot, so its clients resume once the folder is fixed; a removed tenant is given none, and keeps none of the history it has acknowledged. The steps after the tick below are the embedded broker's implementation of that call, run on each tenant's stream at that tenant's cutoff. ```mermaid flowchart TD Tick["every 1 minute"] --> Read["read buffer-consumer AckFloor"] Read --> Gap["binary-search the gap-window sequence"] Gap --> Target["target = MIN(ackFloor + 1, gapSeq)"] Target --> Purge["stream.Purge below target"] Purge -->|"deletes msgs that are BOTH
written to ClickHouse AND past the gap window"| Stream[("INGEST_TENANT stream")] ``` `MIN(ackFloor+1, gapSeq)` is the safety argument: never purge past what is in ClickHouse, and never past the SSE replay window. If ClickHouse is down the `AckFloor` stops advancing, purging freezes, and the stream fills toward `MaxBytes` — backpressure by construction. The sweeper is one of `app.Run`'s components (`Sweeper.Start` blocks until the run context is canceled), but an interrupted sweep is harmless and idempotent, so it returns on `ctx.Done()` with no drain of its own — unlike the worker's bounded `stopFunc`. It runs under the `sweeper` lease (`coord.RunElected`), so only the process holding the lease sweeps; with the in-process coordinator that is always the one process. Under [`mq.backend: nats`](/deployment#external-nats) no sweeper is wired: the partition streams use work-queue retention, so the server deletes each row once the worker acks it, and the replay history is a separate stream the server expires by its `max_age`. The API processes instead warn (`ExternalNATS.CheckReplayWindows`), at boot and after each reload, once per tenant and window, about a gap window longer than that `max_age`. ## Scaling to multiple instances The embedded broker is single-process by construction: it listens on no port, so its "connection" cannot blip independently of the process. [`mq.backend: nats`](/deployment#external-nats) is how several processes share one queue: ```mermaid flowchart TD subgraph NATS["Operator's NATS JetStream"] P0[("ingest partition 0
work queue")] P1[("ingest partition N-1")] H[("history stream
limits, max_age")] D[("dead-letter stream")] end API["API processes"] -->|"publish: jump hash of tenant.table → (partition, shard)"| P0 API --> P1 P0 -->|"one pinned-client durable per shard"| W["ingest workers (one per shard at a time)"] P1 --> W P0 -. republish .-> H P1 -. republish .-> H H -->|"per-process consumer"| Hub["each API process's SSE hub + replay"] W --> CH[("ClickHouse")] W --> D ``` - **Work distribution.** Each partition is split into V shards, and a table's rows all go to one shard. Each shard has its own durable with the `pinned_client` priority policy, so the server delivers a shard to one pulling process at a time: a table's rows are batched and written by one process at a time. The ingest processes share the shards out: each holds a membership lease (`ingest.m`), and every process computes the same owner for every shard from the live members, by rendezvous hashing capped at an even share (`ClaimShards`). A shard given up stops fetching but keeps its pin until the rows it delivered are written, then releases it, so the next owner receives at once and only newer rows; a stopping process first halts every shard the same way (`mq.Halter`), writes what it holds, and only then releases them. A shard whose owner died is reset to its ack floor before its new owner pulls it, so the rows the dead owner held come back at once rather than after `ack_wait`; the broker refuses the reset while the shard is pinned or was active within its pinned TTL. The worker never blocks a shard's pulls: at the shard's share of the process's 10,000-row budget of unsettled rows (an even share over its shards, never under 1,000, so a process with more than ten shards holds 1,000 per shard), the shard fetches only what keeps its pin (a row every 5 seconds, at most one `ack_wait` of them past its share, then pulls that take none), so a stuck table neither loses its shard nor stalls the process's other shards. - **Idempotent inserts matter more.** At-least-once delivery plus redelivery after a crash means another process can re-insert a batch the dead one had written but not acked. Use `ReplacingMergeTree` (or a dedup key). A single process already re-inserts after a crash, or after an insert whose outcome it could not see ([When ClickHouse cannot take an insert](#when-clickhouse-cannot-take-an-insert)); several make it routine. - **NATS resilience.** The external broker reconnects on its own, with backoff; while it is disconnected a publish answers `503` with `Retry-After: 5`, and consumption resumes after the reconnect. A consumer whose delivery ends for good (one of its configured shard durables deleted, or its stream, even while no pull of it is waiting, which the next pull finds; a durable that no longer fits the worker's config; or the connection closed) ends the worker and the process, as the embedded one does. - **The sweeper.** Work-queue retention deletes each row once it is acked, one row at a time, so one tenant's unwritten rows never hold back another's reclaim, which a shared ack floor would, and it keeps a row no consumer covers rather than dropping it. SSE replay reads the history stream, which the partitions republish to and which expires by `max_age`; it is best effort, so an outage of the history costs SSE the rows republished meanwhile and never holds up ingest. So there is nothing for the sweeper to purge. ## Deferred / not yet implemented Tracked under [#191](https://github.com/Wave-RF/WaveHouse/issues/191): - **Pipelining beyond coalescing** — more than one insert in flight per tenant table (with a documented bound), once benchmarks justify the added concurrency. - **`tableLoop` reaping** — loops are spawned per distinct tenant table and never reaped; safe while tenants and table names are bounded (a settings folder per tenant, schema-validated tables, only WaveHouse publishing). Needs idle-reaping before untrusted publishers can create unbounded cardinality. Under `mq.backend: nats`, keep publish rights on `.ingest.>` to the `wavehouse` user alone: any other publisher bypasses schema validation. Tracked in [#263](https://github.com/Wave-RF/WaveHouse/issues/263). - **A hot table**: a table lives in one shard, so one process writes it. Spreading one table's rows over several shards, which gives up its order and single writer, waits for a measured need. The embedded broker keeps one stream per tenant and the sweeper. - **Parallel e2e test files.** The e2e suite now isolates tables **per file** (`tests/e2e/sdk/tables.ts` — each file gets its own `clicks_`/`events_`/`users_`), so cross-file *data* contamination is structurally impossible. Running the files in parallel (dropping `maxWorkers: 1` in `vitest.config.ts`) is still deferred: several files do read-modify-write on the **single global policy document** and `streaming.test.ts` flips the global `default_role`, so concurrent files would race those writes. Parallelism needs per-table policy storage with atomic per-table updates first — tracked in [#214](https://github.com/Wave-RF/WaveHouse/issues/214).