Batcher flush retries starve intake: a ClickHouse stall drops incoming events while retrying

Summary

The ClickHouse events batcher processes everything on a single goroutine: the same loop that pulls items off the queue also runs the flush (and, with #3774, the flush retries). While that goroutine is blocked flushing/retrying, it is not draining the queue, so incoming events pile up and Push drops them. During a ClickHouse slowdown the two mechanisms work against each other — retrying to save the current batch causes new events (including terminal killed events) to be dropped.

Follow-up to #3770 (silent terminal-event drops) and #3774 (bounded-blocking push + bounded retry + bounded drain). #3774 notes this interaction explicitly; this issue proposes the structural fix.

Where

packages/clickhouse/pkg/batcher/batcher.go — processBatches is one goroutine; its flush() closure calls flushWithRetry synchronously:

for {
  select {
  case item, ok := <-b.ch:   // intake
      ...
      if len(batch) >= maxBatchSize { flush() }   // flush blocks the same loop
  case <-ticker.C:
      flush()
  }
}

Push is non-blocking and drops on a full queue:

select {
case b.ch <- item: ...
default:
    mItemsDropped.Add(...)        // dropped
    return ErrBatcherQueueFull
}

So any time flush()/flushWithRetry is in-flight — and especially while it backs off and retries a failing ClickHouse — nothing is reading b.ch. Under sustained pressure the queue fills and subsequent events are dropped, regardless of the bounded-blocking push added for terminal events in #3774 (their bounded wait simply expires because the consumer isn't moving).

Impact

• A transient ClickHouse slowdown converts from "delay" into "data loss on intake": the batch being retried survives, but events arriving during the retry window are dropped.
• Makes the per-type drop counter from fix(clickhouse): stop silently dropping terminal lifecycle events under burst #3774 spike precisely when operators least want loss, and undercuts the bounded-blocking push (the queue never frees because the single consumer is parked in retry).
• Larger prod batcher settings (max-batch-size, longer flushes) widen the window during which intake is starved.

Proposed fix

Decouple intake from flush so retries never block the queue reader:

• A dedicated consumer goroutine owns b.ch and batch assembly; it hands completed batches to a separate flush worker (or a small bounded pool) via an internal channel. Intake keeps draining while a flush/retry is in progress.
• Bound in-flight flushes (e.g. 1–2) so a stalled ClickHouse applies back-pressure to the consumer (consumer blocks handing off the next batch) rather than silently dropping at Push. This turns "drop on intake" into "bounded back-pressure", which is the intended behavior and composes correctly with fix(clickhouse): stop silently dropping terminal lifecycle events under burst #3774's PushWithContext and StopWithContext (shutdown still bounded; in-flight flush still cancellable).
• Keep it flag-gated / behavior-preserving by default; the single-goroutine path stays until the worker path is validated.

Net effect: a ClickHouse blip delays delivery (bounded, retried) without dropping freshly-arriving terminal events — closing the A↔B interaction called out in #3774.

Alternatives considered

• Just enlarging the queue (done in prod): raises the overflow threshold but doesn't fix the starvation — a long enough stall still fills any finite queue while the single consumer is parked.
• Dropping retries: regresses fix(clickhouse): stop silently dropping terminal lifecycle events under burst #3774's transient-blip durability.

Acceptance criteria

• Under an injected ClickHouse stall, events pushed during the stall are retained (up to a bounded back-pressure limit), not dropped, while the stalled batch is retried.
• Shutdown remains bounded (interop with fix(clickhouse): stop silently dropping terminal lifecycle events under burst #3774's StopWithContext): no leaked flush goroutine, in-flight flush cancellable within the drain budget.
• Default behavior unchanged until the worker path is enabled.