Struct Tier 

pub struct Tier<R, St, S> { /* private fields */ }

Buffering tier over a pluggable Store.

Records ingest into the wall-clock-aligned window align(meta.end) (aligned to the policy’s every); when the policy fires, stored windows drain downstream oldest-first, each removed from the store only after the downstream sink accepted it — a transient outage of the terminal sink does not lose data held in this tier.

One window key can drain downstream more than once: the max_records valve fires mid-window and later records re-open the same key, and a failed drain is retried after the window has grown. Both deliveries carry the same WindowMeta, so a terminal sink that keys storage by window start alone would overwrite the earlier chunk — key by content as well, as HfSink (feature huggingface) does.

A window whose delivery keeps failing (for example a record the terminal sink deterministically rejects) does not stall the pipeline: each drain pass attempts every closed window oldest-first, retains the failed ones for the next pass, and surfaces the first error. Such a poisoned window is retried on every pass and retained in the store indefinitely — remove its segment by hand (for JsonlStore, delete the window’s .jsonl file) if it must be discarded.

The policy is checked on each ingest (collector ticks are frequent compared to flush windows, so no extra timer task is needed); flush force-drains. Whatever a previous run left in the store is replayed downstream on the first ingest or flush call, with WindowMeta reconstructed from the window key alone — replayed windows land at the same storage path (idempotent).

Implementations

impl<R, St, S> Tier<R, St, S>

pub fn new(store: St, policy: FlushPolicy, inner: S) -> Self

Create a tier holding records in store until policy fires.

pub fn with_pipeline_name(self, name: impl Into<String>) -> Self

Override the pipeline name used in drained WindowMeta (default: the first live meta seen, then the store’s pipeline_hint).

pub fn inner(&self) -> &S

Access the wrapped sink.

Trait Implementations

impl<R, St, S> Sink<R> for Tier<R, St, S>

where R: Send + 'static, St: Store<R>, S: Sink<R>,

type Error = TierError<<St as Store<R>>::Error, <S as Sink<R>>::Error>

Concrete error type (a thiserror enum, not a boxed error).

async fn ingest( &mut self, meta: &WindowMeta, records: Vec<R>, ) -> Result<(), Self::Error>

Hand records to this layer. A buffering layer may hold them; a terminal sink ships them immediately.

async fn flush(&mut self) -> Result<(), Self::Error>

Force-drain this layer and everything downstream (shutdown, final flush, startup recovery).