# Ingest Broker

Date: 2026-03-23
Status: Frozen architecture note
Authors: Search Platform Architecture

Component: Write Plane (Ingress Stabilizer)

Scale target: 1B collections | 5 EB raw vector + metadata

## 1. Scope

This document defines the component-level architecture of `Ingest Broker`, the shared write-plane stabilizer between `Ingest Gateway` and `Write Planner`.

It applies after upstream admission has already produced canonical `MutationBatch` payloads for both ingress families:
- Flow A: `TiCDC -> CDC Adapter -> Ingest Gateway -> Ingest Broker`
- Flow B: Search-Fabric-owned write gateways -> `Ingest Gateway -> Ingest Broker`

The canonical contract stack remains:
- `api/proto/searchfabric/v1/search_fabric.proto`
- internal owner file for this boundary: `api/proto/searchfabric/v1/ingest_broker.proto`
- `docs/specs/core/search-fabric-api.md`
- `docs/specs/core/error-codes.md`
- `docs/specs/core/search-fabric-architecture.md`

This document does not redefine public RPC fields, `MutationBatch` framing, replay identity, `MutationAck` semantics, or canonical `SF_*` error meaning.
It explains how the broker should realize buffering, coalescing, fairness, and backpressure at production scale while remaining subordinate to the checked-out write contract.

Out of scope:
- external authentication termination and coarse scope derivation owned by `Unified Edge/Auth Ingress`; write-plane client/scope revalidation, route resolution, and source-payload normalization owned by `Ingest Gateway`
- Flow-A source transformation and DDL-fence behavior owned by `CDC Adapter`
- `commit_ts` assignment/preservation owned by `Commit Coordinator`
- modality planning and writer selection owned by `Write Planner`
- WAL, manifest, and visibility semantics owned by downstream write/execution components

## 2. Executive Summary & Role

`Ingest Broker` is the memory-resident stabilizer of the write path.
It sits between the stateless admission layer (`Ingest Gateway`) and the write-execution control layer (`Write Planner`).
Its job is not to reinterpret write meaning.
Its job is to absorb burstiness, pack many tiny admitted writes into efficient collection-local dispatch frames, enforce approximate fairness across namespaces, and propagate downstream pressure early enough that the ingest plane stays bounded.

Consistent with Search Fabric's CQRS stance, the broker is logic-agnostic:
- it does not perform schema validation
- it does not resolve catalog ownership
- it does not do document-level indexing
- it does not assign Flow-B `commit_ts`
- it does not create a second durable replay authority

Its responsibilities are:
- accept already normalized canonical `MutationBatch` payloads from upstream
- keep those batches in bounded memory using pooled buffers and shared references
- coalesce compatible tiny batches into larger broker-private dispatch frames per `CollectionRef`
- schedule ready frames with approximate multi-tenant fairness and whale protection
- dispatch frames asynchronously to `Write Planner`
- preserve original batch identity so downstream completion still resolves per canonical `MutationBatch`

Its non-responsibilities are:
- mutating `batch_id`, `batch_fingerprint`, `stream_identity`, `schema_generation_token`, or entry framing
- inventing broker-local schema or lifecycle decisions
- performing durable write-ahead logging as the normal correctness path
- acknowledging multiple public batches as if they were one new canonical batch

High-level data path:

```text
Flow A / Flow B
  -> Ingest Gateway
  -> canonical MutationBatch
  -> Memory-Pinned Ingress
  -> Batch Coalescer (collection-local)
  -> Virtual-Partition Scheduler (namespace-aware)
  -> Egress Dispatcher
  -> Write Planner
  -> Commit Coordinator / Writers / Visibility Barrier
```

## 3. Logical Architecture

The broker is composed of four internal modules.

| Module | Primary state | Purpose |
|---|---|---|
| Memory-Pinned Ingress | pooled buffers, admission accounting, per-batch completion slots | admit canonical batches into bounded memory without repeated deep copies |
| Batch Coalescer | `DashMap<CollectionRef, PendingBucket>`, `DelayQueue` | aggregate tiny compatible batches into larger dispatch frames while preserving original batch boundaries |
| Multi-tenant Scheduler | virtual partition queues, deficit/weight accounting, namespace pending-byte counters | approximate weighted fairness across active namespaces without resident global per-collection queues |
| Egress Dispatcher | bounded async planner permits, in-flight frame tracking | hand ready frames to `Write Planner` and fan downstream completion back to original batches |

Recommended internal shape:

```text
AdmittedBatch
  = Arc<MutationBatch> + batch_bytes + entry_count + arrival_ts + completion_slot

PendingBucket
  = collection_ref
  + namespace_id
  + current_frame
  + ready_frames
  + delay_handle

DispatchFrame
  = collection_ref
  + flush_reason
  + ordered batches[AdmittedBatch...]
  + approx_bytes
  + approx_entries
```

Important boundary:
- `DispatchFrame` is a broker-private transport unit.
- It is not a new canonical write envelope.
- Every original `MutationBatch` inside a frame keeps its own replay identity and completion outcome.

## 4. Local Design Invariants

| Driver | Local consequence inside Ingest Broker |
|---|---|
| Canonical `MutationBatch` contract is authoritative | Broker may group batches for dispatch efficiency, but it must not rewrite public batch or entry identity. |
| S3 and downstream write pipeline are the only durable truth | Broker memory is disposable; crash recovery relies on upstream replay and downstream idempotency. |
| 1B collections forbid always-on per-collection resident queues | Broker creates state only for active collections and uses fixed virtual slots for fairness. |
| CQRS keeps write control and write execution separate | Broker stabilizes and schedules; `Write Planner` and later stages still own execution policy and durability/visibility gates. |
| Backpressure is safer than hidden spill complexity | Under pressure, broker slows or rejects admission before it becomes an unbounded memory sink. |
| Collection affinity is required for effective coalescing | All writes for one `(namespace_id, collection_id)` should land on the same broker instance whenever the tier is healthy. |

Two additional invariants are critical:

1. Within one collection bucket, the broker preserves admission order.
2. Any batch containing `ProgressMark` or `BarrierMark` is emitted as a singleton dispatch frame, because mark-bearing batches are sequencing fences and the broker must not invent a second ACK/framing model around them.

## 5. Core Design Challenges

### 5.1 High-Performance Memory Management

The broker must absorb bursty write traffic at 1B-collection scale without turning itself into an allocation storm.

The key rule is:
- zero-copy begins after the broker ingress boundary, not across every deployment hop.

In a split deployment, the upstream gateway may still serialize once on the internal RPC hop.
Once the broker has admitted the batch, all broker-internal stages should move only shared references plus small metadata.

Recommended implementation rules:
- use `bytes::Bytes` / `bytes::BytesMut` pools for owned byte slabs and decode-side scratch space
- store structured admitted work as `Arc<MutationBatch>` or an equivalent immutable shared envelope
- keep separate accounting for:
  - pooled buffer bytes
  - queued-but-not-dispatched bytes
  - dispatched-but-not-yet-completed bytes
- treat bucket state as short-lived:
  - create on first active batch
  - destroy after the bucket flushes and becomes idle

This avoids the pathological case where a hyperscale deployment pays the cost of many tiny heap allocations for short-lived micro-batches.

Recommended watermarks:
- below normal watermark: admit and coalesce normally
- above high watermark: shorten linger and flush partially full frames aggressively
- above critical watermark: stop granting new ingress permits and let pressure propagate upstream

The broker should prefer stable pointer movement over serialization churn.
CPU should be spent on downstream throughput, not on repeatedly cloning the same mutation payload.

### 5.2 Intelligent Batch Coalescing (The Stabilizer)

The broker's central optimization is to merge many small admitted write requests into larger collection-local dispatch frames before they hit `Write Planner`.
This improves downstream scheduling efficiency and reduces tiny-frame overhead without redefining the public write contract.

The coalescing unit is deliberately broker-private:
- the broker never synthesizes a new public `MutationBatch`
- the broker never re-hashes or re-IDs merged work
- downstream completion still resolves per original batch

Recommended state:
- active bucket map: `DashMap<CollectionRef, PendingBucket>`
- one bucket exists only while that collection is active on the current broker instance

Recommended compatibility rule for appending to the open frame:
- same `CollectionRef`
- same canonical non-zero `policy.durability`
- same canonical non-zero `policy.visibility`
- same canonical `policy.ingest_mode`
- same `schema_generation_token`
- neither the open frame nor the incoming batch contains `ProgressMark` or `BarrierMark`

If any compatibility condition fails, the broker closes the open frame and starts a new one in the same bucket.
This stricter cut is required because the current broker -> planner boundary remains unary:
one `DispatchFrame` completes as one planner RPC response, so the broker must not place batches with different public ACK gates into the same frame and accidentally strengthen a lighter batch's latency contract.

Recommended flush triggers:
- size trigger:
  - flush when `approx_bytes >= target_frame_bytes`
  - initial production default: `1 MiB`
- row/entry trigger:
  - flush when `approx_entries >= target_frame_entries`
  - initial production default: `1000`
- time trigger:
  - flush when the oldest batch in the open frame exceeds `linger_ms`
  - initial production default: `10-50 ms` depending on workload tier
- fence trigger:
  - flush immediately before and after any mark-bearing batch
- pressure trigger:
  - flush early when global memory exceeds the high watermark
- drain trigger:
  - flush all open frames immediately during shutdown or topology rebalance

`ProgressMark` and `BarrierMark` handling must be strict:
- if a bucket already has a row-only open frame, flush it before accepting the mark-bearing batch
- emit the mark-bearing batch as its own singleton frame
- do not append later row batches to that singleton frame

This rule is intentionally conservative.
It keeps broker optimization below the canonical framing boundary and avoids semantic drift in source ordering, replay, and ACK behavior.

### 5.3 Multi-Tenant Fairness and Whale Protection

The broker cannot afford one always-resident queue per potential collection or namespace.
It therefore uses active-bucket state plus a fixed set of virtual partition queues (VPQ).

Recommended model:
- compute `slot = hash(namespace_id) % N`
- initial production default: `N = 1024`
- each slot keeps only active ready collections, not the full keyspace
- scheduler applies WFQ-style deficit accounting at the slot level
- within one slot, ready collections are served in FIFO/round-robin order

This is an approximate fairness model, not an exact global per-namespace WFQ heap.
That tradeoff is intentional.
Exact global scheduling state does not scale cleanly to 1B potential collections, while hashed virtual slots preserve bounded runtime structures and still provide strong whale isolation in practice.

Recommended cost charging per dispatch frame:
- bytes cost
- entry cost
- optional multiplier for batch-mode frames when offline work would otherwise starve interactive traffic

Recommended whale-protection rules:
- each namespace tracks pending bytes and in-flight frames
- broker applies a namespace soft quota before the global memory pool is exhausted
- once a namespace exceeds its soft quota:
  - the broker stops granting new admission permits for that namespace
  - upstream sees canonical pressure rather than cluster-wide collapse

Public projection rules remain canonical:
- namespace/global pressure because of queue or memory saturation -> retryable `SF_RATE_LIMITED`
- required diagnostic keys remain:
  - `retry_after_ms`
  - `throttled_dimension`
  - `server_limit`
- for broker-driven ingest pressure, `throttled_dimension` should stay within the canonical ingest-family values such as `ingest_bytes` or `ingest_entries`

This keeps the public quota/error owner stable:
- `Ingest Gateway` still owns public admission projection
- `Ingest Broker` performs only defensive local protection for the write plane

### 5.4 Bidirectional Backpressure

The broker must translate downstream slowness into bounded upstream behavior before memory is exhausted.

Upstream pressure path:
- ingress uses bounded async channels and permit-based reservation
- when broker memory is healthy, ingress can `reserve()` normally
- when memory crosses the critical watermark, the broker stops granting new permits
- upstream gateway then either:
  - waits for capacity, or
  - fails fast according to request deadline and policy

Downstream pressure path:
- dispatcher tracks `Write Planner` queue depth, available permits, and recent service time
- if planner throughput drops, dispatcher polls ready slots more slowly
- bucket memory rises in a controlled way
- once high watermarks are reached, that pressure naturally propagates back to upstream admission

Failure classification must stay simple:
- saturation / adaptive shedding / pressure -> `SF_RATE_LIMITED`
- internal dependency or transport outage between broker and planner -> `SF_BACKEND_UNAVAILABLE`

The broker should not mask a broken downstream dependency as mere tenant throttling.
It should also not treat ordinary high write pressure as generic backend unavailability.

## 6. End-to-End Runtime Flow

The runtime path for one admitted batch is:

1. `Ingest Gateway` forwards one canonical `MutationBatch`.
2. Memory-Pinned Ingress validates internal broker prerequisites:
   - batch metadata needed for scheduling/accounting is derivable
   - memory reservation succeeds
   - namespace soft quota is not already exceeded
3. The broker stores the admitted work as one shared `AdmittedBatch` plus a per-batch completion slot.
4. Batch Coalescer looks up the active `PendingBucket` for the target `CollectionRef`.
5. The batch either:
   - appends to the compatible open frame, or
   - forces the current frame to close and starts a new one.
6. Once a frame is ready, it is linked into the namespace-hashed VPQ slot.
7. Egress Dispatcher waits for `Write Planner` capacity, selects the next eligible slot/frame, and sends one broker-private `DispatchFrame`.
8. `Write Planner` consumes the frame but still sees the original ordered batches inside it.
9. Downstream completion resolves each original batch independently.
10. Broker fans that completion back through the saved slots, releases memory, and retires the bucket if it becomes idle.

Important completion rule:
- one `DispatchFrame` may contain many original batches
- completion must still fan out one outcome per original batch
- transport-`OK` planner responses must echo the frame batch count and identify each returned outcome by input batch index so broker can mechanically reject under-coverage, over-coverage, duplicates, or reorder drift
- the broker must never collapse those outcomes into one synthesized ACK

This rule preserves:
- batch-level replay identity
- per-batch `RequestMeta.idempotency_key` meaning
- batch-level `batch_fingerprint` validation
- original error/ACK attribution

## 7. Horizontal Scaling & Availability

### 7.1 Collection-Affinity Routing

Coalescing is most effective when all writes for one collection land on the same broker instance.

Recommended routing rule:
- `Ingest Gateway` selects broker instance by consistent hashing on `(namespace_id, collection_id)`
- the hash ring should use virtual nodes to reduce rebalance noise during scale events

Benefits:
- higher coalescing ratio for micro-batches
- warmer bucket-local memory and scheduler state
- reduced cross-instance hot-collection thrash

This affinity is an optimization, not a second correctness authority.
If the preferred broker is unavailable, the gateway may reroute to another healthy instance.
Correctness still comes from the canonical replay contract and downstream idempotency.

### 7.2 Stateless Failover

`Ingest Broker` owns no durable truth.
If a broker crashes, only in-memory, not-yet-acknowledged work is lost on that node.

Recovery model:
- upstream caller retransmits any unacknowledged canonical batches
- retransmit follows the canonical rules already owned by `search_fabric.proto` and `search-fabric-api.md`
- downstream logical exactly-once behavior still relies on:
  - batch-level idempotency (`batch_id` + `batch_fingerprint`)
  - entry-level replay identity
  - `Commit Coordinator` and writer-side canonical ordering

Broker-local disk spill is optional and emergency-only.
It must not become a hidden second write authority.
If implemented, it is for operator-assisted recovery of not-yet-dispatched frames after a process-level emergency, not for ordinary correctness.

## 8. Graceful Shutdown (Drain Mechanism)

During rolling restart or controlled shutdown, the broker should follow a strict drain protocol:

1. `SIGTERM` reception:
   - stop accepting new ingress work
   - unregister the instance from broker-service discovery so new affinity routing stops landing here
2. Force flush:
   - close every open coalescing frame immediately
   - enqueue those frames onto the ready scheduler path
3. Wait for empty:
   - continue dispatching until `pending_count == 0`
   - exit successfully only after all original batches have been completed
   - enforce a bounded timeout, for example `30s`
4. Emergency spillover (optional):
   - if timeout or fatal downstream outage prevents a clean drain, serialize still-pending not-yet-acknowledged frames to an ephemeral local WAL
   - on next startup, operators may replay or discard that file only under an explicit recovery procedure

Normal correctness does not depend on Step 4.
The primary recovery mechanism remains upstream replay of unacknowledged canonical batches.

## 9. Failure Semantics and Observability

### 9.1 Required Fail-Closed Behaviors

| Condition | Broker action | Public projection |
|---|---|---|
| namespace soft quota exceeded | stop admitting new work for that namespace | retryable `SF_RATE_LIMITED` |
| global memory critical watermark | stop granting ingress permits | retryable `SF_RATE_LIMITED` |
| planner queue saturation | slow dispatch and let pressure propagate | retryable `SF_RATE_LIMITED` |
| planner transport/dependency outage | fail new dispatch attempts | retryable `SF_BACKEND_UNAVAILABLE` |
| broker crash before completion | lose only in-memory unacked work | upstream replay, no silent success |
| drain timeout during shutdown | exit after bounded wait | upstream replay for anything unacked |

### 9.2 Required Telemetry

The broker should emit:
- pooled-buffer bytes and pool hit/miss rate
- queued bytes, in-flight bytes, and pending batch count
- active bucket count and active virtual-slot count
- frame-size histograms (`bytes`, `entries`, `batches_per_frame`)
- flush-reason counters (`size`, `time`, `mark`, `pressure`, `drain`)
- scheduler deficit and skip counters by slot
- namespace soft-quota rejection counters
- planner backlog and dispatch latency
- admission-to-dispatch latency and dispatch-to-completion latency
- coalescing ratio:
  - original batches per dispatch frame
  - bytes saved in queue wakeups versus one-batch dispatch
- drain duration and forced-exit count

Logs and traces should include:
- `namespace_id`
- `collection_id`
- `batch_id`
- flush reason
- broker instance ID

They should not log full row payloads or arbitrary source document bodies.

## 10. Suggested Runtime Decomposition

Recommended Rust-side module split inside `rust/crates/ingest/*`:

- `ingest_broker::ingress`
  - internal receive path
  - permit reservation
  - admission accounting
- `ingest_broker::memory`
  - pooled buffers
  - watermark controller
  - namespace pending-byte accounting
- `ingest_broker::coalescer`
  - active bucket map
  - compatibility cuts
  - delay-queue driven flush
- `ingest_broker::scheduler`
  - virtual partition queues
  - WFQ-style deficit accounting
  - whale protection
- `ingest_broker::dispatcher`
  - planner client
  - bounded in-flight dispatch
  - completion fanout
- `ingest_broker::drain`
  - shutdown state machine
  - forced flush
  - optional emergency spill
- `ingest_broker::observability`
  - metrics, tracing, and debug snapshots

The important point is not the exact file names.
The important point is to keep memory control, coalescing, scheduling, and completion fanout as separate units so the broker does not become one large opaque queue handler.

## 11. Rejected Alternatives

### 11.1 Rewriting Multiple Public Batches into One New Canonical `MutationBatch`

Rejected because it would redefine:
- `batch_id`
- `batch_fingerprint`
- per-batch ACK attribution
- request replay identity
- mark framing

The broker may build one `DispatchFrame`, but that frame must remain a private transport unit containing original batches.

### 11.2 Always-Resident Per-Collection Queues for the Full Keyspace

Rejected because 1B-collection scale makes that memory model and scheduling metadata footprint unreasonable.
State should exist only for active collections.

### 11.3 Broker-Local Durable WAL as the Normal Correctness Path

Rejected because it would create a second write authority between admission and the actual write pipeline.
The checked-out architecture already relies on upstream replay plus downstream idempotency rather than on a new broker durability tier.

### 11.4 Exact Global WFQ Heap Across All Namespaces

Rejected because exact fairness state at full keyspace cardinality is too expensive for the stabilizer role.
Approximate WFQ over hashed virtual partitions provides bounded state, predictable runtime cost, and sufficient whale protection for the active design.