laminar_core/checkpoint/subscription/
node_manifest.rs

1use std::collections::BTreeMap;
2
3use serde::Serialize;
4
5use super::{
6    OutputDistributionCertificate, OutputPartitionId, OutputSegmentRef, PartitionFrontier,
7    PartitionSequence, SubscriptionCheckpointManifest, SubscriptionContractError,
8    SubscriptionDigest, SubscriptionProtocolVersion, MAX_OUTPUT_SEGMENTS_PER_MANIFEST,
9    MAX_SUBSCRIPTION_MANIFEST_BYTES,
10};
11use crate::checkpoint::{canonical_json_bytes, CheckpointAssignmentFence};
12use crate::state::KeyGroupCount;
13
14/// One participant-owned output interval in a checkpoint attempt.
15#[derive(
16    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize,
17)]
18#[serde(deny_unknown_fields)]
19pub struct NodePartitionRange {
20    /// Stable vnode output partition.
21    pub partition: OutputPartitionId,
22    /// First sequence introduced after the preceding committed cut.
23    pub first_sequence: PartitionSequence,
24    /// First sequence not covered by this checkpoint.
25    pub through_sequence: PartitionSequence,
26}
27
28/// One stream's participant-local portion of a checkpointed output cut.
29#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
30#[serde(deny_unknown_fields)]
31pub struct NodeSubscriptionStreamManifest {
32    /// Planner-owned proof of the stable output distribution.
33    pub distribution_certificate: OutputDistributionCertificate,
34    /// Canonical participant-owned partition ranges.
35    pub ranges: Vec<NodePartitionRange>,
36    /// Canonical immutable segments covering every non-empty range exactly.
37    pub segments: Vec<OutputSegmentRef>,
38}
39
40impl NodeSubscriptionStreamManifest {
41    fn validate(
42        &self,
43        key_group_count: KeyGroupCount,
44        owned_vnodes: &[u16],
45    ) -> Result<(), SubscriptionContractError> {
46        self.distribution_certificate.validate(key_group_count)?;
47        if self.ranges.is_empty()
48            || !self
49                .ranges
50                .windows(2)
51                .all(|pair| pair[0].partition < pair[1].partition)
52        {
53            return Err(SubscriptionContractError::NonCanonicalPartitionRanges);
54        }
55        for range in &self.ranges {
56            range.partition.validate(key_group_count)?;
57            if !self
58                .distribution_certificate
59                .distribution
60                .contains(range.partition)
61                || owned_vnodes.binary_search(&range.partition.get()).is_err()
62            {
63                return Err(SubscriptionContractError::PartitionOwnerMismatch {
64                    partition: range.partition.get(),
65                });
66            }
67            if range.first_sequence > range.through_sequence {
68                return Err(SubscriptionContractError::InvalidSegmentRange);
69            }
70        }
71        validate_node_segments(self)
72    }
73}
74
75fn validate_node_segments(
76    stream: &NodeSubscriptionStreamManifest,
77) -> Result<(), SubscriptionContractError> {
78    if stream.segments.len() > MAX_OUTPUT_SEGMENTS_PER_MANIFEST {
79        return Err(SubscriptionContractError::TooManySegments {
80            actual: stream.segments.len(),
81            limit: MAX_OUTPUT_SEGMENTS_PER_MANIFEST,
82        });
83    }
84    if !stream.segments.windows(2).all(|pair| {
85        (pair[0].partition, pair[0].first_sequence) < (pair[1].partition, pair[1].first_sequence)
86    }) {
87        return Err(SubscriptionContractError::NonCanonicalSegments);
88    }
89    for segment in &stream.segments {
90        segment.validate(&stream.distribution_certificate)?;
91    }
92    let mut segment_index = 0;
93    for range in &stream.ranges {
94        let mut cursor = range.first_sequence;
95        while let Some(segment) = stream.segments.get(segment_index) {
96            if segment.partition < range.partition {
97                return Err(SubscriptionContractError::NonCanonicalSegments);
98            }
99            if segment.partition != range.partition {
100                break;
101            }
102            if segment.first_sequence != cursor {
103                return Err(SubscriptionContractError::SequenceGap {
104                    partition: range.partition.get(),
105                    expected: cursor.get(),
106                    actual: segment.first_sequence.get(),
107                });
108            }
109            cursor = segment.exclusive_end_sequence;
110            if cursor > range.through_sequence {
111                return Err(SubscriptionContractError::SegmentBeyondFrontier);
112            }
113            segment_index += 1;
114        }
115        if cursor != range.through_sequence {
116            return Err(SubscriptionContractError::SequenceGap {
117                partition: range.partition.get(),
118                expected: range.through_sequence.get(),
119                actual: cursor.get(),
120            });
121        }
122    }
123    if segment_index != stream.segments.len() {
124        return Err(SubscriptionContractError::NonCanonicalSegments);
125    }
126    Ok(())
127}
128
129/// Participant-local subscription output bound into its checkpoint manifest.
130#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
131#[serde(deny_unknown_fields)]
132pub struct NodeSubscriptionManifest {
133    /// Exact subscription protocol.
134    pub protocol_version: SubscriptionProtocolVersion,
135    /// Checkpoint epoch.
136    pub epoch: u64,
137    /// Checkpoint id; equal to `epoch`.
138    pub checkpoint_id: u64,
139    /// Participant publishing this manifest.
140    pub participant_id: u64,
141    /// Exact assignment and process-incarnation fence.
142    pub assignment_certificate: CheckpointAssignmentFence,
143    /// Canonical stream inventory.
144    pub streams: Vec<NodeSubscriptionStreamManifest>,
145    /// SHA-256 of every preceding field.
146    pub manifest_digest: SubscriptionDigest,
147}
148
149impl NodeSubscriptionManifest {
150    /// Compute and install the canonical participant-manifest digest.
151    ///
152    /// # Errors
153    /// Returns the first malformed binding, range, segment, or size error.
154    pub fn seal(&mut self, owned_vnodes: &[u16]) -> Result<(), SubscriptionContractError> {
155        self.manifest_digest = self.computed_digest()?;
156        self.validate(owned_vnodes)
157    }
158
159    /// Validate the exact participant binding and owned partition ranges.
160    ///
161    /// # Errors
162    /// Returns the first malformed binding, range, segment, digest, or size error.
163    pub fn validate(&self, owned_vnodes: &[u16]) -> Result<(), SubscriptionContractError> {
164        self.protocol_version.validate()?;
165        if self.epoch == 0 || self.checkpoint_id != self.epoch {
166            return Err(SubscriptionContractError::CheckpointAttempt);
167        }
168        if self.participant_id == 0
169            || !self.assignment_certificate.is_canonical()
170            || self
171                .assignment_certificate
172                .participant_incarnation(self.participant_id)
173                .is_none()
174        {
175            return Err(SubscriptionContractError::NodeParticipant);
176        }
177        let key_group_count = KeyGroupCount::try_from(self.assignment_certificate.vnode_count)
178            .map_err(|_| SubscriptionContractError::AssignmentCertificate)?;
179        if !self.streams.windows(2).all(|pair| {
180            pair[0].distribution_certificate.stream_id < pair[1].distribution_certificate.stream_id
181        }) {
182            return Err(SubscriptionContractError::NonCanonicalNodeStreams);
183        }
184        for stream in &self.streams {
185            stream.validate(key_group_count, owned_vnodes)?;
186        }
187        if self.manifest_digest != self.computed_digest()? {
188            return Err(SubscriptionContractError::NodeManifestDigest);
189        }
190        let encoded = canonical_json_bytes(self)
191            .map_err(|error| SubscriptionContractError::Encode(error.to_string()))?;
192        if encoded.len() > MAX_SUBSCRIPTION_MANIFEST_BYTES {
193            return Err(SubscriptionContractError::ManifestTooLarge {
194                actual: encoded.len(),
195                limit: MAX_SUBSCRIPTION_MANIFEST_BYTES,
196            });
197        }
198        Ok(())
199    }
200
201    fn computed_digest(&self) -> Result<SubscriptionDigest, SubscriptionContractError> {
202        let body = NodeManifestBody {
203            protocol_version: self.protocol_version,
204            epoch: self.epoch,
205            checkpoint_id: self.checkpoint_id,
206            participant_id: self.participant_id,
207            assignment_certificate: &self.assignment_certificate,
208            streams: &self.streams,
209        };
210        let encoded = canonical_json_bytes(&body)
211            .map_err(|error| SubscriptionContractError::Encode(error.to_string()))?;
212        Ok(SubscriptionDigest::for_bytes(
213            b"laminardb-node-subscription-manifest-v1",
214            &encoded,
215        ))
216    }
217}
218
219#[derive(Serialize)]
220struct NodeManifestBody<'a> {
221    protocol_version: SubscriptionProtocolVersion,
222    epoch: u64,
223    checkpoint_id: u64,
224    participant_id: u64,
225    assignment_certificate: &'a CheckpointAssignmentFence,
226    streams: &'a [NodeSubscriptionStreamManifest],
227}
228
229/// Whole-cluster stream cut reconstructed from exact participant manifests.
230#[derive(Debug, Clone, PartialEq, Eq)]
231pub struct MergedSubscriptionCheckpoint {
232    /// Canonical checkpoint manifest exposed to committed readers.
233    pub manifest: SubscriptionCheckpointManifest,
234    /// Exact newly committed range per output partition.
235    pub ranges: Vec<NodePartitionRange>,
236}
237
238impl MergedSubscriptionCheckpoint {
239    /// Validate this cut's starting frontiers against an explicitly loaded predecessor.
240    ///
241    /// A missing predecessor is valid only for ranges beginning at sequence zero. A stream absent
242    /// from the predecessor is a new generation and follows the same rule.
243    ///
244    /// # Errors
245    /// Returns a distribution mismatch or the first partition sequence gap.
246    pub fn validate_continuity(
247        &self,
248        predecessor: Option<&SubscriptionCheckpointManifest>,
249    ) -> Result<(), SubscriptionContractError> {
250        let predecessor = predecessor
251            .filter(|manifest| manifest.stream_generation == self.manifest.stream_generation);
252        if let Some(predecessor) = predecessor {
253            self.manifest
254                .distribution_certificate
255                .require_match(&predecessor.distribution_certificate)?;
256        }
257        for range in &self.ranges {
258            let expected = predecessor
259                .and_then(|manifest| {
260                    manifest
261                        .frontiers
262                        .binary_search_by_key(&range.partition, |frontier| frontier.partition)
263                        .ok()
264                        .map(|index| manifest.frontiers[index].through_sequence)
265                })
266                .unwrap_or(PartitionSequence::FIRST);
267            if range.first_sequence != expected {
268                return Err(SubscriptionContractError::SequenceGap {
269                    partition: range.partition.get(),
270                    expected: expected.get(),
271                    actual: range.first_sequence.get(),
272                });
273            }
274        }
275        Ok(())
276    }
277}
278
279struct MergedStream {
280    certificate: OutputDistributionCertificate,
281    ranges: Vec<NodePartitionRange>,
282    segments: Vec<OutputSegmentRef>,
283}
284
285/// Merge one complete participant roster into canonical whole-cluster subscription cuts.
286/// Each manifest is paired with the exact owned-vnode roster from its enclosing checkpoint.
287///
288/// # Errors
289/// Returns an error for a missing participant manifest, certificate disagreement, duplicate or
290/// missing partition, wrong owner contribution, sequence gap, or malformed segment.
291pub fn merge_node_subscription_manifests(
292    epoch: u64,
293    checkpoint_id: u64,
294    assignment_certificate: &CheckpointAssignmentFence,
295    manifests: &[(&NodeSubscriptionManifest, &[u16])],
296) -> Result<Vec<MergedSubscriptionCheckpoint>, SubscriptionContractError> {
297    let expected_participants = assignment_certificate.participant_ids();
298    let actual_participants = manifests
299        .iter()
300        .map(|(manifest, _)| manifest.participant_id)
301        .collect::<Vec<_>>();
302    if actual_participants != expected_participants {
303        return Err(SubscriptionContractError::NodeParticipant);
304    }
305    validate_node_ownership(assignment_certificate, manifests)?;
306    let mut streams = BTreeMap::<String, MergedStream>::new();
307    for (node, owned_vnodes) in manifests {
308        if node.epoch != epoch
309            || node.checkpoint_id != checkpoint_id
310            || node.assignment_certificate != *assignment_certificate
311        {
312            return Err(SubscriptionContractError::AssignmentCertificate);
313        }
314        node.validate(owned_vnodes)?;
315        for stream in &node.streams {
316            let stream_id = stream.distribution_certificate.stream_id.clone();
317            let merged = streams.entry(stream_id).or_insert_with(|| MergedStream {
318                certificate: stream.distribution_certificate.clone(),
319                ranges: Vec::new(),
320                segments: Vec::new(),
321            });
322            merged
323                .certificate
324                .require_match(&stream.distribution_certificate)?;
325            merged.ranges.extend_from_slice(&stream.ranges);
326            merged.segments.extend_from_slice(&stream.segments);
327        }
328    }
329
330    let mut merged_checkpoints = Vec::with_capacity(streams.len());
331    for (_, mut stream) in streams {
332        stream.ranges.sort_unstable_by_key(|range| range.partition);
333        if !stream
334            .ranges
335            .windows(2)
336            .all(|pair| pair[0].partition < pair[1].partition)
337        {
338            return Err(SubscriptionContractError::NonCanonicalPartitionRanges);
339        }
340        stream
341            .segments
342            .sort_unstable_by_key(|segment| (segment.partition, segment.first_sequence));
343        let frontiers = stream
344            .ranges
345            .iter()
346            .map(|range| PartitionFrontier {
347                partition: range.partition,
348                through_sequence: range.through_sequence,
349            })
350            .collect();
351        let mut manifest = SubscriptionCheckpointManifest {
352            protocol_version: SubscriptionProtocolVersion::CURRENT,
353            stream_generation: stream.certificate.stream_generation,
354            schema_fingerprint: stream.certificate.schema_fingerprint,
355            distribution_certificate: stream.certificate,
356            epoch,
357            checkpoint_id,
358            assignment_certificate: assignment_certificate.clone(),
359            frontiers,
360            segments: stream.segments,
361            manifest_digest: SubscriptionDigest::from_bytes([0; 32]),
362        };
363        manifest.seal()?;
364        merged_checkpoints.push(MergedSubscriptionCheckpoint {
365            manifest,
366            ranges: stream.ranges,
367        });
368    }
369    Ok(merged_checkpoints)
370}
371
372fn validate_node_ownership(
373    assignment_certificate: &CheckpointAssignmentFence,
374    manifests: &[(&NodeSubscriptionManifest, &[u16])],
375) -> Result<(), SubscriptionContractError> {
376    let vnode_count = usize::try_from(assignment_certificate.vnode_count)
377        .map_err(|_| SubscriptionContractError::AssignmentCertificate)?;
378    let mut owners = vec![None; vnode_count];
379    for (manifest, owned_vnodes) in manifests {
380        if owned_vnodes.is_empty() || !owned_vnodes.windows(2).all(|pair| pair[0] < pair[1]) {
381            return Err(SubscriptionContractError::AssignmentCertificate);
382        }
383        for vnode in *owned_vnodes {
384            let owner = owners
385                .get_mut(usize::from(*vnode))
386                .ok_or(SubscriptionContractError::AssignmentCertificate)?;
387            if owner.replace(manifest.participant_id).is_some() {
388                return Err(SubscriptionContractError::AssignmentCertificate);
389            }
390        }
391    }
392    if owners.iter().any(Option::is_none) {
393        return Err(SubscriptionContractError::AssignmentCertificate);
394    }
395    let owner_map = owners.into_iter().flatten().collect::<Vec<_>>();
396    if !assignment_certificate.matches_owner_map(&owner_map) {
397        return Err(SubscriptionContractError::AssignmentCertificate);
398    }
399    Ok(())
400}
401
402#[cfg(test)]
403mod tests {
404    use super::*;
405    use crate::checkpoint::{
406        ChangelogMode, CheckpointParticipant, PipelineIdentity, StreamGeneration,
407        OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION,
408    };
409    use crate::state::PARTITIONING_ABI_VERSION;
410
411    fn certificate() -> OutputDistributionCertificate {
412        OutputDistributionCertificate {
413            version: OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION,
414            protocol_version: SubscriptionProtocolVersion::CURRENT,
415            stream_id: "positions".into(),
416            catalog_generation: 1,
417            stream_generation: StreamGeneration::from_digest(SubscriptionDigest::from_bytes(
418                [1; 32],
419            )),
420            final_operator_id: "stream:positions".into(),
421            distribution: super::super::OutputDistribution::VnodePartitioned {
422                key_expressions_fingerprint: SubscriptionDigest::from_bytes([2; 32]),
423                partition_abi: PARTITIONING_ABI_VERSION,
424                vnode_count: 4,
425            },
426            schema_fingerprint: SubscriptionDigest::from_bytes([3; 32]),
427            changelog_mode: ChangelogMode::WeightedRetractInsert,
428            history_retention_bytes: 0,
429            query_fingerprint: SubscriptionDigest::from_bytes([4; 32]),
430            pipeline_identity: PipelineIdentity::empty(),
431        }
432    }
433
434    fn assignment() -> CheckpointAssignmentFence {
435        CheckpointAssignmentFence::from_owner_map(
436            5,
437            &[1, 2, 1, 2],
438            vec![
439                CheckpointParticipant {
440                    node_id: 1,
441                    boot_incarnation: uuid::Uuid::parse_str("11111111-1111-4111-8111-111111111111")
442                        .unwrap(),
443                },
444                CheckpointParticipant {
445                    node_id: 2,
446                    boot_incarnation: uuid::Uuid::parse_str("22222222-2222-4222-8222-222222222222")
447                        .unwrap(),
448                },
449            ],
450        )
451        .unwrap()
452    }
453
454    fn node(
455        participant_id: u64,
456        partitions: &[u16],
457        first_sequence: u64,
458        through_sequence: u64,
459    ) -> NodeSubscriptionManifest {
460        let assignment = assignment();
461        let mut manifest = NodeSubscriptionManifest {
462            protocol_version: SubscriptionProtocolVersion::CURRENT,
463            epoch: 10,
464            checkpoint_id: 10,
465            participant_id,
466            assignment_certificate: assignment,
467            streams: vec![NodeSubscriptionStreamManifest {
468                distribution_certificate: certificate(),
469                ranges: partitions
470                    .iter()
471                    .map(|partition| NodePartitionRange {
472                        partition: OutputPartitionId::new(*partition),
473                        first_sequence: PartitionSequence::new(first_sequence),
474                        through_sequence: PartitionSequence::new(through_sequence),
475                    })
476                    .collect(),
477                segments: Vec::new(),
478            }],
479            manifest_digest: SubscriptionDigest::from_bytes([0; 32]),
480        };
481        manifest.seal(partitions).unwrap();
482        manifest
483    }
484
485    fn segment(partition: u16, first: u64, end: u64, suffix: &str) -> OutputSegmentRef {
486        OutputSegmentRef {
487            protocol_version: SubscriptionProtocolVersion::CURRENT,
488            object_key: format!("subscription-output/test/{partition}/{suffix}.arrow"),
489            stream_generation: certificate().stream_generation,
490            partition: OutputPartitionId::new(partition),
491            first_sequence: PartitionSequence::new(first),
492            exclusive_end_sequence: PartitionSequence::new(end),
493            frame_count: end - first,
494            row_count: 1,
495            encoded_length: 1,
496            schema_fingerprint: certificate().schema_fingerprint,
497            payload_digest: SubscriptionDigest::from_bytes([5; 32]),
498        }
499    }
500
501    #[test]
502    fn participant_ranges_merge_into_one_exact_frontier_vector() {
503        let first = node(1, &[0, 2], 0, 0);
504        let second = node(2, &[1, 3], 0, 0);
505        let merged = merge_node_subscription_manifests(
506            10,
507            10,
508            &assignment(),
509            &[(&first, &[0, 2]), (&second, &[1, 3])],
510        )
511        .unwrap();
512        assert_eq!(merged.len(), 1);
513        assert_eq!(
514            merged[0]
515                .manifest
516                .frontiers
517                .iter()
518                .map(|frontier| frontier.partition.get())
519                .collect::<Vec<_>>(),
520            vec![0, 1, 2, 3]
521        );
522    }
523
524    #[test]
525    fn missing_participant_and_wrong_owner_fail_closed() {
526        let first = node(1, &[0, 2], 0, 0);
527        assert_eq!(
528            merge_node_subscription_manifests(10, 10, &assignment(), &[(&first, &[0, 2])]),
529            Err(SubscriptionContractError::NodeParticipant)
530        );
531
532        let mut wrong_owner = node(1, &[0, 2], 0, 0);
533        wrong_owner.streams[0].ranges[1].partition = OutputPartitionId::new(1);
534        wrong_owner.manifest_digest = wrong_owner.computed_digest().unwrap();
535        assert_eq!(
536            wrong_owner.validate(&[0, 2]),
537            Err(SubscriptionContractError::PartitionOwnerMismatch { partition: 1 })
538        );
539    }
540
541    #[test]
542    fn predecessor_frontier_gap_is_explicit() {
543        let predecessor_first = node(1, &[0, 2], 0, 0);
544        let predecessor_second = node(2, &[1, 3], 0, 0);
545        let predecessor = merge_node_subscription_manifests(
546            10,
547            10,
548            &assignment(),
549            &[
550                (&predecessor_first, &[0, 2]),
551                (&predecessor_second, &[1, 3]),
552            ],
553        )
554        .unwrap()
555        .remove(0);
556
557        let current_first = node(1, &[0, 2], 1, 1);
558        let current_second = node(2, &[1, 3], 1, 1);
559        let current = merge_node_subscription_manifests(
560            10,
561            10,
562            &assignment(),
563            &[(&current_first, &[0, 2]), (&current_second, &[1, 3])],
564        )
565        .unwrap()
566        .remove(0);
567        assert_eq!(
568            current.validate_continuity(Some(&predecessor.manifest)),
569            Err(SubscriptionContractError::SequenceGap {
570                partition: 0,
571                expected: 0,
572                actual: 1,
573            })
574        );
575    }
576
577    #[test]
578    fn duplicate_segment_and_partition_gap_are_rejected() {
579        let mut duplicate = node(1, &[0, 2], 0, 0);
580        duplicate.streams[0].ranges[0].through_sequence = PartitionSequence::new(1);
581        let output = segment(0, 0, 1, "duplicate");
582        duplicate.streams[0].segments = vec![output.clone(), output];
583        duplicate.manifest_digest = duplicate.computed_digest().unwrap();
584        assert_eq!(
585            duplicate.validate(&[0, 2]),
586            Err(SubscriptionContractError::NonCanonicalSegments)
587        );
588
589        let mut gap = node(1, &[0, 2], 0, 0);
590        gap.streams[0].ranges[0].through_sequence = PartitionSequence::new(3);
591        gap.streams[0].segments = vec![segment(0, 0, 1, "first"), segment(0, 2, 3, "last")];
592        gap.manifest_digest = gap.computed_digest().unwrap();
593        assert_eq!(
594            gap.validate(&[0, 2]),
595            Err(SubscriptionContractError::SequenceGap {
596                partition: 0,
597                expected: 1,
598                actual: 2,
599            })
600        );
601    }
602
603    #[test]
604    fn merge_revalidates_node_digest_segments_and_assignment_ownership() {
605        let mut unvalidated = node(1, &[0, 2], 0, 0);
606        unvalidated.streams[0].ranges[0].through_sequence = PartitionSequence::new(1);
607        let second = node(2, &[1, 3], 0, 0);
608        assert_eq!(
609            merge_node_subscription_manifests(
610                10,
611                10,
612                &assignment(),
613                &[(&unvalidated, &[0, 2]), (&second, &[1, 3])],
614            ),
615            Err(SubscriptionContractError::SequenceGap {
616                partition: 0,
617                expected: 1,
618                actual: 0,
619            })
620        );
621
622        let first = node(1, &[0, 2], 0, 0);
623        assert_eq!(
624            merge_node_subscription_manifests(
625                10,
626                10,
627                &assignment(),
628                &[(&first, &[0, 1]), (&second, &[2, 3])],
629            ),
630            Err(SubscriptionContractError::AssignmentCertificate)
631        );
632    }
633}
laminar_core/checkpoint/subscription/node_manifest.rs

laminar_core/checkpoint/subscription/
node_manifest.rs
