Method run_export

src/storage-controller/src/lib.rs:3429–3486

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

(&mut self, id: GlobalId)

    /// Runs the identified export using the current definition of the export
    /// that we have in memory.
    fn run_export(&mut self, id: GlobalId) -> Result<(), StorageError> {
        let DataSource::Sink { desc: description } = &self.collections[&id].data_source else {
            return Err(StorageError::IdentifierMissing(id));
        };

        let from_storage_metadata = self
            .storage_collections
            .collection_metadata(description.sink.from)?;
        let to_storage_metadata = self.storage_collections.collection_metadata(id)?;

        // Choose an as-of frontier for this execution of the sink. If the write frontier of the sink
        // is strictly larger than its read hold, it must have at least written out its snapshot, and we can skip
        // reading it; otherwise assume we may have to replay from the beginning.
        let export_state = self.storage_collections.collection_frontiers(id)?;
        let mut as_of = description.sink.as_of.clone();
        as_of.join_assign(&export_state.implied_capability);
        let with_snapshot = description.sink.with_snapshot
            && !PartialOrder::less_than(&as_of, &export_state.write_frontier);

        info!(
            sink_id = %id,
            from_id = %description.sink.from,
            write_frontier = ?export_state.write_frontier,
            ?as_of,
            ?with_snapshot,
            "run_export"
        );

        let cmd = RunSinkCommand {
            id,
            description: StorageSinkDesc {
                from: description.sink.from,
                from_desc: description.sink.from_desc.clone(),
                connection: description.sink.connection.clone(),
                envelope: description.sink.envelope,
                as_of,
                version: description.sink.version,
                from_storage_metadata,
                with_snapshot,
                to_storage_metadata,
                commit_interval: description.sink.commit_interval,
            },
        };

        let storage_instance_id = description.instance_id.clone();

        let instance = self
            .instances
            .get_mut(&storage_instance_id)
            .ok_or_else(|| StorageError::ExportInstanceMissing {
                storage_instance_id,
                export_id: id,
            })?;

        instance.send(StorageCommand::RunSink(Box::new(cmd)));

        Ok(())
    }