> ## Documentation Index
> Fetch the complete documentation index at: https://opendata.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Buffer Architecture
> How Buffer works
Buffer is a library with two main components: **multiple producers** and
**a single consumer**. Producers and the consumer communicate exclusively
through a queue manifest in object storage — there is no direct network path
between them and no stateful broker to operate.
{`
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒ buffer architecture ▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒┃
┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
┃ ┃
┃ ╔═Producers══════════════════════════════════════════════════╗ ┃
┃ ║ ║░ ┃
┃ ║ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ║░ ┃
┃ ║ │ Producer 1 │ │ Producer 2 │ │ Producer N │ ║░ ┃
┃ ║ │ q-producer │ │ q-producer │ │ q-producer │ ║░ ┃
┃ ║ └───────┬──────┘ └──────────────┘ └───────┬──────┘ ║░ ┃
┃ ║ │ │ ║░ ┃
┃ ╚══════════╪═══════════════════════════════════════╪═════════╝░ ┃
┃ ░░░░░░░░░░│░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░│░░░░░░░░░░░ ┃
┃ └flush enqueue ┃
┃ │ │ ┃
┃ ╔═Object Store Bucket═════════════════════════╪══════════════╗ ┃
┃ ║ │ │ ║░ ┃
┃ ║ │ │ ║░ ┃
┃ ║ ┏━data━━━━━━━▼━━━━━━━━━━━┓ ┏━queue━━━━━▼━━━━━━━━━━┓ ║░ ┃
┃ ║ ┃ ┃ ┃ ┃ ║░ ┃
┃ ║ ┃ ┃ ┃ ╔════════════════╗ ┃ ║░ ┃
┃ ║ ┃ 01J5T4R3.batch ┃ ┃ ║ ║ ┃ ║░ ┃
┃ ║ ┃ 01J5T4R7.batch ┃ ┃ ║ q-manifest ║ ┃ ║░ ┃
┃ ║ ┃ 01J5T4RB.batch ┃ ┃ ║ ║ ┃ ║░ ┃
┃ ║ ┃ ┃ ┃ ╚════════════════╝ ┃ ║░ ┃
┃ ║ ┃ ┃ ┃ ┃ ║░ ┃
┃ ║ ┃ ┃ ┃ ┃ ║░ ┃
┃ ║ ┗━━━━━━━━━━━━┬━━━━━━━━━━━┛ ┗━━━━━━━━━━━▲━━━━━━━━━━┛ ║░ ┃
┃ ║ │ │ ║░ ┃
┃ ║ │ │ ║░ ┃
┃ ╚═══════════════╪═════════════════════════════╪══════════════╝░ ┃
┃ ░░░░░░░░░░░░░░░│░░░░░░░░░░░░░░░░░░░░░░░░░░░░░│░░░░░░░░░░░░░░░░ ┃
┃ │ ┌───────poll─────────┘ ┃
┃ read │ ┃
┃ │ │ ┃
┃ ╔═Writer═╪════════╪═════════╗ ┃
┃ ║ │ │ ║░ ┃
┃ ║ │┌───────┴──────┐ ║░ ╔════════════════╗ ┃
┃ ║ ││ Consumer │ ║░ ║ ║ ┃
┃ ║ └▶ q-consumer ├──write──▶ Database ║ ┃
┃ ║ └──────────────┘ ║░ ║ ║ ┃
┃ ║ ║░ ╚════════════════╝ ┃
┃ ╚═══════════════════════════╝░ ┃
┃ ░░░░░░░░░░░░░░░░░░░░░░░░░░░░░ ┃
┃ ┃
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
`}
## Producers
Producers accept arbitrary byte entries from callers, buffer them in
memory, and periodically flush them as binary data batches to object
storage. Each producer exposes a minimal API:
```rust theme={null}
pub async fn produce(
&self,
entries: Vec,
metadata: Bytes,
) -> Result
```
Flushes are triggered either by elapsed time (default: 100 ms) or by batch
size (default: 64 MiB). Data batches are binary and optionally compressed.
Each batch is named with a [ULID](https://github.com/ulid/spec), which
encodes a millisecond-precision timestamp. Once a batch is flushed,
producers append its location and metadata to the queue.
The queue is a single binary manifest in object storage that tracks the
locations and metadata of data batches in append order. Producers interact
with it through a queue producer, which uses compare-and-swap (CAS) writes
to ensure that concurrent producers do not overwrite each other's entries:
an append succeeds only if the manifest has not been modified since it was
read. On conflict, the queue producer re-reads the manifest and retries.
Additionally, the queue producer assigns consecutive sequence numbers to
queue entries.
{`
╔═queue manifest═══════════════════════════════════════════╗
║ ║░
║ seq │ location │ metadata ║░
║ ─────┼───────────────────────┼───────────────── ║░
║ 001 │ ingest/01J5T4R3.batch │ src=app-1 n=128 ║░
║ 002 │ ingest/01J5T4R7.batch │ src=app-2 n=64 ║░
║ 003 │ ingest/01J5T4RB.batch │ src=app-1 n=256 ║░
║ 004 │ ingest/01J5T4RF.batch │ src=app-3 n=192 ║░
║ 005 │ ingest/01J5T4RJ.batch │ src=app-2 n=512 ║░
║ … │ … │ … ║░
║ ║░
╚══════════════════════════════════════════════════════════╝░
░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
`}
The manifest format is designed for append efficiency. Existing entries
are never deserialized during append, so appending a new entry is O(1) in
the length of the queue.
The `WriteHandle` returned by `produce()` contains a `DurabilityWatcher`.
Callers can `await_durable()` to block until the entries have been
persisted to object storage and enqueued in the manifest. This avoids data
loss during failover, but it does not provide read-your-own-writes
consistency — durability refers to the data batch, not to the database
write.
## Consumer
The consumer is the read-side counterpart to the producers. It iterates
over data batches in object storage and delivers them to the database in
the order recorded in the queue:
```rust theme={null}
pub async fn next_batch(&mut self) -> Result>
pub async fn ack(&mut self, sequence: u64) -> Result<()>
```
A call to `next_batch()` reads the next location in the queue manifest,
fetches the data batch from object storage, and returns the data,
metadata, and a unique sequence number to the caller. Sequence numbers
are assigned by the queue producer at ingestion time in increasing order
without gaps, and they are used both for acknowledgement and for tracking
progress. Acknowledgements must be in order: acknowledging a sequence
number that is not immediately after the last one is rejected, ensuring
that no batch is silently skipped.
While multiple producers can write to a queue manifest, only one consumer
can read from it. This matches OpenData's single-writer paradigm and
prevents zombie consumers — surviving after an infrastructure failure —
from acknowledging batches and making them invisible to the active
consumer. Single-consumer exclusivity is enforced by an epoch stored in
the queue manifest: when a consumer starts, it reads the epoch,
increments it, and writes it back. Reads with a stale epoch are rejected,
so any zombie consumer is fenced as soon as a new one takes over.
The consumer is also responsible for cleaning up processed data batches.
At startup and every 100 acknowledged batches, all entries for
acknowledged batches are dequeued from the manifest. A dedicated garbage
collector task then deletes the corresponding data batches from object
storage.
The consumer API supports exactly-once delivery. To achieve it, the
writer atomically persists the batch data and the sequence number from
`next_batch()` to the database. If the atomic write succeeds but a failure
occurs before `ack()` returns, a new writer can construct a new consumer
with that last persisted sequence number so it resumes right after it. The
new consumer also bumps the epoch, fencing the old one and preventing any
duplicate writes.
## Read-ahead and batched acks
The serial `next_batch` / `ack` API caps a consumer at one manifest read
and one ack write per batch. Fetching a batch is an I/O wait; decoding it
is CPU work; and writing it to another system is yet more I/O work. Doing
everything on one task in a tight loop wastes both the
object-store bandwidth and the machine's cores. For high-throughput
consumers, `Consumer` exposes a parallel-fetch path:
```rust theme={null}
pub async fn next_descriptors(&mut self, max: usize)
-> Result>;
pub async fn fetch_descriptor(&mut self, d: BatchDescriptor)
-> Result;
pub fn fetch_handle(&self) -> ConsumerFetchHandle;
pub async fn ack_through(&mut self, sequence: u64) -> Result<()>;
```
The design batches the expensive low-volume operations (manifest read,
ack) and parallelizes the high-volume operations (fetch,
decode). The serial `next_batch` / `ack` API still works without changes:
`next_batch` is a thin wrapper over `next_descriptors(1)` plus
`fetch_descriptor`, and `ack` keeps its amortize-every-100-calls behavior.
### How the split works
`next_descriptors(max)` reads the manifest once and returns up to `max`
contiguous descriptors past an internal read-ahead cursor. It does no
object-store fetch and does not mutate the durable ack frontier.
`ConsumerFetchHandle` is a cheap, cloneable handle obtained from
`fetch_handle()`. It holds an `Arc` to the object store and the manifest
path. Its single method, `fetch(&self, descriptor) -> Result`,
fetches and decodes one batch, touches no `Consumer` cursor, and is safe
to call concurrently from many tasks. Cloning the handle is O(1).
`ack_through(sequence)` advances the durable ack frontier through
`sequence` in one `dequeue` write, no matter how far the frontier moves.
The dequeue runs first; in-memory state and the `buffer.acks` counter
advance only on success, so a fence or storage error leaves the
consumer's state untouched and the call is safe to retry.
### Pipelined consumer pattern
One task owns the `Consumer` and the manifest; N fetch workers each hold
a clone of the handle and pull descriptors off a bounded channel:
```rust theme={null}
let consumer = Consumer::new(config, last_acked).await?;
let fetcher = consumer.fetch_handle();
// N fetch workers, each holding a clone of the handle:
for _ in 0..fetch_concurrency {
let f = fetcher.clone();
let rx = descriptor_rx.clone();
tokio::spawn(async move {
while let Some(d) = rx.recv().await {
let batch = f.fetch(d).await?;
// ...hand off to a decode / sink stage...
}
Ok::<(), Error>(())
});
}
// Owner task: pull a run of descriptors, fan them out, ack the
// contiguous high watermark.
loop {
let descriptors = consumer.next_descriptors(K).await?;
for d in descriptors { descriptor_tx.send(d).await?; }
// ...wait for the contiguous-completed high watermark, then:
consumer.ack_through(high).await?;
}
```
Owner and workers run in parallel because their state is disjoint: two
concurrent fetches against distinct descriptors do not contend on the manifest.
### Descriptor handout contract
`next_descriptors` advances an internal `last_handed_out_sequence`
*before* the caller has fetched the objects. Read-ahead is the point.
That places one obligation on the caller:
`ack_through(n)` does not verify that anything below `n` was processed.
It unconditionally advances the durable frontier to `n`, rejecting only
`n <= last_acked_sequence`. The caller must track which sequences have
actually been processed and only call `ack_through` with the highest
fully-processed *contiguous* sequence. Fetches may complete out of order;
ack the contiguous-complete watermark, never past a still-pending
sequence.
If a descriptor is permanently lost (a worker panics, a channel send is
dropped), the frontier stalls at that sequence. Recovery is process
restart: `Consumer::new(config, last_acked_sequence)` re-emits everything
the durable frontier has not acked.
### Fencing
`next_descriptors` returns `Error::Fenced` on the next manifest read
after fencing. A worker holding a stale handle keeps fetching
successfully against object storage; the manifest owner learns of the
fence and propagates it, typically by closing the descriptor channel.
`ack_through` against a fenced manifest returns `Error::Fenced` and
leaves state untouched, so a fenced consumer cannot advance the durable
frontier.
### When to use which API
The serial `next_batch` / `ack` API stays the right choice for
low-throughput single-task consumers (the timeseries and vector buffer
consumers use it) and for backfill tooling where a tight per-batch ack
loop is simpler than the pipeline.
Reach for the read-ahead path when consumer throughput is bounded by
manifest round-trips (one `next_batch` cycle per batch caps throughput
at `1 / manifest_RTT`), when fetch I/O and decode CPU need to overlap
across batches, or when the workload is bursty enough that an in-flight
bytes ceiling fits better than per-batch flow control.
The opendata-contrib [generic ingest runtime](https://github.com/opendata-oss/opendata-contrib/blob/main/rfcs/0002-generic-ingest-runtime.md) is one pipelined consumer
built on these primitives. Full design rationale, the descriptor handout
contract in detail, and the failure-mode catalog live in [RFC 0003:
Consumer Read-Ahead and `ack_through`](https://github.com/opendata-oss/opendata/blob/main/buffer/rfcs/0003-consumer-read-ahead.md).
## Usage example
For an example of how Buffer can be integrated into an ingestion pipeline see
[ingest into Timeseries](/docs/timeseries/ingest).