laminar_db/db/topology/
retirement.rs

1//! Observe the held parent generation through existing lifecycle and connector owners.
2
3use std::sync::atomic::Ordering;
4use std::time::Duration;
5
6use laminar_core::cluster::control::{TopologyError, TopologyRestoreInput};
7
8use super::{DbError, DbState, LaminarDB, PreparedTopologyRestore};
9
10impl LaminarDB {
11    /// Retire the old actors for this database's privately restored exact-root target.
12    ///
13    /// Revalidates the full current preparation roster, leader, process and assignment before
14    /// signalling stop and after observing completion. Existing lifecycle code joins compute,
15    /// settles decision/sink work and observes source, sink and connector-child termination.
16    /// No request to cancel or close is accepted as terminal proof.
17    ///
18    /// The parent catalog/coordinator identity, cut hold and namespace lock remain owned, with
19    /// the runtime in `ShuttingDown`. The target stays unstarted. No receipt or authority append,
20    /// topology Commit, target install or Release occurs. Public start/stop remain fenced.
21    ///
22    /// The total budget is 45 seconds. Cancellation or error retains unresolved DB-owned handles
23    /// and the held boundary; retry with the same image, or use coordinated recovery after abort.
24    /// Dropping the image never reopens intake. Its observation must be revalidated at installation.
25    ///
26    /// # Errors
27    /// Rejects another database's image, stale authority, loss of the held parent, concurrent
28    /// recovery/shutdown/fault, and unobserved runtime or connector termination.
29    pub async fn retire_cluster_topology_parent(
30        &self,
31        image: &mut PreparedTopologyRestore,
32    ) -> Result<(), DbError> {
33        if !image.belongs_to(self) {
34            return Err(TopologyError::Conflict(
35                "target image belongs to a different database runtime".into(),
36            )
37            .into());
38        }
39        image.parent_retirement_observed = false;
40        tokio::time::timeout(Duration::from_secs(45), async {
41            self.validate_topology_retirement_input(&image.input)
42                .await?;
43            self.stop_pipeline_for_topology_retirement().await?;
44            self.validate_topology_retirement_input(&image.input)
45                .await?;
46            if DbState::load(&self.state) != DbState::ShuttingDown
47                || !self.runtime_shutdown.read().is_cancelled()
48            {
49                return Err(TopologyError::Fenced.into());
50            }
51            image.parent_retirement_observed = true;
52            Ok(())
53        })
54        .await
55        .map_err(|_| TopologyError::Contended)?
56    }
57
58    pub(super) async fn validate_topology_retirement_input(
59        &self,
60        input: &TopologyRestoreInput,
61    ) -> Result<(), DbError> {
62        self.ensure_topology_retirement_available()?;
63        let controller = self.cluster_controller.lock().clone().ok_or_else(|| {
64            TopologyError::Protocol("retirement requires the configured controller".into())
65        })?;
66        if !controller
67            .topology_restore_input(input.operation().operation_id)
68            .await?
69            .same_restore_requirements(input)
70        {
71            return Err(TopologyError::Fenced.into());
72        }
73        self.validate_bound_parent_pipeline(&input.descriptor().parent_pipeline)
74            .await?;
75        if self.catalog_manifest_inventory()? != input.parent().entries
76            || self.topology_definition_identities().await?.pipeline
77                != input.descriptor().parent_pipeline
78        {
79            return Err(TopologyError::Fenced.into());
80        }
81        self.ensure_topology_retirement_available()?;
82        // The coordinator lock above can suspend after the durable input read. Recheck the
83        // exact local process at the final synchronous boundary, before recording observation.
84        if controller.is_recovering()
85            || controller.is_draining()
86            || controller.try_live_local_process_authority_identity().ok() != Some(input.process())
87        {
88            return Err(TopologyError::Fenced.into());
89        }
90        Ok(())
91    }
92
93    pub(super) fn ensure_topology_retirement_available(&self) -> Result<(), DbError> {
94        if self.is_closed() {
95            return Err(DbError::Shutdown);
96        }
97        if !self.is_cluster_runtime()
98            || !matches!(
99                DbState::load(&self.state),
100                DbState::Running | DbState::ShuttingDown
101            )
102            || !self.topology_cut_hold.load(Ordering::Acquire)
103            || !self.source_gate.load(Ordering::Acquire)
104            || self.cluster_authority_revoked.load(Ordering::Acquire)
105            || self.durable_terminal_recovery_fence.load(Ordering::Acquire)
106            || self.terminal_pipeline_halt.load(Ordering::Acquire)
107            || self.coordinated_recovery_in_progress()
108            || self.pending_recovery_fault.load(Ordering::Acquire) != 0
109            || self.last_fault.lock().is_some()
110        {
111            return Err(TopologyError::Conflict(
112                "parent retirement requires the exact held runtime without a recovery or fault"
113                    .into(),
114            )
115            .into());
116        }
117        self.ensure_catalog_cleanup_unfenced("topology retirement")
118    }
119}
laminar_db/db/topology/retirement.rs

laminar_db/db/topology/
retirement.rs
