Struct mz_storage_client::controller::rehydration::RehydrationTask

struct RehydrationTask<T> {
    build_info: &'static BuildInfo,
    command_rx: UnboundedReceiver<RehydrationCommand<T>>,
    response_tx: UnboundedSender<StorageResponse<T>>,
    sources: BTreeMap<GlobalId, CreateSourceCommand<T>>,
    sinks: BTreeMap<GlobalId, CreateSinkCommand<T>>,
    uppers: BTreeMap<GlobalId, Antichain<T>>,
    sinces: BTreeMap<GlobalId, Antichain<T>>,
    initialized: bool,
    current_epoch: ClusterStartupEpoch,
    config: StorageParameters,
    metrics: RehydratingStorageClientMetrics,
}

A task that manages rehydration.

Fields

build_info: &'static BuildInfo

The build information for this process.

command_rx: UnboundedReceiver<RehydrationCommand<T>>

A channel upon which commands intended for the storage replica are delivered.

response_tx: UnboundedSender<StorageResponse<T>>

A channel upon which responses from the storage replica are delivered.

sources: BTreeMap<GlobalId, CreateSourceCommand<T>>

The sources that have been observed.

sinks: BTreeMap<GlobalId, CreateSinkCommand<T>>

The exports that have been observed.

uppers: BTreeMap<GlobalId, Antichain<T>>

The upper frontier information received.

sinces: BTreeMap<GlobalId, Antichain<T>>

The since frontiers that have been observed.

initialized: bool

Set to true once StorageCommand::InitializationComplete has been observed.

current_epoch: ClusterStartupEpoch

The current epoch for the replica we are connecting to.

config: StorageParameters

Storage configuration that has been observed.

metrics: RehydratingStorageClientMetrics

Prometheus metrics

Implementations

impl<T> RehydrationTask<T>where
    T: Timestamp + Lattice + Codec64,
    StorageGrpcClient: StorageClient<T>,

async fn run(&mut self)

async fn step_await_address(&mut self) -> RehydrationTaskState<T>

async fn step_rehydrate(
    &mut self,
    location: ClusterReplicaLocation
) -> RehydrationTaskState<T>

async fn step_pump(
    &mut self,
    location: ClusterReplicaLocation,
    client: Partitioned<StorageGrpcClient, StorageCommand<T>, StorageResponse<T>>
) -> RehydrationTaskState<T>

async fn send_commands(
    &mut self,
    location: ClusterReplicaLocation,
    client: Partitioned<StorageGrpcClient, StorageCommand<T>, StorageResponse<T>>,
    commands: impl IntoIterator<Item = StorageCommand<T>>
) -> RehydrationTaskState<T>

fn send_response(
    &mut self,
    location: ClusterReplicaLocation,
    client: Partitioned<StorageGrpcClient, StorageCommand<T>, StorageResponse<T>>,
    response: Result<StorageResponse<T>, Error>
) -> RehydrationTaskState<T>

fn absorb_command(&mut self, command: &StorageCommand<T>)

fn absorb_response(
    &mut self,
    response: StorageResponse<T>
) -> Option<StorageResponse<T>>