laminar_core/checkpoint/subscription/
distribution.rs

1use super::{
2    OutputPartitionId, StreamGeneration, SubscriptionContractError, SubscriptionDigest,
3    SubscriptionProtocolVersion,
4};
5use crate::checkpoint::PipelineIdentity;
6use crate::state::{KeyGroupCount, PARTITIONING_ABI_VERSION};
7
8/// Current canonical output-distribution certificate format.
9pub const OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION: u16 = 1;
10
11const MAX_STREAM_ID_BYTES: usize = 256;
12const MAX_OPERATOR_ID_BYTES: usize = 512;
13
14/// Recoverable mapping from final operator output to subscription partitions.
15#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
16#[serde(rename_all = "snake_case", tag = "kind", deny_unknown_fields)]
17pub enum OutputDistribution {
18    /// Group keys use the stable partition ABI and produce output on the same vnode.
19    VnodePartitioned {
20        /// Digest of canonical grouping-key expressions and types.
21        key_expressions_fingerprint: SubscriptionDigest,
22        /// Exact key encoding and vnode mapping ABI.
23        partition_abi: u16,
24        /// Complete logical vnode domain.
25        vnode_count: u16,
26    },
27    /// Existing global-aggregate execution on the singleton vnode.
28    Singleton {
29        /// Singleton/global vnode.
30        partition: OutputPartitionId,
31    },
32}
33
34impl OutputDistribution {
35    /// Number of certified output partitions.
36    #[must_use]
37    pub const fn partition_count(&self) -> u16 {
38        match self {
39            Self::VnodePartitioned { vnode_count, .. } => *vnode_count,
40            Self::Singleton { .. } => 1,
41        }
42    }
43
44    pub(crate) fn validate(
45        &self,
46        key_group_count: KeyGroupCount,
47    ) -> Result<(), SubscriptionContractError> {
48        match self {
49            Self::VnodePartitioned {
50                key_expressions_fingerprint,
51                partition_abi,
52                vnode_count,
53            } => {
54                key_expressions_fingerprint.validate("key_expressions_fingerprint")?;
55                if *partition_abi != PARTITIONING_ABI_VERSION {
56                    return Err(SubscriptionContractError::PartitionAbi {
57                        actual: *partition_abi,
58                        expected: PARTITIONING_ABI_VERSION,
59                    });
60                }
61                if *vnode_count != key_group_count.get() {
62                    return Err(SubscriptionContractError::VnodeCount {
63                        actual: *vnode_count,
64                        expected: key_group_count.get(),
65                    });
66                }
67            }
68            Self::Singleton { partition } => {
69                partition.validate(key_group_count)?;
70                if partition.get() != 0 {
71                    return Err(SubscriptionContractError::NonCanonicalSingleton {
72                        partition: partition.get(),
73                    });
74                }
75            }
76        }
77        Ok(())
78    }
79
80    /// Whether `partition` belongs to this certified output domain.
81    #[must_use]
82    pub fn contains(&self, partition: OutputPartitionId) -> bool {
83        match self {
84            Self::VnodePartitioned { vnode_count, .. } => partition.get() < *vnode_count,
85            Self::Singleton {
86                partition: singleton,
87            } => partition == *singleton,
88        }
89    }
90}
91
92/// Changelog representation bound into an output-distribution certificate.
93#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
94#[serde(rename_all = "snake_case")]
95pub enum ChangelogMode {
96    /// Rows are append-only.
97    Append,
98    /// Each frame replaces the visible contents of its output partition.
99    FullPartitionSnapshot,
100    /// Updates use the existing `__weight` retract-before-insert convention.
101    WeightedRetractInsert,
102}
103
104/// Planner proof that a final operator has a stable, recoverable output distribution.
105#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
106#[serde(deny_unknown_fields)]
107pub struct OutputDistributionCertificate {
108    /// Exact certificate envelope format.
109    pub version: u16,
110    /// Subscription protocol used by the output.
111    pub protocol_version: SubscriptionProtocolVersion,
112    /// Canonical catalog stream name.
113    pub stream_id: String,
114    /// Monotonic durable catalog incarnation.
115    pub catalog_generation: u64,
116    /// Derived durable incarnation identity.
117    pub stream_generation: StreamGeneration,
118    /// Stable final graph operator identity.
119    pub final_operator_id: String,
120    /// Certified output-to-partition mapping.
121    pub distribution: OutputDistribution,
122    /// Fingerprint of the unchanged user output schema.
123    pub schema_fingerprint: SubscriptionDigest,
124    /// Changelog representation emitted by the final operator.
125    pub changelog_mode: ChangelogMode,
126    /// Maximum committed output bytes retained for epoch replay; zero admits tail only.
127    pub history_retention_bytes: u64,
128    /// Canonical query fingerprint.
129    pub query_fingerprint: SubscriptionDigest,
130    /// Complete pipeline/state ABI identity.
131    pub pipeline_identity: PipelineIdentity,
132}
133
134impl OutputDistributionCertificate {
135    /// Validate the canonical certificate and all external bindings.
136    ///
137    /// # Errors
138    /// Returns the first non-canonical or mismatched durable binding.
139    pub fn validate(
140        &self,
141        key_group_count: KeyGroupCount,
142    ) -> Result<(), SubscriptionContractError> {
143        if self.version != OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION {
144            return Err(SubscriptionContractError::DistributionCertificateVersion {
145                actual: self.version,
146                expected: OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION,
147            });
148        }
149        self.protocol_version.validate()?;
150        validate_identity("stream_id", &self.stream_id, MAX_STREAM_ID_BYTES)?;
151        validate_identity(
152            "final_operator_id",
153            &self.final_operator_id,
154            MAX_OPERATOR_ID_BYTES,
155        )?;
156        if self.catalog_generation == 0 {
157            return Err(SubscriptionContractError::ZeroCatalogGeneration);
158        }
159        self.stream_generation.validate()?;
160        self.schema_fingerprint.validate("schema_fingerprint")?;
161        self.query_fingerprint.validate("query_fingerprint")?;
162        if !self.pipeline_identity.is_canonical() {
163            return Err(SubscriptionContractError::PipelineIdentity);
164        }
165        self.distribution.validate(key_group_count)
166    }
167
168    /// Validate exact equality with the certificate selected by the current planner.
169    ///
170    /// # Errors
171    /// Returns a certificate mismatch when any durable binding differs.
172    pub fn require_match(&self, expected: &Self) -> Result<(), SubscriptionContractError> {
173        if self != expected {
174            return Err(SubscriptionContractError::DistributionCertificateMismatch);
175        }
176        Ok(())
177    }
178
179    /// Whether this certificate represents the initial externally admitted stream class.
180    #[must_use]
181    pub fn certifies_keyed_aggregate(&self) -> bool {
182        matches!(
183            self.distribution,
184            OutputDistribution::VnodePartitioned { .. }
185        ) && self.changelog_mode == ChangelogMode::WeightedRetractInsert
186    }
187}
188
189fn validate_identity(
190    field: &'static str,
191    value: &str,
192    max_bytes: usize,
193) -> Result<(), SubscriptionContractError> {
194    if value.is_empty() || value.trim() != value || value.len() > max_bytes {
195        return Err(SubscriptionContractError::NonCanonicalIdentity { field, max_bytes });
196    }
197    Ok(())
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203
204    fn certificate() -> OutputDistributionCertificate {
205        OutputDistributionCertificate {
206            version: OUTPUT_DISTRIBUTION_CERTIFICATE_VERSION,
207            protocol_version: SubscriptionProtocolVersion::CURRENT,
208            stream_id: "positions".into(),
209            catalog_generation: 1,
210            stream_generation: StreamGeneration::from_digest(SubscriptionDigest::from_bytes(
211                [1; 32],
212            )),
213            final_operator_id: "stream:positions".into(),
214            distribution: OutputDistribution::VnodePartitioned {
215                key_expressions_fingerprint: SubscriptionDigest::from_bytes([2; 32]),
216                partition_abi: PARTITIONING_ABI_VERSION,
217                vnode_count: 4,
218            },
219            schema_fingerprint: SubscriptionDigest::from_bytes([3; 32]),
220            changelog_mode: ChangelogMode::WeightedRetractInsert,
221            history_retention_bytes: 0,
222            query_fingerprint: SubscriptionDigest::from_bytes([4; 32]),
223            pipeline_identity: PipelineIdentity::empty(),
224        }
225    }
226
227    #[test]
228    fn certificate_mismatch_is_structured() {
229        let actual = certificate();
230        let mut expected = actual.clone();
231        expected.schema_fingerprint = SubscriptionDigest::from_bytes([9; 32]);
232        assert_eq!(
233            actual.require_match(&expected),
234            Err(SubscriptionContractError::DistributionCertificateMismatch)
235        );
236    }
237
238    #[test]
239    fn protocol_and_schema_fingerprint_fail_closed() {
240        let groups = KeyGroupCount::try_from(4_u16).unwrap();
241        let mut wrong_protocol = certificate();
242        wrong_protocol.protocol_version = SubscriptionProtocolVersion::new(2);
243        assert!(matches!(
244            wrong_protocol.validate(groups),
245            Err(SubscriptionContractError::ProtocolVersion { .. })
246        ));
247
248        let mut wrong_schema = certificate();
249        wrong_schema.schema_fingerprint = SubscriptionDigest::from_bytes([0; 32]);
250        assert_eq!(
251            wrong_schema.validate(groups),
252            Err(SubscriptionContractError::ZeroDigest {
253                field: "schema_fingerprint"
254            })
255        );
256    }
257}
laminar_core/checkpoint/subscription/distribution.rs

laminar_core/checkpoint/subscription/
distribution.rs
