Struct mz_storage_controller::collection_mgmt::DifferentialWriteTask

struct DifferentialWriteTask<T, R>
where
    T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation,
    R: FnMut() -> Pin<Box<dyn Future<Output = ReadHandle<SourceData, (), T, Diff>> + Send>> + Send + 'static,
{
    id: GlobalId,
    write_handle: WriteHandle<SourceData, (), T, Diff>,
    read_handle_fn: R,
    read_only: bool,
    now: NowFn,
    upper_tick_interval: Interval,
    cmd_rx: UnboundedReceiver<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>,
    shutdown_rx: Receiver<()>,
    desired: Vec<(Row, i64)>,
    to_write: Vec<(Row, i64)>,
    current_upper: T,
}

A task that will make it so that the state in persist matches the desired state and continuously bump the upper for the specified collection.

NOTE: This implementation is a bit clunky, and could be optimized by not keeping all of desired in memory (see commend below). It is meant to showcase the general approach.

Fields

id: GlobalId

The collection that we are writing to.

write_handle: WriteHandle<SourceData, (), T, Diff>

read_handle_fn: R

For getting a ReadHandle to sync our state to persist contents.

read_only: bool

now: NowFn

upper_tick_interval: Interval

In the absence of updates, we regularly bump the upper to “now”, on this interval. This makes it so the collection remains readable at recent timestamps.

cmd_rx: UnboundedReceiver<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>

Receiver for write commands. These change our desired state.

shutdown_rx: Receiver<()>

We have to shut down when receiving from this.

desired: Vec<(Row, i64)>

The contents of the collection as it should be according to whoever is driving us around.

to_write: Vec<(Row, i64)>

Updates that we have to write when next writing to persist. This is determined by looking at what is desired and what is in persist.

current_upper: T

Current upper of the persist shard. We keep track of this so that we realize when someone else writes to the shard, in which case we have to update our state of the world, that is update our to_write based on desired and the contents of the persist shard.

Implementations

impl<T, R> DifferentialWriteTask<T, R>

where T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation, R: FnMut() -> Pin<Box<dyn Future<Output = ReadHandle<SourceData, (), T, Diff>> + Send>> + Send + Sync + 'static,

fn spawn( id: GlobalId, write_handle: WriteHandle<SourceData, (), T, Diff>, read_handle_fn: R, read_only: bool, now: NowFn, introspection_config: DifferentialIntrospectionConfig<T>, ) -> (UnboundedSender<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>, AbortOnDropHandle<()>, Sender<()>)

Spawns a DifferentialWriteTask in an mz_ore::task and returns handles for interacting with it.

async fn prepare( &self, introspection_config: DifferentialIntrospectionConfig<T>, )

Does any work that is required before this background task starts writing to the given introspection collection.

This might include consolidation, deleting older entries or seeding in-memory state of, say, scrapers, with current collection contents.

async fn run(self) -> ControlFlow<String>

async fn tick_upper(&mut self) -> ControlFlow<String>

fn handle_shutdown(&mut self)

async fn handle_updates( &mut self, batch: &mut Vec<(StorageWriteOp, Sender<Result<(), StorageError<T>>>)>, ) -> ControlFlow<String>

fn apply_write_op(&mut self, op: StorageWriteOp)

Apply the given write operation to the desired/to_write state.

async fn write_to_persist( &mut self, responders: Vec<Sender<Result<(), StorageError<T>>>>, ) -> ControlFlow<String>

Attempt to write what is currently in Self::to_write to persist, retrying and re-syncing to persist when necessary, that is when the upper was not what we expected.

async fn sync_to_persist(&mut self)

Re-derives Self::to_write by looking at Self::desired and the current state in persist. We want to insert everything in desired and retract everything in persist. But ideally most of that cancels out in consolidation.

To be called when a compare_and_append failed because the upper didn’t match what we expected.