Struct ObjectStoreLogStore 

pub struct ObjectStoreLogStore {
    prefix: String,
    ack_mode: AckMode,
    io: Arc<dyn WalObjectIo>,
    catalog: Arc<RwLock<ObjectCatalog>>,
    obsolete_entry_ids: Arc<Mutex<HashMap<RegionId, EntryId>>>,
    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.

Fields

prefix: String

ack_mode: AckMode

io: Arc<dyn WalObjectIo>

catalog: Arc<RwLock<ObjectCatalog>>

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

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. 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

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<()>

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, _index: Option<WalIndex>, ) -> 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.

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.

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. 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 in memory. The caller re-establishes the watermark after a restart.

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.

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.