Struct Consumer 

pub struct Consumer { /* private fields */ }

Reads batches of ingested entries from object storage via a queue consumer.

The consumer iterates over entries in the queue manifest in ingestion order, fetches the corresponding data batches from object storage, and makes them available to the caller. Epoch-based fencing ensures only a single active consumer processes entries at any time.

Implementations

impl Consumer

pub async fn new( config: ConsumerConfig, last_acked_sequence: Option<u64>, ) -> Result<Self>

Create a new consumer from the given configuration.

Initializes the queue consumer (fencing any previous instance) and spawns the garbage collector. If last_acked_sequence is Some(seq), the consumer resumes after that sequence; if None, it discovers the first available entry.

pub async fn with_object_store( config: ConsumerConfig, object_store: Arc<dyn ObjectStore>, last_acked_sequence: Option<u64>, ) -> Result<Self>

pub async fn next_batch(&mut self) -> Result<Option<ConsumedBatch>>

Read the next data batch from object storage.

Serial fetch of the next batch: peeks the next manifest entry past the last handed-out / fetched sequence, fetches the corresponding object, and returns it. Returns None if no entry is available. May be called repeatedly to walk successive batches; the read cursor is independent of the ack frontier.

Cancellation-safe and fetch-failure-safe. The cursor (last_handed_out_sequence) advances only after a successful fetch. If the fetch fails, or if the future is dropped mid-fetch, the cursor is unchanged and a subsequent next_batch re-fetches the same entry.

This is written as an inline peek-fetch-advance rather than a wrapper over Consumer::next_descriptors + Consumer::fetch_descriptor because next_descriptors advances the cursor at handout time (a handed-out descriptor is not reissued within a process; see the Descriptor Handout Contract). Inlining keeps the cursor advance after the fetch await, so a failed or dropped fetch leaves the cursor untouched — the behavior next_batch callers expect.

pub async fn next_descriptors( &mut self, max: usize, ) -> Result<Vec<BatchDescriptor>>

Read the manifest once and return up to max contiguous BatchDescriptors past the consumer’s read-ahead cursor.

Does not perform any object-store GET. Does not mutate the durable ack frontier. Advances the in-memory read-ahead cursor (last_handed_out_sequence) by the number of descriptors returned.

Returns an empty Vec if no new entries are available; returns Err(Error::Fenced) if the consumer’s epoch no longer matches the manifest’s.

Caller contract: once a descriptor is returned, the caller is responsible for either fetching and processing it or accepting that it will be re-handed-out only via process restart (Consumer::new with a last_acked_sequence argument). Lost descriptors are not reissued within a process. See RFC 0003 “Descriptor Handout Contract” for the full rules.

pub fn fetch_handle(&self) -> ConsumerFetchHandle

Construct a cloneable fetch handle. O(1); each call returns a fresh handle, and the handle itself implements Clone for further duplication into worker tasks.

pub async fn fetch_descriptor( &mut self, descriptor: BatchDescriptor, ) -> Result<ConsumedBatch>

Fetch and decode a single batch via the consumer’s serial wrapper. Equivalent to self.fetch_handle().fetch(descriptor), but additionally maintains the consumer’s serial lag cursor (last_fetched_sequence + the consumer_lag_seconds gauge) used by the legacy next_batch path.

Parallel-fetch callers should use Consumer::fetch_handle directly; the runtime owns its own per-stage latency histograms (RFC 0002).

pub async fn ack(&mut self, sequence: u64) -> Result<()>

Acknowledge that the batch with the given sequence number has been processed.

Acks must be in order — the sequence must immediately follow the last acked sequence, otherwise an error is returned. To amortize manifest writes, the consumer only calls dequeue() on the queue consumer every 100 acks.

pub async fn ack_through(&mut self, sequence: u64) -> Result<()>

Advance the durable ack frontier through (and including) sequence. Performs dequeue(sequence) against the manifest first, then updates in-memory state on success.

On error (storage, fence), last_acked_sequence and the buffer.acks counter remain at their pre-call values. A retry against the same sequence is safe.

Errors if sequence <= last_acked_sequence (the frontier is monotonic).

pub async fn flush(&mut self) -> Result<()>

Flush any pending acks by dequeueing up to the last acked sequence.

pub async fn close(self) -> Result<()>

Flush pending acks, shut down the garbage collector, and consume the handle.

pub fn len(&self) -> usize

Return the number of entries in the queue as of the last manifest read or write.

pub fn is_empty(&self) -> bool

Return true if the queue had no entries as of the last manifest read or write.

pub fn conflict_rate(&self) -> f64

Return the percentage of manifest writes that encountered a conflict.