Method add_replica

src/compute-client/src/controller/instance.rs:1213–1279

(
        &mut self,
        id: ReplicaId,
        mut config: ReplicaConfig,
        epoch: Option<u64>,
    )

    /// Add a new instance replica, by ID.
    #[mz_ore::instrument(level = "debug")]
    pub fn add_replica(
        &mut self,
        id: ReplicaId,
        mut config: ReplicaConfig,
        epoch: Option<u64>,
    ) -> Result<(), ReplicaExists> {
        if self.replica_exists(id) {
            return Err(ReplicaExists(id));
        }

        config.logging.index_logs = self.log_sources.clone();

        let epoch = epoch.unwrap_or(1);
        let metrics = self.metrics.for_replica(id);
        let client = ReplicaClient::spawn(
            id,
            self.build_info,
            config.clone(),
            epoch,
            metrics.clone(),
            Arc::clone(&self.dyncfg),
            self.replica_tx.clone(),
        );

        // Take this opportunity to clean up the history we should present.
        self.history.reduce();

        // Advance the uppers of source imports
        self.history.update_source_uppers(&self.storage_collections);

        // Replay the commands at the client, creating new dataflow identifiers.
        for command in self.history.iter() {
            // Skip `CreateDataflow` commands targeted at different replicas.
            if let Some(target_replica) = self.target_replica(command)
                && target_replica != id
            {
                continue;
            }

            // Re-apply this replica's dyncfg override to replayed config commands, and rebuild the
            // create-instance snapshot from the current dyncfg.
            let command = Self::specialize_command_for_replica(
                command.clone(),
                id,
                &self.replica_dyncfg_overrides,
                &self.dyncfg,
            );
            if client.send(command).is_err() {
                // We swallow the error here. On the next send, we will fail again, and
                // restart the connection as well as this rehydration.
                tracing::warn!("Replica {:?} connection terminated during hydration", id);
                break;
            }
        }

        // Add replica to tracked state.
        if self.add_replica_state(id, client, config, epoch).is_err() {
            // A storage read hold issuer hung up, which only happens during process shutdown.