Struct Actor 

struct Actor {

    io: Arc<dyn WalObjectIo>,
    started_regions: Arc<Mutex<HashSet<RegionId>>>,
    withdrawal: Withdrawal,
    catalog: Arc<RwLock<ObjectCatalog>>,
    obsolete_entry_ids: Arc<Mutex<HashMap<RegionId, EntryId>>>,
    wal_holes: Arc<Mutex<HashMap<RegionId, BTreeMap<u64, WalHole>>>>,
    terminal_error: Arc<Mutex<Option<Arc<Error>>>>,
    stopped: Arc<AtomicBool>,
    command_rx: Receiver<Command>,
    append_rx: Receiver<(Vec<Entry>, Sender<Result<AppendBatchResponse>>)>,
    ack_mode: AckMode,
    max_unpersisted_bytes: usize,
    max_unpersisted_age: Duration,
    open_batch: OpenBatch,
    issued_entry_ids: HashMap<RegionId, EntryId>,
    pending: Vec<PendingAppend>,
    sealed: VecDeque<SealedBatch>,
    creates: FuturesUnordered<BoxFuture<'static, (u64, Result<PutResult>)>>,
    stalled: Option<((Vec<Entry>, Sender<Result<AppendBatchResponse>>), Instant)>,
    failing: Option<Failing>,
    durable_waiters: Vec<DurableWaiter>,
    stop: Vec<Sender<Result<()>>>,
    stop_error: Option<Arc<Error>>,
    next_object_seq: Option<u64>,
    epoch: u64,
    last_indexed: ChainLink,
    reclaim_boundary: u64,
    durable_reclaim_boundary: u64,
    unindexed: BTreeSet<u64>,
    failed_releases: BTreeMap<RegionId, EntryId>,
    creating: HashSet<u64>,
    deleting: Arc<DeletingObjects>,
    deletes: FuturesUnordered<BoxFuture<'static, (u64, Result<()>)>>,
    failed_deletes: HashSet<u64>,
    flush_interval: Duration,
}

The actor that owns the open batch and the sealed batches until they are durable.

Sealed batches form a pipeline in sequence order. A batch is created under its sequence as soon as one of MAX_IN_FLIGHT_CREATES slots is free, but it is indexed and acknowledged only once every earlier batch is, so the acknowledged history of a region never has a missing predecessor. Every batch links to its pipeline predecessor, the batch sealed before it, or to the last indexed object when no earlier batch waits, so recovery replays a batch only on the chain of a later object. The outcomes of a create are:

Situation	Sequence	Waiters	Store
created, or identical retry	advances	acknowledged in order	healthy
transient error, durable mode	never reused: a create that reported an error may have written its object	every batch that is not indexed fails, created ones included; retries are assigned new ids	healthy: the next batch links to the last indexed object, so no chain reaches a failed batch
transient error, enqueued mode	unchanged	already acknowledged, except a release, which waits for the repeat	healthy: the create is repeated with the same bytes after CREATE_RETRY_DELAY, which an identical retry accepts
error that requires intervention, enqueued mode	unchanged	already acknowledged, except a release, which fails while its record stays in the backlog; later appends and releases are rejected with a retryable error	failing, see Failing: the lowest pending create is repeated alone with backoff, and the first create that succeeds ends it
object store error, enqueued mode after stop began	never reused	already acknowledged, except a release, which fails with the stopped error	the backlog is dropped and stop reports the error
an object of an earlier epoch holds the sequence, durable mode	never reused	as for a transient error	healthy: that object can never be on the chain
an object of the same or a later epoch holds the sequence, any conflicting object in the enqueued mode, encoding or catalog error	unchanged	every batch that is not indexed fails	poisoned
created at the last representable sequence	cannot advance	acknowledged	poisoned: no later batch can be allocated a sequence

A create still in flight when its batch failed runs to completion and its outcome is ignored: whatever it stored is off the chain. After stop began nothing is admitted and no create starts, except that the enqueued mode uploads its backlog and the durable mode creates one object with the release records of failed batches; creates in flight run to completion and acknowledge if they succeed.

Appends arrive on their own channel. The actor holds one back while the sealed batches leave no room for the two an admission can seal, or at a backlog threshold of the enqueued mode, and does not read the channel meanwhile, so commands such as stop are handled while admission is held back. A failing store reads the channel to reject every append.

The actor also collects garbage. Every batch carries the reclaim boundary current when it was sealed, see ObjectCatalog::reclaim_boundary, which is recomputed after every obsolete, every release and every indexed object and never moves down. The boundary of the last indexed object is durable: every present object below it, indexed or not, is deleted in the background, up to MAX_IN_FLIGHT_DELETES at a time, after an obsolete, a release or an indexing that moved it. An indexed object is unindexed once its delete succeeded; a failed delete is attempted again at the next collection. Nothing is collected once stop began or the store is poisoned, and deletes in flight complete before stop returns.

Fields

io: Arc<dyn WalObjectIo>

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

withdrawal: Withdrawal

catalog: Arc<RwLock<ObjectCatalog>>

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

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

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

stopped: Arc<AtomicBool>

command_rx: Receiver<Command>

append_rx: Receiver<(Vec<Entry>, Sender<Result<AppendBatchResponse>>)>

ack_mode: AckMode

max_unpersisted_bytes: usize

max_unpersisted_age: Duration

open_batch: OpenBatch

issued_entry_ids: HashMap<RegionId, EntryId>

Largest entry id ever handed out per region, whether it became durable or failed. Unlike the accepted ids of the open batch it never moves down.

pending: Vec<PendingAppend>

Waiters of the open batch in the durable mode.

sealed: VecDeque<SealedBatch>

Batches that are not durable yet, in sequence order.

creates: FuturesUnordered<BoxFuture<'static, (u64, Result<PutResult>)>>

Creates in flight, including those of batches that already failed.

stalled: Option<((Vec<Entry>, Sender<Result<AppendBatchResponse>>), Instant)>

The append held back while the sealed batches are full or, in the enqueued mode, the unpersisted backlog is at a threshold, and when it was held back. Later appends wait in their channel.

failing: Option<Failing>

Set while the object store refuses the enqueued backlog with an error that requires intervention.

durable_waiters: Vec<DurableWaiter>

stop: Vec<Sender<Result<()>>>

Callers of stop, answered once nothing is in flight.

stop_error: Option<Arc<Error>>

The failure stop reports in the enqueued mode once an acknowledged backlog was dropped, recorded when it happens.

next_object_seq: Option<u64>

Sequence of the next sealed batch, None once the sequence is exhausted. The open batch assigns its entry ids from it. It is above every sealed batch and never moves back while the store runs.

epoch: u64

Epoch of this instance, carried by every object it writes.

last_indexed: ChainLink

The last indexed object, which is the start object until a batch is indexed. A batch sealed while no earlier batch waits extends it.

reclaim_boundary: u64

The reclaim boundary the next sealed batch carries.

durable_reclaim_boundary: u64

The reclaim boundary of the last indexed object.

unindexed: BTreeSet<u64>

Sequences of objects that may be present but are not indexed: the objects recovery listed off the chain and the batches that failed.

failed_releases: BTreeMap<RegionId, EntryId>

Release records of batches that failed, which the next sealed batch carries again, so a release whose call failed still becomes durable while the store runs. A timer tick seals them when no entry is admitted.

creating: HashSet<u64>

Sequences whose create is in flight.

deleting: Arc<DeletingObjects>

deletes: FuturesUnordered<BoxFuture<'static, (u64, Result<()>)>>

failed_deletes: HashSet<u64>

Deletes that failed since the last collection started.

flush_interval: Duration

Implementations

impl Actor

async fn run(self)

fn handle_append( &mut self, entries: Vec<Entry>, response: Sender<Result<AppendBatchResponse>>, )

fn admit( &mut self, entries: Vec<Entry>, response: Sender<Result<AppendBatchResponse>>, )

Admits entries into the open batch, which assigns their ids under the next object sequence. In the durable mode the caller waits for the object, in the enqueued mode it is answered now.

fn should_hold_back(&self) -> bool

Returns true while an append has to wait for an upload: the sealed batches leave no room for the two an admission can seal, or in the enqueued mode the unpersisted backlog is at a threshold.

fn backlog_at_threshold(&self) -> bool

Returns true once the unpersisted backlog, the open batch and every sealed batch that is not durable, reaches the size or the age threshold.

fn ensure_create_in_flight(&mut self)

Seals the open batch when no create is in flight, so that a stalled append has an upload to wait for.

fn release_stalled(&mut self)

Admits the stalled append once it no longer has to be held back.

fn take_stalled( &mut self, ) -> Option<(Vec<Entry>, Sender<Result<AppendBatchResponse>>)>

Takes the held-back append out on whichever path releases it and records how long it was held back.

fn flush_open_batch(&mut self) -> bool

Seals the open batch as the object next_object_seq and starts its create when a slot is free. Returns whether a batch was sealed.

fn seal(&mut self, releases: Vec<Release>) -> bool

Seals the open batch together with releases and the release records of failed batches as the object next_object_seq, even when the open batch holds no entry, and starts its create when a slot is free. Returns whether a batch was sealed.

A failing store keeps the records of failed batches coalesced instead of sealing them alone behind its backlog on every tick, until stop seals them for its last attempt of the backlog.

fn start_creates(&mut self)

Starts the creates of pending batches in sequence order while fewer than MAX_IN_FLIGHT_CREATES are in flight, counting the creates of batches that already failed. Nothing starts once stop began, except the backlog of the enqueued mode and, once the stop command was handled, the release records handle_stop seals in the durable mode. A failing store only schedules the next attempt once nothing is in flight, see Failing; after stop began it uploads its backlog at once.

fn probe(&mut self)

Attempts the lowest pending batch of a failing store, whose delay elapsed. After stop began start_creates uploads the backlog instead.

fn start_create(&mut self, index: usize, delay: Duration)

Starts the create of the sealed batch at index after delay.

fn on_create_completed(&mut self, object_seq: u64, result: Result<PutResult>)

fn on_create_failed(&mut self, error: Error)

Records the failure of a create the enqueued mode repeats. An error that requires intervention makes the store failing; while it is, every error becomes the one rejected appends carry.

fn enter_failing(&mut self, error: Error)

fn leave_failing(&mut self, reason: &str)

Ends the failing state, if the store is in it, because of reason.

fn settle(&mut self)

Indexes and acknowledges the sealed batches from the front as far as they are created, and starts the creates that a free slot allows.

fn index_front(&mut self) -> bool

Indexes the created object at the front and acknowledges its waiters. Returns false when the catalog rejected it, which poisons the store.

fn resolve_durable_waiters(&mut self)

fn is_durable_through(&self, region_id: RegionId, entry_id: EntryId) -> bool

Returns true when no id of the region at or below entry_id waits to become durable. An id of a batch that failed will never be durable and no longer waits.

fn lowest_pending_entry_id(&self, region_id: RegionId) -> Option<EntryId>

Returns the lowest id of the region that was handed out and is not durable yet. The sealed batches are in sequence order and the open batch is above all of them.

fn roll_back(&mut self, error: Arc<Error>)

Fails every batch that is not indexed after a create failed transiently, including batches already created and batches whose create is in flight. Their sequences are not reused: a create that reported an error may still have written its object. The next batch links to the last indexed object, so no later chain reaches a failed batch. Waiters of a store that was stopped meanwhile learn that instead of the I/O error, like every other entry that never became durable.

fn keep_releases(&mut self, batch: &mut SealedBatch)

Keeps the release records of a batch that failed for the next sealed batch.

fn keep_release(&mut self, release: Release)

fn poison(&mut self, error: Error) -> Arc<Error>

Records error as terminal and fails every waiter that is not acknowledged with it, or with the stopped error if the store was stopped meanwhile. Creates in flight run to completion, but their outcome is ignored: in the durable mode the entries of an object they create were never acknowledged, like those of a crash between creation and acknowledgement; in the enqueued mode the acknowledged backlog is discarded and stop reports it. Returns the recorded error.

fn reset_open_batch(&mut self)

Drops the entries of the open batch; the next admission hands out the same ids again under the sequence the batch is at.

fn fail_unacknowledged(&mut self, error: impl Fn() -> Error)

Fails the waiters of the open batch, the stalled append and the durability waiters.

fn fail_waiters(&mut self, error: impl Fn() -> Error)

Fails the waiters of the open batch and the durability waiters.

fn handle_wait_durable( &mut self, region_id: RegionId, entry_id: EntryId, response: Sender<Result<()>>, )

fn handle_withdraw(&mut self, response: Sender<()>)

Gives up the start object while it is the only object this store knows of: the catalog indexes nothing else, nothing was ever handed an id, no batch, append or release waits, no other object was listed or created, no region’s history started in the store, and the store is neither failing nor terminal. A release seals its own batch without handing an id, so the catalog decides once its object is indexed, and the object it may then leave alone holds a segment. The store then records a terminal error, so nothing is admitted that would link to the object, and deletes it like a collected object; response is answered once that delete settled, or at once when the store keeps the object. A later request waits for the same delete.

fn handle_obsolete( &mut self, region_id: RegionId, entry_id: EntryId, watermark: EntryId, response: Sender<Result<()>>, )

Makes sure no id of the region at or below entry_id is assigned from now on, then records the obsolete watermark of the region; neither is done when the sequence cannot be raised.

fn handle_release( &mut self, region_id: RegionId, full: bool, response: Sender<Result<()>>, )

Moves the watermark of the region up to the release and seals the open batch with a release record of the region; response is answered once that object is durable and indexed. The batch carries the reclaim boundary the watermark allows, so the object makes both durable at once.

The watermark is not clamped to the durable entry id in the enqueued mode: a released region is never replayed, and the record is durable only once every object sealed before it is. A release arrives as a command, so MAX_SEALED_BATCHES does not hold it back; every caller waits for its own object, so releases add at most one batch per caller.

fn update_reclaim_boundary(&mut self)

Raises the reclaim boundary to what the catalog and the obsolete watermarks allow.

fn collect(&mut self)

Starts a collection, which also attempts the deletes that failed since the last one again.

fn start_deletes(&mut self)

Starts deleting the objects below the durable reclaim boundary while fewer than MAX_IN_FLIGHT_DELETES deletes are in flight, skipping objects whose create is in flight or whose delete failed since the collection started.

fn on_delete_completed(&mut self, object_seq: u64, result: Result<()>)

Unindexes the object object_seq and drops its holes once its delete succeeded, then settles its attempt, so a read that the attempt releases sees the outcome in the catalog.

fn raise_sequence_floor( &mut self, region_id: RegionId, entry_id: EntryId, ) -> Result<()>

Moves the sequence of the next object above the object that holds entry_id unless it is there already. The move fails while the open batch has handed out ids under the next sequence. A floor that does not fit an entry id poisons the store like an exhausted sequence.

fn handle_stop(&mut self, response: Sender<Result<()>>)

Begins stopping. Nothing is admitted from now on and the append held back is refused; the durable mode drops the open batch and the batches whose create has not started, and seals the release records of failed and dropped batches once more; the enqueued mode seals its backlog and uploads it, a failing store included. Stop is answered by finish_stop once nothing is in flight.

fn finish_stop(&mut self) -> bool

Answers the callers of stop once every sealed batch is settled and every create and delete has completed. Returns true when the actor is done.

fn is_stopped(&self) -> bool

Trait Implementations

impl Drop for Actor

fn drop(&mut self)

Settles the deletes the actor still had in flight, so no read waits for an attempt that will never complete, and releases the append still held back, whose caller is gone or about to be.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.