laminar_core/checkpoint/subscription/
manifest.rs

1use serde::Serialize;
2
3use super::{
4    OutputDistribution, OutputDistributionCertificate, OutputPartitionId, PartitionSequence,
5    StreamGeneration, SubscriptionDigest, SubscriptionProtocolVersion,
6};
7use crate::checkpoint::{canonical_json_bytes, CheckpointAssignmentFence};
8use crate::state::KeyGroupCount;
9
10/// Hard segment-reference count bound for one stream checkpoint manifest.
11pub const MAX_OUTPUT_SEGMENTS_PER_MANIFEST: usize = 65_536;
12
13/// Hard canonical encoded size bound for one stream checkpoint manifest.
14pub const MAX_SUBSCRIPTION_MANIFEST_BYTES: usize = 8 * 1024 * 1024;
15
16/// Hard encoded-object bound for one immutable Arrow output segment.
17pub const MAX_OUTPUT_SEGMENT_BYTES: usize = 16 * 1024 * 1024;
18
19/// Hard frame-count bound for one immutable Arrow output segment.
20pub const MAX_OUTPUT_FRAMES_PER_SEGMENT: u64 = 1_024;
21
22const MAX_OBJECT_KEY_BYTES: usize = 2_048;
23
24/// First partition sequence not covered by a committed checkpoint.
25#[derive(
26    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize,
27)]
28#[serde(deny_unknown_fields)]
29pub struct PartitionFrontier {
30    /// Stable vnode output partition.
31    pub partition: OutputPartitionId,
32    /// Every sequence below this exclusive frontier is covered by the checkpoint.
33    pub through_sequence: PartitionSequence,
34}
35
36/// Exact immutable object containing a contiguous partition-frame range.
37#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
38#[serde(deny_unknown_fields)]
39pub struct OutputSegmentRef {
40    /// Exact segment envelope protocol.
41    pub protocol_version: SubscriptionProtocolVersion,
42    /// Provider-neutral object key inside the checkpoint namespace.
43    pub object_key: String,
44    /// Durable stream incarnation.
45    pub stream_generation: StreamGeneration,
46    /// Stable vnode partition.
47    pub partition: OutputPartitionId,
48    /// First frame sequence in the segment.
49    pub first_sequence: PartitionSequence,
50    /// Exclusive frame sequence at the end of the segment.
51    pub exclusive_end_sequence: PartitionSequence,
52    /// Exact number of frames.
53    pub frame_count: u64,
54    /// Exact number of Arrow rows across all frames.
55    pub row_count: u64,
56    /// Exact encoded object length.
57    pub encoded_length: u64,
58    /// Fingerprint of every frame's user schema.
59    pub schema_fingerprint: SubscriptionDigest,
60    /// SHA-256 of the exact object bytes.
61    pub payload_digest: SubscriptionDigest,
62}
63
64impl OutputSegmentRef {
65    pub(crate) fn validate(
66        &self,
67        certificate: &OutputDistributionCertificate,
68    ) -> Result<(), SubscriptionContractError> {
69        self.protocol_version.validate()?;
70        if self.object_key.is_empty()
71            || self.object_key.len() > MAX_OBJECT_KEY_BYTES
72            || !self.object_key.starts_with("subscription-output/")
73            || self.object_key.starts_with('/')
74            || self.object_key.contains('\\')
75            || self
76                .object_key
77                .split('/')
78                .any(|component| component.is_empty() || matches!(component, "." | ".."))
79        {
80            return Err(SubscriptionContractError::NonCanonicalObjectKey);
81        }
82        if self.stream_generation != certificate.stream_generation {
83            return Err(SubscriptionContractError::GenerationMismatch);
84        }
85        if !certificate.distribution.contains(self.partition) {
86            return Err(SubscriptionContractError::PartitionOutOfRange {
87                partition: self.partition.get(),
88                vnode_count: certificate.distribution.partition_count(),
89            });
90        }
91        if self.schema_fingerprint != certificate.schema_fingerprint {
92            return Err(SubscriptionContractError::SchemaMismatch);
93        }
94        self.payload_digest.validate("payload_digest")?;
95        let range = self
96            .exclusive_end_sequence
97            .get()
98            .checked_sub(self.first_sequence.get())
99            .ok_or(SubscriptionContractError::InvalidSegmentRange)?;
100        if range == 0
101            || range != self.frame_count
102            || self.frame_count > MAX_OUTPUT_FRAMES_PER_SEGMENT
103            || self.row_count == 0
104            || self.encoded_length == 0
105            || self.encoded_length > u64::try_from(MAX_OUTPUT_SEGMENT_BYTES).unwrap_or(u64::MAX)
106        {
107            return Err(SubscriptionContractError::InvalidSegmentRange);
108        }
109        Ok(())
110    }
111}
112
113/// Canonical subscription output bound into one whole-cluster checkpoint.
114#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
115#[serde(deny_unknown_fields)]
116pub struct SubscriptionCheckpointManifest {
117    /// Exact subscription protocol.
118    pub protocol_version: SubscriptionProtocolVersion,
119    /// Planner-owned distribution proof.
120    pub distribution_certificate: OutputDistributionCertificate,
121    /// Durable stream incarnation, repeated for direct validation.
122    pub stream_generation: StreamGeneration,
123    /// Checkpoint epoch.
124    pub epoch: u64,
125    /// Checkpoint id; equal to `epoch`.
126    pub checkpoint_id: u64,
127    /// Exact assignment and process-incarnation fence.
128    pub assignment_certificate: CheckpointAssignmentFence,
129    /// Fingerprint of the unchanged user output schema.
130    pub schema_fingerprint: SubscriptionDigest,
131    /// Complete canonical output-partition roster and exclusive frontiers.
132    pub frontiers: Vec<PartitionFrontier>,
133    /// Canonically sorted immutable segment references.
134    pub segments: Vec<OutputSegmentRef>,
135    /// SHA-256 of every preceding field in this manifest.
136    pub manifest_digest: SubscriptionDigest,
137}
138
139impl SubscriptionCheckpointManifest {
140    /// Compute and install the canonical manifest digest.
141    ///
142    /// # Errors
143    /// Returns the first encoding, binding, roster, range, or size validation failure.
144    pub fn seal(&mut self) -> Result<(), SubscriptionContractError> {
145        self.manifest_digest = self.computed_digest()?;
146        self.validate()
147    }
148
149    /// Validate canonical shape, exact roster, bindings, and digest.
150    ///
151    /// # Errors
152    /// Returns the first malformed, mismatched, discontinuous, or oversized field.
153    pub fn validate(&self) -> Result<(), SubscriptionContractError> {
154        ensure_encoded_size_within_limit(self)?;
155        self.protocol_version.validate()?;
156        if self.epoch == 0 || self.checkpoint_id != self.epoch {
157            return Err(SubscriptionContractError::CheckpointAttempt);
158        }
159        let key_group_count = KeyGroupCount::try_from(self.assignment_certificate.vnode_count)
160            .map_err(|_| SubscriptionContractError::AssignmentCertificate)?;
161        if !self.assignment_certificate.is_canonical() {
162            return Err(SubscriptionContractError::AssignmentCertificate);
163        }
164        self.distribution_certificate.validate(key_group_count)?;
165        if self.stream_generation != self.distribution_certificate.stream_generation {
166            return Err(SubscriptionContractError::GenerationMismatch);
167        }
168        if self.schema_fingerprint != self.distribution_certificate.schema_fingerprint {
169            return Err(SubscriptionContractError::SchemaMismatch);
170        }
171        validate_frontiers(
172            &self.frontiers,
173            &self.distribution_certificate.distribution,
174            key_group_count,
175        )?;
176        validate_segments(
177            &self.segments,
178            &self.frontiers,
179            &self.distribution_certificate,
180        )?;
181        if self.manifest_digest != self.computed_digest()? {
182            return Err(SubscriptionContractError::ManifestDigest);
183        }
184        let encoded = canonical_json_bytes(self)
185            .map_err(|error| SubscriptionContractError::Encode(error.to_string()))?;
186        if encoded.len() > MAX_SUBSCRIPTION_MANIFEST_BYTES {
187            return Err(SubscriptionContractError::ManifestTooLarge {
188                actual: encoded.len(),
189                limit: MAX_SUBSCRIPTION_MANIFEST_BYTES,
190            });
191        }
192        Ok(())
193    }
194
195    fn computed_digest(&self) -> Result<SubscriptionDigest, SubscriptionContractError> {
196        let body = ManifestBody {
197            protocol_version: self.protocol_version,
198            distribution_certificate: &self.distribution_certificate,
199            stream_generation: self.stream_generation,
200            epoch: self.epoch,
201            checkpoint_id: self.checkpoint_id,
202            assignment_certificate: &self.assignment_certificate,
203            schema_fingerprint: self.schema_fingerprint,
204            frontiers: &self.frontiers,
205            segments: &self.segments,
206        };
207        ensure_encoded_size_within_limit(&body)?;
208        let encoded = canonical_json_bytes(&body)
209            .map_err(|error| SubscriptionContractError::Encode(error.to_string()))?;
210        Ok(SubscriptionDigest::for_bytes(
211            b"laminardb-subscription-checkpoint-manifest-v1",
212            &encoded,
213        ))
214    }
215}
216
217struct EncodedSizeWriter {
218    written: usize,
219    overflow_at: Option<usize>,
220}
221
222impl std::io::Write for EncodedSizeWriter {
223    fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
224        let next = self.written.saturating_add(bytes.len());
225        if next > MAX_SUBSCRIPTION_MANIFEST_BYTES {
226            self.overflow_at = Some(next);
227            return Err(std::io::Error::new(
228                std::io::ErrorKind::OutOfMemory,
229                "subscription manifest exceeds its encoded-size bound",
230            ));
231        }
232        self.written = next;
233        Ok(bytes.len())
234    }
235
236    fn flush(&mut self) -> std::io::Result<()> {
237        Ok(())
238    }
239}
240
241fn ensure_encoded_size_within_limit<T: Serialize>(
242    value: &T,
243) -> Result<(), SubscriptionContractError> {
244    let mut writer = EncodedSizeWriter {
245        written: 0,
246        overflow_at: None,
247    };
248    let result = serde_json::to_writer(&mut writer, value);
249    match (result, writer.overflow_at) {
250        (Ok(()), _) => Ok(()),
251        (Err(_), Some(actual)) => Err(SubscriptionContractError::ManifestTooLarge {
252            actual,
253            limit: MAX_SUBSCRIPTION_MANIFEST_BYTES,
254        }),
255        (Err(error), None) => Err(SubscriptionContractError::Encode(error.to_string())),
256    }
257}
258
259#[derive(Serialize)]
260struct ManifestBody<'a> {
261    protocol_version: SubscriptionProtocolVersion,
262    distribution_certificate: &'a OutputDistributionCertificate,
263    stream_generation: StreamGeneration,
264    epoch: u64,
265    checkpoint_id: u64,
266    assignment_certificate: &'a CheckpointAssignmentFence,
267    schema_fingerprint: SubscriptionDigest,
268    frontiers: &'a [PartitionFrontier],
269    segments: &'a [OutputSegmentRef],
270}
271
272fn validate_frontiers(
273    frontiers: &[PartitionFrontier],
274    distribution: &OutputDistribution,
275    key_group_count: KeyGroupCount,
276) -> Result<(), SubscriptionContractError> {
277    if frontiers.is_empty() {
278        return Err(SubscriptionContractError::MissingPartitionFrontier);
279    }
280    for frontier in frontiers {
281        frontier.partition.validate(key_group_count)?;
282        if !distribution.contains(frontier.partition) {
283            return Err(SubscriptionContractError::PartitionOutOfRange {
284                partition: frontier.partition.get(),
285                vnode_count: distribution.partition_count(),
286            });
287        }
288    }
289    if !frontiers
290        .windows(2)
291        .all(|pair| pair[0].partition < pair[1].partition)
292    {
293        return Err(SubscriptionContractError::NonCanonicalFrontiers);
294    }
295    let exact_roster = match distribution {
296        OutputDistribution::VnodePartitioned { vnode_count, .. } => {
297            frontiers.len() == usize::from(*vnode_count)
298                && frontiers
299                    .iter()
300                    .map(|frontier| frontier.partition.get())
301                    .eq(0..*vnode_count)
302        }
303        OutputDistribution::Singleton { partition } => {
304            frontiers.len() == 1 && frontiers[0].partition == *partition
305        }
306    };
307    if !exact_roster {
308        return Err(SubscriptionContractError::MissingPartitionFrontier);
309    }
310    Ok(())
311}
312
313fn validate_segments(
314    segments: &[OutputSegmentRef],
315    frontiers: &[PartitionFrontier],
316    certificate: &OutputDistributionCertificate,
317) -> Result<(), SubscriptionContractError> {
318    if segments.len() > MAX_OUTPUT_SEGMENTS_PER_MANIFEST {
319        return Err(SubscriptionContractError::TooManySegments {
320            actual: segments.len(),
321            limit: MAX_OUTPUT_SEGMENTS_PER_MANIFEST,
322        });
323    }
324    for segment in segments {
325        segment.validate(certificate)?;
326        let frontier = frontiers
327            .binary_search_by_key(&segment.partition, |frontier| frontier.partition)
328            .ok()
329            .map(|index| frontiers[index].through_sequence)
330            .ok_or(SubscriptionContractError::MissingPartitionFrontier)?;
331        if segment.exclusive_end_sequence > frontier {
332            return Err(SubscriptionContractError::SegmentBeyondFrontier);
333        }
334    }
335    if !segments.windows(2).all(|pair| {
336        (pair[0].partition, pair[0].first_sequence) < (pair[1].partition, pair[1].first_sequence)
337    }) {
338        return Err(SubscriptionContractError::NonCanonicalSegments);
339    }
340    for pair in segments.windows(2) {
341        if pair[0].partition == pair[1].partition
342            && pair[0].exclusive_end_sequence > pair[1].first_sequence
343        {
344            return Err(SubscriptionContractError::OverlappingSegments);
345        }
346    }
347    Ok(())
348}
349
350/// Structured validation failure for durable subscription contracts.
351#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
352pub enum SubscriptionContractError {
353    /// Unsupported persisted or wire protocol.
354    #[error("unsupported subscription protocol version {actual}; expected {expected}")]
355    ProtocolVersion {
356        /// Decoded protocol version.
357        actual: u16,
358        /// Runtime protocol version.
359        expected: u16,
360    },
361    /// Unsupported planner certificate encoding.
362    #[error("unsupported output-distribution certificate version {actual}; expected {expected}")]
363    DistributionCertificateVersion {
364        /// Decoded certificate version.
365        actual: u16,
366        /// Runtime certificate version.
367        expected: u16,
368    },
369    /// Digest is the reserved all-zero value.
370    #[error("subscription field '{field}' uses the reserved zero digest")]
371    ZeroDigest {
372        /// Field containing the reserved digest.
373        field: &'static str,
374    },
375    /// Catalog generations begin at one.
376    #[error("subscription catalog generation must be nonzero")]
377    ZeroCatalogGeneration,
378    /// Catalog or graph identity is empty, padded, or oversized.
379    #[error("subscription identity '{field}' is not canonical (maximum {max_bytes} bytes)")]
380    NonCanonicalIdentity {
381        /// Identity field.
382        field: &'static str,
383        /// Hard encoded bound.
384        max_bytes: usize,
385    },
386    /// Pipeline identity is not canonical.
387    #[error("subscription pipeline identity is not canonical")]
388    PipelineIdentity,
389    /// Stable partitioning ABI differs.
390    #[error("subscription partition ABI {actual} differs from runtime ABI {expected}")]
391    PartitionAbi {
392        /// Persisted ABI.
393        actual: u16,
394        /// Runtime ABI.
395        expected: u16,
396    },
397    /// Certified vnode domain differs.
398    #[error("subscription vnode count {actual} differs from runtime count {expected}")]
399    VnodeCount {
400        /// Certified vnode count.
401        actual: u16,
402        /// Runtime vnode count.
403        expected: u16,
404    },
405    /// Output partition lies outside its certified vnode domain.
406    #[error("subscription partition {partition} is outside vnode domain 0..{vnode_count}")]
407    PartitionOutOfRange {
408        /// Invalid partition.
409        partition: u16,
410        /// Exclusive vnode-domain end.
411        vnode_count: u16,
412    },
413    /// Global singleton output must use vnode zero.
414    #[error("global singleton output uses non-canonical partition {partition}")]
415    NonCanonicalSingleton {
416        /// Invalid singleton partition.
417        partition: u16,
418    },
419    /// Restored and planned certificates are not identical.
420    #[error("subscription output-distribution certificate mismatch")]
421    DistributionCertificateMismatch,
422    /// Partition sequence cannot advance without wrapping.
423    #[error("subscription partition sequence overflow")]
424    SequenceOverflow,
425    /// Checkpoint attempt is zero or internally inconsistent.
426    #[error("subscription checkpoint must use one nonzero canonical attempt")]
427    CheckpointAttempt,
428    /// Assignment fence is absent or malformed.
429    #[error("subscription assignment certificate is not canonical")]
430    AssignmentCertificate,
431    /// Generation binding differs.
432    #[error("subscription stream generation mismatch")]
433    GenerationMismatch,
434    /// User schema fingerprint differs.
435    #[error("subscription output schema fingerprint mismatch")]
436    SchemaMismatch,
437    /// Frontier vector is not sorted and duplicate-free.
438    #[error("subscription partition frontiers must be strictly ascending and unique")]
439    NonCanonicalFrontiers,
440    /// Complete certified partition roster is absent.
441    #[error("subscription checkpoint is missing a certified partition frontier")]
442    MissingPartitionFrontier,
443    /// Segment object key is not a bounded provider-neutral relative path.
444    #[error("subscription segment object key is not canonical")]
445    NonCanonicalObjectKey,
446    /// Segment range or counts are inconsistent.
447    #[error("subscription segment has an invalid sequence range or count")]
448    InvalidSegmentRange,
449    /// Segment references are not canonically sorted and unique.
450    #[error("subscription segment references are not canonical")]
451    NonCanonicalSegments,
452    /// Segment ranges overlap within one partition.
453    #[error("subscription segment ranges overlap")]
454    OverlappingSegments,
455    /// Segment extends beyond the checkpoint frontier.
456    #[error("subscription segment extends beyond its checkpoint frontier")]
457    SegmentBeyondFrontier,
458    /// Segment reference count exceeds the hard limit.
459    #[error("subscription manifest has {actual} segments; maximum is {limit}")]
460    TooManySegments {
461        /// Segment count.
462        actual: usize,
463        /// Hard segment-count bound.
464        limit: usize,
465    },
466    /// Encoded manifest exceeds the hard limit.
467    #[error("subscription manifest is {actual} bytes; maximum is {limit}")]
468    ManifestTooLarge {
469        /// Encoded byte count.
470        actual: usize,
471        /// Hard encoded-size bound.
472        limit: usize,
473    },
474    /// Canonical metadata encoding failed.
475    #[error("subscription manifest encoding failed: {0}")]
476    Encode(String),
477    /// Manifest body does not match its content digest.
478    #[error("subscription manifest digest mismatch")]
479    ManifestDigest,
480    /// Participant-local stream inventory is not sorted and unique.
481    #[error("subscription node streams must be strictly sorted and unique")]
482    NonCanonicalNodeStreams,
483    /// Participant-local partition ranges are not sorted and unique.
484    #[error("subscription node partition ranges must be strictly sorted and unique")]
485    NonCanonicalPartitionRanges,
486    /// A participant claimed an output partition it does not own.
487    #[error("subscription partition {partition} was published by the wrong owner")]
488    PartitionOwnerMismatch {
489        /// Partition claimed by the wrong participant.
490        partition: u16,
491    },
492    /// Participant identity is missing from its assignment certificate.
493    #[error("subscription node participant binding is invalid")]
494    NodeParticipant,
495    /// Participant manifest body does not match its content digest.
496    #[error("subscription node manifest digest mismatch")]
497    NodeManifestDigest,
498    /// A committed partition interval is discontinuous.
499    #[error(
500        "subscription partition {partition} sequence gap: expected {expected}, found {actual}"
501    )]
502    SequenceGap {
503        /// Partition containing the gap.
504        partition: u16,
505        /// Required next sequence.
506        expected: u64,
507        /// Observed next sequence.
508        actual: u64,
509    },
510}
511
512#[cfg(test)]
513mod tests {
514    use super::*;
515    use crate::checkpoint::{CheckpointParticipant, PipelineIdentity};
516    use crate::state::PARTITIONING_ABI_VERSION;
517
518    fn certificate() -> OutputDistributionCertificate {
519        OutputDistributionCertificate {
520            version: super::super::OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION,
521            protocol_version: SubscriptionProtocolVersion::CURRENT,
522            stream_id: "positions".into(),
523            catalog_generation: 1,
524            stream_generation: StreamGeneration::from_digest(SubscriptionDigest::from_bytes(
525                [1; 32],
526            )),
527            final_operator_id: "stream:positions".into(),
528            distribution: OutputDistribution::VnodePartitioned {
529                key_expressions_fingerprint: SubscriptionDigest::from_bytes([2; 32]),
530                partition_abi: PARTITIONING_ABI_VERSION,
531                vnode_count: 4,
532            },
533            schema_fingerprint: SubscriptionDigest::from_bytes([3; 32]),
534            changelog_mode: super::super::ChangelogMode::WeightedRetractInsert,
535            history_retention_bytes: 0,
536            query_fingerprint: SubscriptionDigest::from_bytes([4; 32]),
537            pipeline_identity: PipelineIdentity::empty(),
538        }
539    }
540
541    fn assignment() -> CheckpointAssignmentFence {
542        CheckpointAssignmentFence::from_owner_map(
543            1,
544            &[7, 7, 7, 7],
545            vec![CheckpointParticipant {
546                node_id: 7,
547                boot_incarnation: uuid::Uuid::parse_str("77777777-7777-4777-8777-777777777777")
548                    .unwrap(),
549            }],
550        )
551        .unwrap()
552    }
553
554    fn manifest() -> SubscriptionCheckpointManifest {
555        let certificate = certificate();
556        let mut manifest = SubscriptionCheckpointManifest {
557            protocol_version: SubscriptionProtocolVersion::CURRENT,
558            stream_generation: certificate.stream_generation,
559            schema_fingerprint: certificate.schema_fingerprint,
560            distribution_certificate: certificate,
561            epoch: 10,
562            checkpoint_id: 10,
563            assignment_certificate: assignment(),
564            frontiers: (0..4)
565                .map(|partition| PartitionFrontier {
566                    partition: OutputPartitionId::new(partition),
567                    through_sequence: PartitionSequence::new(0),
568                })
569                .collect(),
570            segments: Vec::new(),
571            manifest_digest: SubscriptionDigest::from_bytes([0; 32]),
572        };
573        manifest.seal().unwrap();
574        manifest
575    }
576
577    #[test]
578    fn canonical_frontier_vector_round_trips() {
579        let manifest = manifest();
580        let encoded = canonical_json_bytes(&manifest).unwrap();
581        let decoded: SubscriptionCheckpointManifest = serde_json::from_slice(&encoded).unwrap();
582        assert_eq!(decoded, manifest);
583        decoded.validate().unwrap();
584    }
585
586    #[test]
587    fn duplicate_missing_and_out_of_range_partitions_are_rejected() {
588        let mut duplicate = manifest();
589        duplicate.frontiers[2].partition = duplicate.frontiers[1].partition;
590        assert_eq!(
591            duplicate.validate(),
592            Err(SubscriptionContractError::NonCanonicalFrontiers)
593        );
594
595        let mut missing = manifest();
596        missing.frontiers.pop();
597        assert_eq!(
598            missing.validate(),
599            Err(SubscriptionContractError::MissingPartitionFrontier)
600        );
601
602        let mut out_of_range = manifest();
603        out_of_range.frontiers[3].partition = OutputPartitionId::new(4);
604        assert_eq!(
605            out_of_range.validate(),
606            Err(SubscriptionContractError::PartitionOutOfRange {
607                partition: 4,
608                vnode_count: 4,
609            })
610        );
611    }
612
613    #[test]
614    fn generation_and_schema_mismatch_are_rejected_before_digest() {
615        let mut wrong_generation = manifest();
616        wrong_generation.stream_generation =
617            StreamGeneration::from_digest(SubscriptionDigest::from_bytes([8; 32]));
618        assert_eq!(
619            wrong_generation.validate(),
620            Err(SubscriptionContractError::GenerationMismatch)
621        );
622
623        let mut wrong_schema = manifest();
624        wrong_schema.schema_fingerprint = SubscriptionDigest::from_bytes([8; 32]);
625        assert_eq!(
626            wrong_schema.validate(),
627            Err(SubscriptionContractError::SchemaMismatch)
628        );
629    }
630
631    #[test]
632    fn manifest_digest_detects_metadata_mutation() {
633        let mut manifest = manifest();
634        manifest.frontiers[0].through_sequence = PartitionSequence::new(1);
635        assert_eq!(
636            manifest.validate(),
637            Err(SubscriptionContractError::ManifestDigest)
638        );
639    }
640
641    #[test]
642    fn segment_reference_rejects_allocation_sized_counts_and_lengths() {
643        let certificate = certificate();
644        let mut segment = OutputSegmentRef {
645            protocol_version: SubscriptionProtocolVersion::CURRENT,
646            object_key: "subscription-output/deployment/stream/generation/0/segment.arrow".into(),
647            stream_generation: certificate.stream_generation,
648            partition: OutputPartitionId::new(0),
649            first_sequence: PartitionSequence::FIRST,
650            exclusive_end_sequence: PartitionSequence::new(MAX_OUTPUT_FRAMES_PER_SEGMENT + 1),
651            frame_count: MAX_OUTPUT_FRAMES_PER_SEGMENT + 1,
652            row_count: 1,
653            encoded_length: 1,
654            schema_fingerprint: certificate.schema_fingerprint,
655            payload_digest: SubscriptionDigest::from_bytes([5; 32]),
656        };
657        assert_eq!(
658            segment.validate(&certificate),
659            Err(SubscriptionContractError::InvalidSegmentRange)
660        );
661
662        segment.exclusive_end_sequence = PartitionSequence::new(1);
663        segment.frame_count = 1;
664        segment.encoded_length = u64::try_from(MAX_OUTPUT_SEGMENT_BYTES).unwrap() + 1;
665        assert_eq!(
666            segment.validate(&certificate),
667            Err(SubscriptionContractError::InvalidSegmentRange)
668        );
669    }
670
671    #[test]
672    fn segment_reference_requires_the_storage_namespace() {
673        let certificate = certificate();
674        let segment = OutputSegmentRef {
675            protocol_version: SubscriptionProtocolVersion::CURRENT,
676            object_key: "other/deployment/stream/generation/0/segment.arrow".into(),
677            stream_generation: certificate.stream_generation,
678            partition: OutputPartitionId::new(0),
679            first_sequence: PartitionSequence::FIRST,
680            exclusive_end_sequence: PartitionSequence::new(1),
681            frame_count: 1,
682            row_count: 1,
683            encoded_length: 1,
684            schema_fingerprint: certificate.schema_fingerprint,
685            payload_digest: SubscriptionDigest::from_bytes([5; 32]),
686        };
687
688        assert_eq!(
689            segment.validate(&certificate),
690            Err(SubscriptionContractError::NonCanonicalObjectKey)
691        );
692    }
693
694    #[test]
695    fn oversized_manifest_is_rejected_before_digest_encoding() {
696        let mut manifest = manifest();
697        let segment = OutputSegmentRef {
698            protocol_version: SubscriptionProtocolVersion::CURRENT,
699            object_key: format!(
700                "subscription-output/{}",
701                "x".repeat(MAX_OBJECT_KEY_BYTES - "subscription-output/".len())
702            ),
703            stream_generation: manifest.stream_generation,
704            partition: OutputPartitionId::new(0),
705            first_sequence: PartitionSequence::FIRST,
706            exclusive_end_sequence: PartitionSequence::new(1),
707            frame_count: 1,
708            row_count: 1,
709            encoded_length: 1,
710            schema_fingerprint: manifest.schema_fingerprint,
711            payload_digest: SubscriptionDigest::from_bytes([5; 32]),
712        };
713        manifest.segments = vec![segment; 4_096];
714
715        assert!(matches!(
716            manifest.computed_digest(),
717            Err(SubscriptionContractError::ManifestTooLarge { .. })
718        ));
719    }
720
721    #[test]
722    fn overlapping_segment_ranges_are_rejected() {
723        let mut manifest = manifest();
724        manifest.frontiers[0].through_sequence = PartitionSequence::new(3);
725        let segment = |first, end, marker| OutputSegmentRef {
726            protocol_version: SubscriptionProtocolVersion::CURRENT,
727            object_key: format!("subscription-output/test/{marker}.arrow"),
728            stream_generation: manifest.stream_generation,
729            partition: OutputPartitionId::new(0),
730            first_sequence: PartitionSequence::new(first),
731            exclusive_end_sequence: PartitionSequence::new(end),
732            frame_count: end - first,
733            row_count: 1,
734            encoded_length: 1,
735            schema_fingerprint: manifest.schema_fingerprint,
736            payload_digest: SubscriptionDigest::from_bytes([5; 32]),
737        };
738        manifest.segments = vec![segment(0, 2, "first"), segment(1, 3, "second")];
739        assert_eq!(
740            manifest.validate(),
741            Err(SubscriptionContractError::OverlappingSegments)
742        );
743    }
744}
laminar_core/checkpoint/subscription/manifest.rs

laminar_core/checkpoint/subscription/
manifest.rs
