laminar_db/db/topology/
commit.rs

1//! Commit owns no actors; private reconstruction is available before any target checkpoint.
2
3use laminar_core::cluster::control::{TopologyAdmissionStatus, TopologyError};
4
5use super::{DbError, LaminarDB, PreparedTopologyRestore};
6
7impl LaminarDB {
8    /// Irreversibly commit this DB's retained target image after complete protocol-four preparation.
9    /// The configured leader supplies all authority/process/assignment fences. The target catalog
10    /// and exact root share one append; target actors, output and intake stay fenced.
11    ///
12    /// The total cooperative budget is 45 seconds, including observed retirement and final reads.
13    /// A cancelled/uncertain append can have committed. Retrying the same image resolves the same
14    /// decision; after image loss use `recover_committed_cluster_topology`, never parent rollback.
15    ///
16    /// # Errors
17    /// Rejects foreign/stale images, incomplete participant capabilities, local faults or unresolved
18    /// authority. Commit alone grants no installation or output permission; runtime DDL remains
19    /// guarded until automatic target recovery is certified.
20    pub async fn commit_cluster_topology_target(
21        &self,
22        image: &mut PreparedTopologyRestore,
23    ) -> Result<TopologyAdmissionStatus, DbError> {
24        if !image.belongs_to(self) {
25            return Err(TopologyError::Conflict(
26                "target image belongs to a different database runtime".into(),
27            )
28            .into());
29        }
30        tokio::time::timeout(std::time::Duration::from_secs(45), async {
31            let controller = self.cluster_controller.lock().clone().ok_or_else(|| {
32                TopologyError::Protocol("Commit requires the configured controller".into())
33            })?;
34            let status = self
35                .cluster_topology_operation_status(image.input.operation().operation_id)
36                .await?
37                .ok_or_else(|| TopologyError::Conflict("unknown topology operation".into()))?;
38            if status.commit.is_none() {
39                self.certify_cluster_topology_target_preparation(image)
40                    .await?;
41                // The retained image's cursor observation may have aged while peers prepared.
42                // Revalidate the sealed positions without resolving latest or starting readers.
43                drop(
44                    super::restore::prepare_source_positions(&image.candidate, &image.input)
45                        .await?,
46                );
47                self.validate_topology_retirement_input(&image.input)
48                    .await?;
49                controller.commit_topology_target(&image.input).await?;
50            }
51            self.ensure_topology_restore_available(true)?;
52            if !image.parent_retirement_observed {
53                return Err(TopologyError::Fenced.into());
54            }
55            // Observe again even on an identical retry; a historical flag is not terminal proof.
56            self.stop_pipeline_for_topology_retirement().await?;
57            let fresh = controller
58                .committed_topology_restore_input(image.input.operation().operation_id)
59                .await?;
60            if !fresh.same_restore_requirements(&image.input)
61                && !fresh.is_committed_successor_of(&image.input)
62            {
63                return Err(TopologyError::Fenced.into());
64            }
65            self.ensure_topology_restore_available(true)?;
66            let result = fresh.operation().clone();
67            image.input = fresh;
68            Ok(result)
69        })
70        .await
71        .map_err(|_| TopologyError::Contended)?
72    }
73}
laminar_db/db/topology/commit.rs

laminar_db/db/topology/
commit.rs
