laminar_db/db/topology/
recovery.rs

1//! Private target recovery reuses the existing compiler, checkpoint reader and operator codecs.
2
3use laminar_core::checkpoint::merge_node_subscription_manifests;
4use laminar_core::cluster::control::{TopologyError, TopologyOperationId, TopologyRestoreInput};
5
6use super::restore::TopologyRestorePurpose;
7use super::{DbError, LaminarDB, PreparedTopologyRestore};
8
9impl LaminarDB {
10    /// Build one unstarted recovery image from the greatest exact target checkpoint, or the
11    /// explicitly mapped root when no target checkpoint has committed. State, source cursors and
12    /// subscription frontiers come from the same selected cut. The existing compiler owns the
13    /// image, and verified encoded state is released immediately after decoding.
14    ///
15    /// This changes no live catalog, transport, actors or output. A recovery image cannot use
16    /// original migration installation/Release APIs; coordinated recovery authorization is required.
17    ///
18    /// # Errors
19    /// Rejects live/faulted local actors, obsolete targets, changed ownership/process adoption,
20    /// damaged state/output, missing source progress and configured read/state/deadline limits.
21    pub async fn prepare_cluster_topology_recovery(
22        &self,
23        operation_id: TopologyOperationId,
24    ) -> Result<PreparedTopologyRestore, DbError> {
25        self.prepare_topology_restore_image(operation_id, TopologyRestorePurpose::Recovery)
26            .await
27    }
28}
29
30pub(super) fn validate_target_subscription_frontiers(
31    graph: &crate::operator_graph::OperatorGraph,
32    recovered: &crate::recovery_manager::RecoveredState,
33    input: &TopologyRestoreInput,
34) -> Result<(), DbError> {
35    let views = recovered
36        .manifests
37        .iter()
38        .filter_map(|manifest| {
39            manifest
40                .subscription_output
41                .as_ref()
42                .map(|subscription| (subscription, manifest.owned_vnodes.as_slice()))
43        })
44        .collect::<Vec<_>>();
45    let merged = if views.is_empty() {
46        Vec::new()
47    } else {
48        merge_node_subscription_manifests(
49            recovered.committed.epoch,
50            recovered.committed.checkpoint_id,
51            recovered
52                .committed
53                .assignment_fence
54                .as_ref()
55                .ok_or(TopologyError::Fenced)?,
56            &views,
57        )
58        .map_err(|error| TopologyError::Invalid(error.to_string()))?
59    };
60    let captures = graph.capture_subscription_frontiers()?;
61    if captures.len() != merged.len() {
62        return Err(TopologyError::Invalid(
63            "target subscription roster differs from the selected checkpoint".into(),
64        )
65        .into());
66    }
67    for (capture, checkpoint) in captures.iter().zip(merged) {
68        let expected = checkpoint
69            .manifest
70            .frontiers
71            .iter()
72            .filter(|frontier| {
73                input
74                    .owned_vnodes()
75                    .binary_search(&u32::from(frontier.partition.get()))
76                    .is_ok()
77            })
78            .collect::<Vec<_>>();
79        if capture.certificate.as_ref() != &checkpoint.manifest.distribution_certificate
80            || capture.frontiers.iter().collect::<Vec<_>>() != expected
81        {
82            return Err(TopologyError::Invalid("target subscription identity or exclusive sequence differs from the selected checkpoint".into()).into());
83        }
84    }
85    Ok(())
86}
laminar_db/db/topology/recovery.rs

laminar_db/db/topology/
recovery.rs
