Struct Actor 

struct Actor {

    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_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>>)>,
    durable_waiters: Vec<DurableWaiter>,
    stop: Vec<Sender<Result<()>>>,
    stop_error: Option<Arc<Error>>,
    next_object_seq: Option<u64>,
    epoch: u64,
    last_indexed: ChainLink,
    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	healthy: the create is repeated with the same bytes after CREATE_RETRY_DELAY, which an identical retry accepts
transient error, enqueued mode after stop began	never reused	already acknowledged	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; creates in flight run to completion and acknowledge if they succeed.

Appends arrive on their own channel, which the actor does not read while MAX_SEALED_BATCHES batches wait or an append is held back at a backlog threshold of the enqueued mode, so commands such as stop are handled while admission is held back.

Fields

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

The append held back in the enqueued mode while the unpersisted backlog is at a threshold. Later appends wait in their channel.

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.

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 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 the backlog is below the thresholds and the sealed batches leave room for the two an admission can seal.

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

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

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 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 handle_wait_durable( &mut self, region_id: RegionId, entry_id: EntryId, response: Sender<Result<()>>, )

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 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; the durable mode drops the open batch and the batches whose create has not started, the enqueued mode seals its backlog so it is uploaded. 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 has completed. Returns true when the actor is done.

fn is_stopped(&self) -> bool