Struct mz_storage_controller::Controller

pub struct Controller<T: Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation> {

    pub(crate) build_info: &'static BuildInfo,
    pub(crate) now: NowFn,
    pub(crate) envd_epoch: NonZeroI64,
    pub(crate) read_only: bool,
    pub(crate) collections: BTreeMap<GlobalId, CollectionState<T>>,
    pub(crate) exports: BTreeMap<GlobalId, ExportState<T>>,
    pub(crate) persist_table_worker: PersistTableWriteWorker<T>,
    pub(crate) txns_read: TxnsRead<T>,
    pub(crate) txns_metrics: Arc<Metrics>,
    pub(crate) stashed_response: Option<StorageResponse<T>>,
    pub(crate) pending_compaction_commands: Vec<PendingCompactionCommand<T>>,
    pub(crate) pending_table_handle_drops_tx: UnboundedSender<GlobalId>,
    pub(crate) pending_table_handle_drops_rx: UnboundedReceiver<GlobalId>,
    pub(crate) collection_manager: CollectionManager<T>,
    pub(crate) introspection_ids: BTreeMap<IntrospectionType, GlobalId>,
    pub(crate) introspection_tokens: Arc<Mutex<BTreeMap<GlobalId, Box<dyn Any + Send + Sync>>>>,
    pub(crate) source_statistics: Arc<Mutex<SourceStatistics>>,
    pub(crate) sink_statistics: Arc<Mutex<BTreeMap<GlobalId, StatsState<SinkStatisticsUpdate>>>>,
    pub(crate) statistics_interval_sender: Sender<Duration>,
    pub(crate) instances: BTreeMap<StorageInstanceId, Instance<T>>,
    pub(crate) initialized: bool,
    pub(crate) config: StorageConfiguration,
    pub(crate) internal_response_queue: UnboundedReceiver<StorageResponse<T>>,
    pub(crate) persist_location: PersistLocation,
    pub(crate) persist: Arc<PersistClientCache>,
    pub(crate) metrics: StorageControllerMetrics,
    pub(crate) internal_response_sender: UnboundedSender<StorageResponse<T>>,
    pub(crate) recorded_frontiers: BTreeMap<GlobalId, (Antichain<T>, Antichain<T>)>,
    pub(crate) recorded_replica_frontiers: BTreeMap<(GlobalId, ReplicaId), Antichain<T>>,
    pub(crate) wallclock_lag: WallclockLagFn<T>,
    pub(crate) wallclock_lag_last_refresh: Instant,
    pub(crate) storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>,
    pub(crate) migrated_storage_collections: BTreeSet<GlobalId>,
    pub(crate) maintenance_ticker: Interval,
    pub(crate) maintenance_scheduled: bool,
    pub(crate) instance_response_tx: UnboundedSender<StorageResponse<T>>,
    pub(crate) instance_response_rx: UnboundedReceiver<StorageResponse<T>>,
    pub(crate) persist_warm_task: Option<AbortOnDropHandle<Box<dyn Debug + Send>>>,
}

A storage controller for a storage instance.

Fields

build_info: &'static BuildInfo

The build information for this process.

now: NowFn

A function that returns the current time.

envd_epoch: NonZeroI64

The fencing token for this instance of the controller.

read_only: bool

Whether or not this controller is in read-only mode.

When in read-only mode, neither this controller nor the instances controlled by it are allowed to affect changes to external systems (largely persist).

collections: BTreeMap<GlobalId, CollectionState<T>>

Collections maintained by the storage controller.

This collection only grows, although individual collections may be rendered unusable. This is to prevent the re-binding of identifiers to other descriptions.

exports: BTreeMap<GlobalId, ExportState<T>>

persist_table_worker: PersistTableWriteWorker<T>

Write handle for table shards.

txns_read: TxnsRead<T>

A shared TxnsCache running in a task and communicated with over a channel.

txns_metrics: Arc<Metrics>

stashed_response: Option<StorageResponse<T>>

pending_compaction_commands: Vec<PendingCompactionCommand<T>>

Compaction commands to send during the next call to StorageController::process.

pending_table_handle_drops_tx: UnboundedSender<GlobalId>

Channel for sending table handle drops.

pending_table_handle_drops_rx: UnboundedReceiver<GlobalId>

Channel for receiving table handle drops.

collection_manager: CollectionManager<T>

Interface for managed collections

introspection_ids: BTreeMap<IntrospectionType, GlobalId>

Tracks which collection is responsible for which IntrospectionType.

introspection_tokens: Arc<Mutex<BTreeMap<GlobalId, Box<dyn Any + Send + Sync>>>>

Tokens for tasks that drive updating introspection collections. Dropping this will make sure that any tasks (or other resources) will stop when needed.

source_statistics: Arc<Mutex<SourceStatistics>>

Consolidated metrics updates to periodically write. We do not eagerly initialize this, and its contents are entirely driven by StorageResponse::StatisticsUpdates’s, as well as webhook statistics.

sink_statistics: Arc<Mutex<BTreeMap<GlobalId, StatsState<SinkStatisticsUpdate>>>>

Consolidated metrics updates to periodically write. We do not eagerly initialize this, and its contents are entirely driven by StorageResponse::StatisticsUpdates’s.

statistics_interval_sender: Sender<Duration>

A way to update the statistics interval in the statistics tasks.

instances: BTreeMap<StorageInstanceId, Instance<T>>

Clients for all known storage instances.

initialized: bool

Set to true once initialization_complete has been called.

config: StorageConfiguration

Storage configuration to apply to newly provisioned instances, and use during purification.

internal_response_queue: UnboundedReceiver<StorageResponse<T>>

Mechanism for returning frontier advancement for tables.

persist_location: PersistLocation

The persist location where all storage collections are being written to

persist: Arc<PersistClientCache>

A persist client used to write to storage collections

metrics: StorageControllerMetrics

Metrics of the Storage controller

internal_response_sender: UnboundedSender<StorageResponse<T>>

Mechanism for the storage controller to send itself feedback, potentially emulating the responses we expect from clusters.

Note: This is used for finalizing shards of webhook sources, once webhook sources are installed on a clusterd this can likely be refactored away.

recorded_frontiers: BTreeMap<GlobalId, (Antichain<T>, Antichain<T>)>

(read, write) frontiers that have been recorded in the Frontiers collection, kept to be able to retract old rows.

recorded_replica_frontiers: BTreeMap<(GlobalId, ReplicaId), Antichain<T>>

Write frontiers that have been recorded in the ReplicaFrontiers collection, kept to be able to retract old rows.

wallclock_lag: WallclockLagFn<T>

A function that computes the lag between the given time and wallclock time.

wallclock_lag_last_refresh: Instant

The last time wallclock lag introspection was refreshed.

storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>

Handle to a StorageCollections.

migrated_storage_collections: BTreeSet<GlobalId>

Migrated storage collections that can be written even in read only mode.

maintenance_ticker: Interval

Ticker for scheduling periodic maintenance work.

maintenance_scheduled: bool

Whether maintenance work was scheduled.

instance_response_tx: UnboundedSender<StorageResponse<T>>

Shared transmit channel for replicas to send responses.

instance_response_rx: UnboundedReceiver<StorageResponse<T>>

Receive end for replica responses.

persist_warm_task: Option<AbortOnDropHandle<Box<dyn Debug + Send>>>

Background task run at startup to warm persist state.

Implementations

impl<T> Controller<T>

where T: Timestamp + Lattice + TotalOrder + Codec64 + From<EpochMillis> + TimestampManipulation + Into<Datum<'static>>, StorageCommand<T>: RustType<ProtoStorageCommand>, StorageResponse<T>: RustType<ProtoStorageResponse>, Self: StorageController<Timestamp = T>,

pub async fn new( build_info: &'static BuildInfo, persist_location: PersistLocation, persist_clients: Arc<PersistClientCache>, now: NowFn, wallclock_lag: WallclockLagFn<T>, txns_metrics: Arc<TxnMetrics>, envd_epoch: NonZeroI64, read_only: bool, metrics_registry: MetricsRegistry, connection_context: ConnectionContext, txn: &dyn StorageTxn<T>, storage_collections: Arc<dyn StorageCollections<Timestamp = T> + Send + Sync>, ) -> Self

Create a new storage controller from a client it should wrap.

Note that when creating a new storage controller, you must also reconcile it with the previous state.

Panics

If this function is called before prepare_initialization.

pub(crate) fn set_hold_policies( &mut self, policies: Vec<(GlobalId, ReadPolicy<T>)>, )

pub(crate) fn update_write_frontiers( &mut self, updates: &[(GlobalId, Antichain<T>)], )

pub(crate) fn update_hold_capabilities( &mut self, updates: &mut BTreeMap<GlobalId, ChangeBatch<T>>, )

pub(crate) fn validate_collection_ids( &self, ids: impl Iterator<Item = GlobalId>, ) -> Result<(), StorageError<T>>

Validate that a collection exists for all identifiers, and error if any do not.

pub(crate) fn validate_export_ids( &self, ids: impl Iterator<Item = GlobalId>, ) -> Result<(), StorageError<T>>

Validate that a collection exists for all identifiers, and error if any do not.

pub(crate) fn active_exports( &self, ) -> impl Iterator<Item = (GlobalId, &ExportState<T>)>

Iterate over exports that have not been dropped.

pub(crate) async fn recent_upper( &self, id: GlobalId, ) -> Result<Antichain<T>, StorageError<T>>

pub(crate) async fn open_data_handles( &self, id: &GlobalId, shard: ShardId, relation_desc: RelationDesc, persist_client: &PersistClient, ) -> WriteHandle<SourceData, (), T, Diff>

Opens a write and critical since handles for the given shard.

since is an optional since that the read handle will be forwarded to if it is less than its current since.

This will halt! the process if we cannot successfully acquire a critical handle with our current epoch.

pub(crate) fn register_introspection_collection( &mut self, id: GlobalId, introspection_type: IntrospectionType, write_handle: WriteHandle<SourceData, (), T, Diff>, persist_client: PersistClient, ) -> Result<(), StorageError<T>>

Registers the given introspection collection and does any preparatory work that we have to do before we start writing to it. This preparatory work will include partial truncation or other cleanup schemes, depending on introspection type.

pub(crate) fn reconcile_dangling_statistics(&self)

Remove statistics for sources/sinks that were dropped but still have statistics rows hanging around.

pub(crate) fn append_shard_mappings<I>(&self, global_ids: I, diff: i64)

where I: Iterator<Item = GlobalId>,

Appends a new global ID, shard ID pair to the appropriate collection. Use a diff of 1 to append a new entry; -1 to retract an existing entry.

Panics

• If self.collections does not have an entry for global_id.
• If IntrospectionType::ShardMapping’s GlobalId is not registered as a managed collection.
• If diff is any value other than 1 or -1.

pub(crate) fn determine_collection_dependencies( &self, self_id: GlobalId, data_source: &DataSource, ) -> Result<Vec<GlobalId>, StorageError<T>>

Determines and returns this collection’s dependencies, if any.

pub(crate) async fn read_handle_for_snapshot( &self, id: GlobalId, ) -> Result<ReadHandle<SourceData, (), T, Diff>, StorageError<T>>

pub(crate) async fn snapshot_and_stream( &self, id: GlobalId, as_of: T, ) -> Result<BoxStream<'_, (SourceData, T, Diff)>, StorageError<T>>

pub(crate) fn record_status_updates(&mut self, updates: Vec<StatusUpdate>)

Handles writing of status updates for sources/sinks to the appropriate status relation

pub(crate) fn collection( &self, id: GlobalId, ) -> Result<&CollectionState<T>, StorageError<T>>

pub(crate) fn run_ingestion( &mut self, id: GlobalId, ) -> Result<(), StorageError<T>>

Runs the identified ingestion using the current definition of the ingestion in-memory.

pub(crate) fn run_export(&mut self, id: GlobalId) -> Result<(), StorageError<T>>

Runs the identified export using the current definition of the export that we have in memory.

pub(crate) fn update_frontier_introspection(&mut self)

Update introspection with the current frontiers of storage objects.

This method is invoked by Controller::maintain, which we expect to be called once per second during normal operation.

pub(crate) fn update_wallclock_lag_introspection(&mut self)

Update introspection with the current wallclock lag values.

We measure the lag of write frontiers behind the wallclock time every second and track the maximum over 60 measurements (i.e., one minute). Every minute, we emit a new lag event to the WallclockLagHistory introspection with the current maximum.

This method is invoked by ComputeController::maintain, which we expect to be called once per second during normal operation.

pub(crate) fn maintain(&mut self)

Run periodic tasks.

This method is invoked roughly once per second during normal operation. It is a good place for tasks that need to run periodically, such as state cleanup or updating of metrics.

Trait Implementations

impl<T> Debug for Controller<T>

where T: Debug + Timestamp + Lattice + Codec64 + From<EpochMillis> + TimestampManipulation,

fn fmt(&self, __f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> StorageController for Controller<T>

where T: Timestamp + Lattice + TotalOrder + Codec64 + From<EpochMillis> + TimestampManipulation + Into<Datum<'static>> + Display, StorageCommand<T>: RustType<ProtoStorageCommand>, StorageResponse<T>: RustType<ProtoStorageResponse>,

fn config(&self) -> &StorageConfiguration

Get the current configuration

fn create_collections_for_bootstrap<'life0, 'life1, 'life2, 'async_trait>( &'life0 mut self, storage_metadata: &'life1 StorageMetadata, register_ts: Option<Self::Timestamp>, collections: Vec<(GlobalId, CollectionDescription<Self::Timestamp>)>, migrated_storage_collections: &'life2 BTreeSet<GlobalId>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Create and “execute” the described collection.

“Execute” is in scare quotes because what executing a collection means varies widely based on the type of collection you’re creating.

The general process creating a collection undergoes is:

1. Enrich the description we get from the user with the metadata only the storage controller’s metadata. This is mostly a matter of separating concerns.
2. Generate write and read persist handles for the collection.
3. Store the collection’s metadata in the appropriate field.
4. “Execute” the collection. What that means is contingent on the type of collection. so consult the code for more details.

fn alter_export_connections<'life0, 'async_trait>( &'life0 mut self, exports: BTreeMap<GlobalId, StorageSinkConnection>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create the sinks described by the ExportDescription.

fn drop_sinks( &mut self, identifiers: Vec<GlobalId>, ) -> Result<(), StorageError<Self::Timestamp>>

Drops the read capability for the sinks and allows their resources to be reclaimed.

type Timestamp = T

fn initialization_complete(&mut self)

Marks the end of any initialization commands.

fn update_parameters(&mut self, config_params: StorageParameters)

Update storage configuration with new parameters.

fn collection_metadata( &self, id: GlobalId, ) -> Result<CollectionMetadata, StorageError<Self::Timestamp>>

Returns the CollectionMetadata of the collection identified by id.

fn collection_hydrated( &self, collection_id: GlobalId, ) -> Result<bool, StorageError<Self::Timestamp>>

Returns true iff the given collection/ingestion has been hydrated.

fn collection_frontiers( &self, id: GlobalId, ) -> Result<(Antichain<Self::Timestamp>, Antichain<Self::Timestamp>), StorageError<Self::Timestamp>>

Returns the since/upper frontiers of the identified collection.

fn collections_frontiers( &self, ids: Vec<GlobalId>, ) -> Result<Vec<(GlobalId, Antichain<T>, Antichain<T>)>, StorageError<Self::Timestamp>>

Returns the since/upper frontiers of the identified collections.

fn active_collection_metadatas(&self) -> Vec<(GlobalId, CollectionMetadata)>

Acquire an iterator over CollectionMetadata for all active collections.

fn active_ingestions( &self, instance_id: StorageInstanceId, ) -> &BTreeSet<GlobalId>

Returns the IDs of all active ingestions for the given storage instance.

fn check_exists( &self, id: GlobalId, ) -> Result<(), StorageError<Self::Timestamp>>

Checks whether a collection exists under the given GlobalId. Returns an error if the collection does not exist.

fn create_instance(&mut self, id: StorageInstanceId)

Creates a storage instance with the specified ID.

fn drop_instance(&mut self, id: StorageInstanceId)

Drops the storage instance with the given ID.

fn connect_replica( &mut self, instance_id: StorageInstanceId, replica_id: ReplicaId, location: ClusterReplicaLocation, )

Connects the storage instance to the specified replica.

fn drop_replica( &mut self, instance_id: StorageInstanceId, replica_id: ReplicaId, )

Disconnects the storage instance from the specified replica.

fn check_alter_ingestion_source_desc( &mut self, ingestion_id: GlobalId, source_desc: &SourceDesc, ) -> Result<(), StorageError<Self::Timestamp>>

Check that the ingestion associated with id can use the provided SourceDesc.

fn alter_ingestion_source_desc<'life0, 'async_trait>( &'life0 mut self, ingestion_id: GlobalId, source_desc: SourceDesc, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Alters the identified collection to use the provided SourceDesc.

fn alter_ingestion_connections<'life0, 'async_trait>( &'life0 mut self, source_connections: BTreeMap<GlobalId, GenericSourceConnection<InlinedConnection>>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Alters each identified collection to use the correlated GenericSourceConnection.

fn alter_ingestion_export_data_configs<'life0, 'async_trait>( &'life0 mut self, source_exports: BTreeMap<GlobalId, SourceExportDataConfig>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Alters the data config for the specified source exports of the specified ingestions.

fn alter_table_desc<'life0, 'async_trait>( &'life0 mut self, table_id: GlobalId, new_desc: RelationDesc, expected_schema: SchemaId, forget_ts: Self::Timestamp, register_ts: Self::Timestamp, ) -> Pin<Box<dyn Future<Output = Result<SchemaId, StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn export( &self, id: GlobalId, ) -> Result<&ExportState<Self::Timestamp>, StorageError<Self::Timestamp>>

Acquire an immutable reference to the export state, should it exist.

fn export_mut( &mut self, id: GlobalId, ) -> Result<&mut ExportState<Self::Timestamp>, StorageError<Self::Timestamp>>

Acquire a mutable reference to the export state, should it exist.

fn create_exports<'life0, 'async_trait>( &'life0 mut self, exports: Vec<(GlobalId, ExportDescription<Self::Timestamp>)>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Create the sinks described by the ExportDescription.

fn alter_export<'life0, 'async_trait>( &'life0 mut self, id: GlobalId, new_description: ExportDescription<Self::Timestamp>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Alter the sink identified by the given id to match the provided ExportDescription.

fn drop_tables( &mut self, storage_metadata: &StorageMetadata, identifiers: Vec<GlobalId>, ts: Self::Timestamp, ) -> Result<(), StorageError<Self::Timestamp>>

Drops the read capability for the tables and allows their resources to be reclaimed.

fn drop_sources( &mut self, storage_metadata: &StorageMetadata, identifiers: Vec<GlobalId>, ) -> Result<(), StorageError<Self::Timestamp>>

Drops the read capability for the sources and allows their resources to be reclaimed.

fn drop_sources_unvalidated( &mut self, storage_metadata: &StorageMetadata, ids: Vec<GlobalId>, ) -> Result<(), StorageError<Self::Timestamp>>

Drops the read capability for the sources and allows their resources to be reclaimed.

fn drop_sinks_unvalidated(&mut self, identifiers: Vec<GlobalId>)

Drops the read capability for the sinks and allows their resources to be reclaimed.

fn append_table( &mut self, write_ts: Self::Timestamp, advance_to: Self::Timestamp, commands: Vec<(GlobalId, Vec<TimestamplessUpdate>)>, ) -> Result<Receiver<Result<(), StorageError<Self::Timestamp>>>, StorageError<Self::Timestamp>>

Append updates into the local input named id and advance its upper to upper.

fn monotonic_appender( &self, id: GlobalId, ) -> Result<MonotonicAppender<Self::Timestamp>, StorageError<Self::Timestamp>>

Returns a MonotonicAppender which is a channel that can be used to monotonically append to the specified GlobalId.

fn webhook_statistics( &self, id: GlobalId, ) -> Result<Arc<WebhookStatistics>, StorageError<Self::Timestamp>>

Returns a shared WebhookStatistics which can be used to report user-facing statistics for this given webhhook, specified by the GlobalId.

fn snapshot( &self, id: GlobalId, as_of: Self::Timestamp, ) -> BoxFuture<Result<Vec<(Row, Diff)>, StorageError<Self::Timestamp>>>

Returns the snapshot of the contents of the local input named id at as_of.

fn snapshot_latest<'life0, 'async_trait>( &'life0 self, id: GlobalId, ) -> Pin<Box<dyn Future<Output = Result<Vec<Row>, StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns the snapshot of the contents of the local input named id at the largest readable as_of.

fn snapshot_cursor<'life0, 'async_trait>( &'life0 mut self, id: GlobalId, as_of: Self::Timestamp, ) -> Pin<Box<dyn Future<Output = Result<SnapshotCursor<Self::Timestamp>, StorageError<Self::Timestamp>>> + 'async_trait>>

where Self::Timestamp: Timestamp + Lattice + Codec64, Self: 'async_trait, 'life0: 'async_trait,

Returns the snapshot of the contents of the local input named id at as_of.

fn snapshot_stats<'life0, 'async_trait>( &'life0 self, id: GlobalId, as_of: Antichain<Self::Timestamp>, ) -> Pin<Box<dyn Future<Output = Result<SnapshotStats, StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns aggregate statistics about the contents of the local input named id at as_of.

fn snapshot_parts_stats<'life0, 'async_trait>( &'life0 self, id: GlobalId, as_of: Antichain<Self::Timestamp>, ) -> Pin<Box<dyn Future<Output = BoxFuture<Result<SnapshotPartsStats, StorageError<Self::Timestamp>>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns aggregate statistics about the contents of the local input named id at as_of.

fn set_read_policy( &mut self, policies: Vec<(GlobalId, ReadPolicy<Self::Timestamp>)>, )

Assigns a read policy to specific identifiers.

fn acquire_read_holds( &self, desired_holds: Vec<GlobalId>, ) -> Result<Vec<ReadHold<Self::Timestamp>>, ReadHoldError>

Acquires and returns the desired read holds, advancing them to the since frontier when necessary.

fn ready<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = ()> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Waits until the controller is ready to process a response.

fn process( &mut self, storage_metadata: &StorageMetadata, ) -> Result<Option<Response<T>>, Error>

Processes the work queued by StorageController::ready.

fn inspect_persist_state<'life0, 'async_trait>( &'life0 self, id: GlobalId, ) -> Pin<Box<dyn Future<Output = Result<Value, Error>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Exposes the internal state of the data shard for debugging and QA.

fn append_introspection_updates( &mut self, type_: IntrospectionType, updates: Vec<(Row, Diff)>, )

Records append-only updates for the given introspection type.

fn append_status_introspection_updates( &mut self, type_: IntrospectionType, updates: Vec<StatusUpdate>, )

Records append-only status updates for the given introspection type.

fn update_introspection_collection( &mut self, type_: IntrospectionType, op: StorageWriteOp, )

Updates the desired state of the given introspection type.

fn initialize_state<'life0, 'life1, 'async_trait>( &'life0 mut self, txn: &'life1 mut (dyn StorageTxn<T> + Send), init_ids: BTreeSet<GlobalId>, drop_ids: BTreeSet<GlobalId>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<T>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

On boot, seed the controller’s metadata/state.

fn prepare_state<'life0, 'life1, 'async_trait>( &'life0 self, txn: &'life1 mut (dyn StorageTxn<T> + Send), ids_to_add: BTreeSet<GlobalId>, ids_to_drop: BTreeSet<GlobalId>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<T>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Update the implementor of StorageTxn with the appropriate metadata given the IDs to add and drop.

fn real_time_recent_timestamp<'life0, 'async_trait>( &'life0 self, timestamp_objects: BTreeSet<GlobalId>, timeout: Duration, ) -> Pin<Box<dyn Future<Output = Result<BoxFuture<Result<Self::Timestamp, StorageError<Self::Timestamp>>>, StorageError<Self::Timestamp>>> + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn create_collections<'life0, 'life1, 'async_trait>( &'life0 mut self, storage_metadata: &'life1 StorageMetadata, register_ts: Option<Self::Timestamp>, collections: Vec<(GlobalId, CollectionDescription<Self::Timestamp>)>, ) -> Pin<Box<dyn Future<Output = Result<(), StorageError<Self::Timestamp>>> + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Create the sources described in the individual RunIngestionCommand commands.

fn acquire_read_hold( &mut self, id: GlobalId, ) -> Result<ReadHold<Self::Timestamp>, ReadHoldError>

Acquires and returns the earliest legal read hold.