laminar_core/checkpoint/committed_checkpoint/
participant_validation.rs

1use super::{
2    merge_manifest_progress, merge_node_subscription_manifests, BTreeMap, CheckpointManifest,
3    CommittedCheckpointIndex, KeyGroupCount,
4};
5
6impl CommittedCheckpointIndex {
7    /// Verify exact participant manifest bytes and complete, exclusive vnode ownership.
8    ///
9    /// # Errors
10    /// Returns an error when the manifests do not exactly represent this committed cut.
11    pub fn validate_participant_manifests(
12        &self,
13        manifests: &[(&CheckpointManifest, &[u8])],
14    ) -> Result<(), String> {
15        self.validate()?;
16        if manifests.len() != self.participants.len() {
17            return Err("participant manifest count differs from the committed index".into());
18        }
19
20        let key_group_count = KeyGroupCount::try_from(u32::from(self.vnode_count))
21            .map_err(|_| "committed checkpoint vnode count is invalid".to_owned())?;
22        let mut owners = vec![None; usize::from(self.vnode_count)];
23        let mut sink_names = None;
24        for ((manifest, encoded), reference) in manifests.iter().zip(&self.participants) {
25            reference.verify_manifest(manifest, encoded)?;
26            let errors = manifest.validate(key_group_count);
27            if let Some(error) = errors.first() {
28                return Err(format!(
29                    "participant {} manifest is invalid: {error}",
30                    reference.participant_id
31                ));
32            }
33            if manifest.checkpoint_id != self.checkpoint_id
34                || manifest.epoch != self.epoch
35                || manifest.deployment_id != self.deployment_id
36                || manifest.pipeline_identity != self.pipeline_identity
37                || manifest.vnode_count != self.vnode_count
38                || manifest.assignment_fence != self.assignment_fence
39                || manifest.reassignment_portable != self.reassignment_portable
40            {
41                return Err(format!(
42                    "participant {} manifest belongs to a different checkpoint cut",
43                    reference.participant_id
44                ));
45            }
46
47            if manifest.source_names != self.source_names {
48                return Err(
49                    "participant manifest source topology differs from the committed index".into(),
50                );
51            }
52            match sink_names {
53                Some(expected) if expected != manifest.sink_names.as_slice() => {
54                    return Err(
55                        "participant manifests disagree on the registered sink topology".into(),
56                    );
57                }
58                None => sink_names = Some(manifest.sink_names.as_slice()),
59                Some(_) => {}
60            }
61            for vnode in &manifest.owned_vnodes {
62                let owner = owners
63                    .get_mut(usize::from(*vnode))
64                    .ok_or_else(|| format!("manifest owns out-of-range vnode {vnode}"))?;
65                if owner.replace(manifest.participant_id).is_some() {
66                    return Err(format!(
67                        "vnode {vnode} is owned by more than one participant"
68                    ));
69                }
70            }
71        }
72        let mut offsets = BTreeMap::new();
73        let mut channels = BTreeMap::new();
74        for (manifest, _) in manifests {
75            merge_manifest_progress(manifest, &mut offsets, &mut channels)?;
76        }
77        if offsets != self.source_offsets {
78            return Err(
79                "participant source offsets do not exactly reconstruct the committed source cut"
80                    .into(),
81            );
82        }
83        if channels.into_values().collect::<Vec<_>>() != self.channel_progress {
84            return Err(
85                "participant channel progress does not exactly reconstruct the committed time cut"
86                    .into(),
87            );
88        }
89
90        if owners.iter().any(Option::is_none) {
91            return Err("participant manifests do not exactly cover the vnode domain".into());
92        }
93        if let Some(fence) = &self.assignment_fence {
94            let owner_map = owners.into_iter().flatten().collect::<Vec<_>>();
95            if !fence.matches_owner_map(&owner_map) {
96                return Err(
97                    "committed manifest vnode owners do not match the assignment fence".into(),
98                );
99            }
100            let subscription_manifests = manifests
101                .iter()
102                .filter_map(|(manifest, _)| {
103                    manifest
104                        .subscription_output
105                        .as_ref()
106                        .map(|output| (output, manifest.owned_vnodes.as_slice()))
107                })
108                .collect::<Vec<_>>();
109            if !subscription_manifests.is_empty() {
110                if subscription_manifests.len() != manifests.len() {
111                    return Err(
112                        "participant manifests do not agree on subscription output presence".into(),
113                    );
114                }
115                if subscription_manifests.iter().any(|(manifest, _)| {
116                    manifest.streams.iter().any(|stream| {
117                        stream.distribution_certificate.pipeline_identity != self.pipeline_identity
118                    })
119                }) {
120                    return Err(
121                        "subscription output pipeline identity differs from the committed index"
122                            .into(),
123                    );
124                }
125                merge_node_subscription_manifests(
126                    self.epoch,
127                    self.checkpoint_id,
128                    fence,
129                    &subscription_manifests,
130                )
131                .map_err(|error| format!("committed subscription output is invalid: {error}"))?;
132            }
133        }
134        Ok(())
135    }
136}
laminar_core/checkpoint/committed_checkpoint/participant_validation.rs

laminar_core/checkpoint/committed_checkpoint/
participant_validation.rs
