Struct ObjectStoreLogStore 

pub struct ObjectStoreLogStore {

    prefix: String,
    storage: String,
    ack_mode: AckMode,
    on_corrupted_segment: CorruptedSegmentAction,
    accept_missing_history: bool,
    found_empty: bool,
    history_end: u64,
    started_regions: Arc<Mutex<HashSet<RegionId>>>,
    io: Arc<dyn WalObjectIo>,
    catalog: Arc<RwLock<ObjectCatalog>>,
    obsolete_entry_ids: Arc<Mutex<HashMap<RegionId, EntryId>>>,
    wal_holes: Arc<Mutex<HashMap<RegionId, BTreeMap<u64, WalHole>>>>,
    deleting: Arc<DeletingObjects>,
    read_prefetch_budget: Arc<Semaphore>,
    terminal_error: Arc<Mutex<Option<Arc<Error>>>>,
    stopped: Arc<AtomicBool>,
    command_tx: Sender<Command>,
    append_tx: Sender<(Vec<Entry>, Sender<Result<AppendBatchResponse>>)>,
}

A log store over the immutable WAL objects under one prefix.

Appends are admitted into an open batch. A background actor seals the batch when it reaches the size limit or the flush interval elapses and creates the object under the next sequence while it keeps admitting entries into the next batch; up to MAX_IN_FLIGHT_CREATES creates run at a time. Objects are indexed in the catalog in sequence order. In the durable acknowledgement mode an append returns once its object is durable and indexed, so an acknowledged entry never has a missing predecessor; in the enqueued mode it returns on admission and the object is created in the background. While MAX_SEALED_BATCHES batches wait to become durable no append is admitted, and in the enqueued mode appends are rejected while the object store refuses the backlog with an error that requires intervention. A segment that a read cannot decode is handled as on_corrupted_segment says. Objects below the reclaim boundary of the last indexed object are deleted in the background. A release of a region seals an object with a release record of the region at once and returns, in both modes, once that object is durable and indexed.

Fields

prefix: String

storage: String

The storage the prefix is read from, as configured, for messages.

ack_mode: AckMode

on_corrupted_segment: CorruptedSegmentAction

accept_missing_history: bool

found_empty: bool

Whether the prefix held no object when the store opened.

history_end: u64

The sequence above the newest object holding entries that recovery indexed, zero when it indexed none.

started_regions: Arc<Mutex<HashSet<RegionId>>>

io: Arc<dyn WalObjectIo>

catalog: Arc<RwLock<ObjectCatalog>>

obsolete_entry_ids: Arc<Mutex<HashMap<RegionId, EntryId>>>

wal_holes: Arc<Mutex<HashMap<RegionId, BTreeMap<u64, WalHole>>>>

Segments that reads skipped, per region and by object sequence.

deleting: Arc<DeletingObjects>

Objects with a delete in flight, which a read consults.

read_prefetch_budget: Arc<Semaphore>

Bytes the reads may fetch ahead, see READ_PREFETCH_BYTES.

terminal_error: Arc<Mutex<Option<Arc<Error>>>>

stopped: Arc<AtomicBool>

command_tx: Sender<Command>

append_tx: Sender<(Vec<Entry>, Sender<Result<AppendBatchResponse>>)>

Implementations

impl ObjectStoreLogStore

pub async fn try_new( object_store: ObjectStore, config: &ObjectStoreWalConfig, node_id: u64, generation: u64, ) -> Result<Arc<Self>>

Builds the store under the node and generation prefix derived from config, recovering the catalog from the objects that already exist and writing the object that starts the epoch of this instance. Before it recovers, it fails unless the object store refuses to overwrite an object on a conditional create. Recovery fails on the first corrupted or conflicting object.

async fn open( io: Arc<dyn WalObjectIo>, config: &ObjectStoreWalConfig, prefix: String, ) -> Result<Arc<Self>>

pub(crate) fn durable_entry_id(&self, provider: &Provider) -> Result<EntryId>

Returns the largest entry id of the provider’s region whose object is durable and indexed, or zero for a region without such entries. In the enqueued acknowledgement mode an entry id that an append returned stays above this value until the object holding it is created.

fn durable_entry_id_of(&self, region_id: RegionId) -> EntryId

pub fn wal_holes(&self, provider: &Provider) -> Result<Vec<WalHole>>

Returns the segments of the provider’s region that reads of this store skipped because they did not decode, ordered by object sequence. Holes are kept in memory only: a store opened later on the same prefix records them again when its reads meet the segments.

fn region_of(&self, provider: &Provider) -> Result<RegionId>

Returns the region of provider, which must select this store’s prefix.

fn check_region(&self, provider: &Provider, region_id: RegionId) -> Result<()>

fn check_terminal(&self) -> Result<()>

async fn release(&self, region_id: RegionId, full: bool) -> Result<()>

Releases the region: its watermark moves up to the release in memory and a release record of it goes into an object sealed at once, and the call returns once that object is durable and indexed, in both acknowledgement modes, or with the error that failed it. A full release reaches EntryId::MAX; any other the largest id the store handed out for the region, which every entry of the region it holds is at or below. When the object fails, the next sealed batch carries the record again. Recovery seeds the watermark from the record, so a store opened later collects the objects of a region that is never opened again.

fn missing_history( &self, region_id: RegionId, entry_id: EntryId, ) -> Option<String>

Returns why the prefix cannot hold the WAL history of region_id through the watermark entry_id, or None when it can.

Two invariants of a prefix decide it. First, a prefix that any store opened is never empty afterwards, unless that store withdrew its start object, see withdraw_start: every open writes its start object before it admits an append, and collection deletes only objects below the reclaim boundary of the last indexed object, which carries a boundary at or below its own sequence (a larger one fails its header check), so the newest indexed object always stays. Second, a watermark a region persisted names an object at or below the newest object holding entries that any later recovery of the prefix indexes. The entry it names was durable and indexed when the watermark was persisted, since an append of the durable mode returns only then and Mito waits for durability before it publishes a watermark in the enqueued mode. No durable boundary passes the newest object holding entries: a batch carries at most the sequence after the last indexed object, which is below the batch, and a start object carries the boundary recovery found, or its own sequence on a chain without segments. So that object stays, and recovery indexes it on the chain of the tip. The segment of release records counts as entries: it never holds the reclaim boundary, so an object holding only releases can be the one that collection keeps while it deletes every older object. Objects without segments, such as the start objects of opens this check refused, do not move it.

So the history is missing when this store found the prefix empty, or when the watermark names an object above the newest one holding entries that it found, unless the region’s history starts in this store or the catalog holds the region’s entries through the watermark: those of a region written since the store opened.

async fn withdraw_start(&self)

Deletes the start object of a store that found its prefix empty and whose open of an existing region is refused, so that the next open finds the prefix empty again and refuses the region as well, instead of passing a region that never flushed. The actor gives the object up only while the store has admitted nothing, has no batch, holds no other object and no region’s history started in it, and from then on the store accepts nothing, since a later object would link to the deleted one. The actor owns the delete like any collection, so stop waits for it and a cancelled caller does not abandon it, and every refusal returns only once it settled. A failed delete, or a crash before the delete completes, leaves the start object behind, and the next open passes a region that never flushed.

Trait Implementations

impl Debug for ObjectStoreLogStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl LogStore for ObjectStoreLogStore

fn stop<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Stops the store. Creates in flight run to completion and acknowledge their entries if they succeed; in the enqueued mode the remaining backlog is uploaded first, and a failure of that upload is returned.

fn read<'life0, 'life1, 'async_trait>( &'life0 self, provider: &'life1 Provider, entry_id: EntryId, ) -> Pin<Box<dyn Future<Output = Result<SendableEntryStream<'static, Entry, Error>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Reads the provider’s region from entry_id, hiding obsolete entries. The catalog locates each segment, so a caller-supplied index is unnecessary. An object collected after the read listed it is skipped. Otherwise a segment that does not decode fails the read or is skipped and recorded as a hole, as on_corrupted_segment says; a failed fetch always fails the read. Up to READ_PREFETCH_DEPTH segments are fetched ahead while READ_PREFETCH_BYTES allows, and each outcome is handled in sequence order.

fn obsolete<'life0, 'life1, 'async_trait>( &'life0 self, provider: &'life1 Provider, region_id: RegionId, entry_id: EntryId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Moves the obsolete watermark of the region up to entry_id. In the enqueued acknowledgement mode the watermark never passes the durable entry id: an entry that is not durable yet is replayed after a crash, and hiding it would skip it.

Every id the region is assigned from now on is greater than entry_id, whatever watermark is recorded: the sequence of the next object is raised above the object entry_id names unless it is there already, so a watermark assigned under another prefix, or one whose object this prefix no longer holds, is never passed by a new id. Zero names no object and needs no floor. Before Mito opens a region that writes this WAL from its persisted state, it calls check_region_history, which refuses a watermark naming an object at or above the next sequence the store found unless accept_missing_history is set, so the floor moves for such a region only once that loss was accepted.

The watermark and the floor are applied together by the actor: a call cancelled before its command is queued publishes neither, while a command already queued still applies both. While the open batch holds ids under the next sequence the floor cannot move, and the call fails with Error::WalObjectSequenceUnsettled and records neither. The watermark is kept in memory; the caller re-establishes it after a restart unless it released the region, see release_region. In the enqueued mode an entry_id this store handed out needs no floor: it is never handed out again while the store runs.

fn obsolete_all<'life0, 'life1, 'async_trait>( &'life0 self, provider: &'life1 Provider, region_id: RegionId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Hides every entry of the region and lets collection delete its objects, by a release in full that is durable once this returns.

fn release_region<'life0, 'life1, 'async_trait>( &'life0 self, provider: &'life1 Provider, region_id: RegionId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Releases the region through the largest id handed out for it, durably once this returns. Should the region id be used again, its new entries have greater ids and stay readable.

fn wait_durable<'life0, 'life1, 'async_trait>( &'life0 self, provider: &'life1 Provider, entry_id: EntryId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Waits until every entry of the provider’s region with an id at or below entry_id is durable and indexed. Returns at once in the durable acknowledgement mode, where a caller only holds durable ids.

fn check_region_history<'life0, 'life1, 'async_trait>( &'life0 self, provider: &'life1 Provider, entry_id: EntryId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Fails with Error::MissingWalHistory when the prefix cannot hold the history of the provider’s region through entry_id, unless accept_missing_history is set: then the region’s writes that were not flushed are lost, which is logged and counted.

fn start_region_history(&self, provider: &Provider) -> Result<()>

Records the provider’s region as one whose history starts in this store, which is kept in memory: a later store on the prefix judges the region by its watermark and the objects it finds.

type Error = Error

fn append_batch<'life0, 'async_trait>( &'life0 self, entries: Vec<Entry>, ) -> Pin<Box<dyn Future<Output = Result<AppendBatchResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Appends a batch of entries and returns a response containing a map where the key is a region id while the value is the id of the last successfully written entry of the region.

fn create_namespace<'life0, 'life1, 'async_trait>( &'life0 self, ns: &'life1 Provider, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Creates a new Namespace from the given ref.

fn delete_namespace<'life0, 'life1, 'async_trait>( &'life0 self, ns: &'life1 Provider, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Deletes an existing Namespace specified by the given ref.

fn list_namespaces<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<Provider>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Lists all existing namespaces.

fn entry( &self, data: Vec<u8>, entry_id: EntryId, region_id: RegionId, provider: &Provider, ) -> Result<Entry>

Makes an entry instance of the associated Entry type

fn latest_entry_id(&self, provider: &Provider) -> Result<EntryId>

Returns the latest entry id in the log store.