Struct Ingest 

pub struct Ingest<R> {
    pub(crate) tx: Arc<[Sender<Job<R>>]>,
    pub(crate) turn: Arc<AtomicU64>,
    pub(crate) rejects: &'static Rejects,
    pub(crate) wal: Option<Arc<Wal>>,
    pub(crate) signal: Signal,
}

The write handle for one signal. R is that signal’s OTLP export request.

Fields

tx: Arc<[Sender<Job<R>>]>

turn: Arc<AtomicU64>

Where the next export that cannot go to shard 0 starts looking. Shared across clones, because the point of it is to spread waiters over shards rather than over handles.

rejects: &'static Rejects

wal: Option<Arc<Wal>>

signal: Signal

Implementations

impl<R> Ingest<R>

fn reserve(&self) -> Option<Permit<'_, Job<R>>>

A slot in the first shard that has one, or None if every shard is full or gone.

First fit from shard 0 rather than round-robin, and that is the whole sharding policy. Round-robin spreads a trickle of exports over every shard, and since each shard owns its own open block, a node doing two exports a second would publish shards nearly-empty blocks every max_block_age instead of one — the small-file explosion section 4 rejects hash sharding for, arrived at from the other direction. First fit keeps a node that is not saturating one flusher behaving exactly as it did with one, and starts using the second shard at the moment the first one’s queue stops draining, which is the moment the consumer’s service time became the curve.

Ordering across shards is not preserved and does not need to be: two exports are two OTLP requests, the spec orders neither against the other, and block timestamps come from the data. Ordering within a shard still is, which is what the carry rule in flusher needs.

impl<R: Message> Ingest<R>

pub async fn submit(&self, req: R) -> Result<(), Rejected>

Enqueue and wait for durability.

What “durable” means here is the one thing Config::wal decides. Without a log this returns once the block containing the request has been fsynced and renamed into place, which is correct and costs a whole max_block_age at the tail. With one it returns once the request is a frame in the log’s page cache, which is section 11’s 2.6 s p99 turned into microseconds and is why the log exists.

impl<R: Message + Default> Ingest<R>

pub fn replay(&self, body: &[u8], seq: u64) -> Result<(), Rejected>

Push one frame recovered from the log back into this signal’s flusher, under the sequence it already has.

Not submit: the frame is in the log already, so re-appending it would number it above every watermark and the block storing it would claim the copy instead of the original — which replays again on the next boot, and the one after that. Nothing waits for the ack either; the client that sent this got its answer before the crash, or gave up long ago.

Blocking, and deliberately: this is called from a spawn_blocking hop at boot, and the bounded channel is the only thing keeping a multi- gigabyte log from being decoded into memory faster than it can be sealed. The two failures are worth telling apart: Rejected::Failed is one frame that will never decode, which is a line in the log and the next frame; Rejected::Closed is the flusher being gone, which means the rest of the replay would go nowhere.

Trait Implementations

impl<R> Clone for Ingest<R>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.