laminar_core/cluster/control/
catalog_manifest.rs

1//! Catalog inventory sealed in the append-only leader authority.
2
3use std::fmt::Write;
4use std::sync::Arc;
5
6use ahash::AHashSet;
7use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
9
10use crate::checkpoint::LeaderProof;
11
12use super::{LeaderLeaseStore, LeaseError};
13
14pub use crate::catalog::CatalogObjectKind;
15
16pub(super) const CATALOG_MANIFEST_FORMAT_VERSION: u16 = 1;
17pub(super) const MAX_CATALOG_MANIFEST_ENTRIES: usize = 4_096;
18pub(super) const MAX_CATALOG_MANIFEST_BYTES: usize = 8 * 1024 * 1024;
19
20const CATALOG_MANIFEST_PREFIX: &str = "control/catalog-manifest/v1/";
21
22fn validate_entries(entries: &[CatalogManifestEntry]) -> Result<(), CatalogManifestError> {
23    if entries.len() > MAX_CATALOG_MANIFEST_ENTRIES {
24        return Err(CatalogManifestError::Invalid(format!(
25            "catalog manifest has {} entries; maximum is {MAX_CATALOG_MANIFEST_ENTRIES}",
26            entries.len()
27        )));
28    }
29    let mut names = AHashSet::with_capacity(entries.len());
30    for entry in entries {
31        if entry.canonical_name.is_empty() || entry.canonical_name.trim() != entry.canonical_name {
32            return Err(CatalogManifestError::Invalid(format!(
33                "catalog manifest has a non-canonical name {:?}",
34                entry.canonical_name
35            )));
36        }
37        if entry.ddl.trim().is_empty() {
38            return Err(CatalogManifestError::Invalid(format!(
39                "catalog manifest entry '{}' has empty DDL",
40                entry.canonical_name
41            )));
42        }
43        if !names.insert(entry.canonical_name.as_str()) {
44            return Err(CatalogManifestError::Invalid(format!(
45                "catalog manifest repeats canonical name '{}'",
46                entry.canonical_name
47            )));
48        }
49    }
50    Ok(())
51}
52
53/// One catalog object's defining DDL.
54#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
55#[serde(deny_unknown_fields)]
56pub struct CatalogManifestEntry {
57    /// Canonical catalog identifier.
58    pub canonical_name: String,
59    /// Exact namespace owner.
60    pub kind: CatalogObjectKind,
61    /// Exact DDL text replayed on every node.
62    pub ddl: String,
63}
64
65/// The complete ordered catalog sealed for one cluster control namespace.
66#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(deny_unknown_fields)]
68pub struct CatalogManifest {
69    /// DDL entries in dependency-safe creation order. An empty inventory is valid and sealed.
70    pub entries: Vec<CatalogManifestEntry>,
71}
72
73impl CatalogManifest {
74    /// Construct and validate a complete inventory.
75    ///
76    /// # Errors
77    /// Rejects empty/non-canonical names, empty DDL, duplicate identifiers, or an inventory that
78    /// exceeds the durable cardinality or encoded-size limits.
79    pub fn new(entries: Vec<CatalogManifestEntry>) -> Result<Self, CatalogManifestError> {
80        let manifest = Self { entries };
81        manifest.encode_and_reference()?;
82        Ok(manifest)
83    }
84
85    pub(super) fn validate(&self) -> Result<(), CatalogManifestError> {
86        validate_entries(&self.entries)?;
87        Ok(())
88    }
89
90    pub(super) fn encode_and_reference(
91        &self,
92    ) -> Result<(Vec<u8>, CatalogManifestRef), CatalogManifestError> {
93        self.validate()?;
94        let encoded = serde_json::to_vec(self)?;
95        if encoded.len() > MAX_CATALOG_MANIFEST_BYTES {
96            return Err(CatalogManifestError::Invalid(format!(
97                "encoded catalog manifest is {} bytes; maximum is {MAX_CATALOG_MANIFEST_BYTES}",
98                encoded.len()
99            )));
100        }
101        let encoded_len = u64::try_from(encoded.len()).map_err(|_| {
102            CatalogManifestError::Invalid("encoded catalog manifest length overflow".into())
103        })?;
104        let entry_count = u32::try_from(self.entries.len()).map_err(|_| {
105            CatalogManifestError::Invalid("catalog manifest entry count overflow".into())
106        })?;
107        let digest = Sha256::digest(&encoded);
108        let mut sha256 = String::with_capacity(64);
109        for byte in digest {
110            write!(&mut sha256, "{byte:02x}").expect("writing to a String cannot fail");
111        }
112        let reference = CatalogManifestRef {
113            version: CATALOG_MANIFEST_FORMAT_VERSION,
114            sha256,
115            encoded_len,
116            entry_count,
117        };
118        reference.validate()?;
119        Ok((encoded, reference))
120    }
121}
122
123/// Small immutable reference carried by every leader-lease renewal.
124#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
125#[serde(deny_unknown_fields)]
126pub struct CatalogManifestRef {
127    /// Canonical manifest encoding version.
128    pub version: u16,
129    /// Lowercase hexadecimal SHA-256 of the exact encoded manifest.
130    pub sha256: String,
131    /// Exact encoded object length.
132    pub encoded_len: u64,
133    /// Exact inventory cardinality.
134    pub entry_count: u32,
135}
136
137impl CatalogManifestRef {
138    pub(super) fn validate(&self) -> Result<(), CatalogManifestError> {
139        if self.version != CATALOG_MANIFEST_FORMAT_VERSION {
140            return Err(CatalogManifestError::Invalid(format!(
141                "unsupported catalog manifest version {}",
142                self.version
143            )));
144        }
145        if self.sha256.len() != 64
146            || !self
147                .sha256
148                .bytes()
149                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
150        {
151            return Err(CatalogManifestError::Invalid(
152                "catalog manifest SHA-256 must be 64 lowercase hexadecimal characters".into(),
153            ));
154        }
155        if self.encoded_len == 0
156            || self.encoded_len
157                > u64::try_from(MAX_CATALOG_MANIFEST_BYTES)
158                    .expect("catalog manifest byte limit fits u64")
159        {
160            return Err(CatalogManifestError::Invalid(format!(
161                "catalog manifest encoded length {} is outside 1..={MAX_CATALOG_MANIFEST_BYTES}",
162                self.encoded_len
163            )));
164        }
165        if !matches!(
166            usize::try_from(self.entry_count),
167            Ok(count) if count <= MAX_CATALOG_MANIFEST_ENTRIES
168        ) {
169            return Err(CatalogManifestError::Invalid(format!(
170                "catalog manifest entry count {} exceeds {MAX_CATALOG_MANIFEST_ENTRIES}",
171                self.entry_count
172            )));
173        }
174        Ok(())
175    }
176
177    pub(super) fn object_path(&self) -> object_store::path::Path {
178        object_store::path::Path::from(format!("{CATALOG_MANIFEST_PREFIX}{}.json", self.sha256))
179    }
180}
181
182/// Result of attempting to seal the immutable inventory.
183#[derive(Debug, Clone, Copy, PartialEq, Eq)]
184pub enum CatalogSealOutcome {
185    /// This caller appended the first lease record carrying the inventory.
186    Created,
187    /// The authority already carries the exact same canonical inventory.
188    ExistingIdentical,
189}
190
191/// Catalog view over the same append-only authority used for leader fencing.
192pub struct CatalogManifestStore {
193    authority: Arc<LeaderLeaseStore>,
194}
195
196impl std::fmt::Debug for CatalogManifestStore {
197    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
198        formatter
199            .debug_struct("CatalogManifestStore")
200            .finish_non_exhaustive()
201    }
202}
203
204/// Errors loading or sealing the catalog inventory.
205#[derive(Debug, thiserror::Error)]
206pub enum CatalogManifestError {
207    /// Shared leader authority failed.
208    #[error("leader lease authority: {0}")]
209    Authority(#[from] LeaseError),
210    /// JSON serialization or decoding failure.
211    #[error("JSON: {0}")]
212    Json(#[from] serde_json::Error),
213    /// The stored or proposed inventory is malformed.
214    #[error("invalid sealed catalog manifest: {0}")]
215    Invalid(String),
216    /// Another writer sealed a different complete inventory.
217    #[error("cluster catalog is already sealed with a different inventory")]
218    Conflict,
219    /// The supplied proof no longer owns the durable leader term.
220    #[error("catalog seal was fenced by a different durable leader term")]
221    Fenced,
222}
223
224impl CatalogManifestStore {
225    /// Share the exact append-only authority used by the leader lease manager.
226    #[must_use]
227    pub fn new(authority: Arc<LeaderLeaseStore>) -> Self {
228        Self { authority }
229    }
230
231    /// Load the sealed catalog, or `None` before the first successful seal.
232    ///
233    /// # Errors
234    /// Fails on object-store I/O, malformed JSON, or an invalid inventory.
235    pub async fn load(&self) -> Result<Option<CatalogManifest>, CatalogManifestError> {
236        let Some(reference) = self
237            .authority
238            .load()
239            .await?
240            .and_then(|lease| lease.catalog_manifest)
241        else {
242            return Ok(None);
243        };
244        self.authority
245            .load_catalog_manifest(&reference)
246            .await
247            .map(Some)
248    }
249
250    /// CAS-append the first inventory under an exact leader proof.
251    ///
252    /// A concurrent exact inventory is idempotent. Any different winner fails closed.
253    ///
254    /// # Errors
255    /// Fails for an invalid proposal, divergent winner, or object-store I/O.
256    pub async fn seal(
257        &self,
258        manifest: &CatalogManifest,
259        proof: &LeaderProof,
260    ) -> Result<CatalogSealOutcome, CatalogManifestError> {
261        Box::pin(self.authority.seal_catalog(proof, manifest)).await
262    }
263}
264
265#[cfg(test)]
266mod tests {
267    use super::*;
268    use bytes::Bytes;
269    use object_store::{memory::InMemory, ObjectStoreExt, PutPayload};
270    use uuid::Uuid;
271
272    async fn store() -> (CatalogManifestStore, Arc<InMemory>, LeaderProof) {
273        let backing = Arc::new(InMemory::new());
274        let authority = Arc::new(LeaderLeaseStore::new(backing.clone(), 1_000));
275        let owner = super::super::LeaderLeaseOwner {
276            node: crate::cluster::discovery::NodeId(1),
277            boot: Uuid::from_u128(1),
278            process_term: 1,
279        };
280        let super::super::LeaseOutcome::Acquired(lease) =
281            authority.begin_new_term(&owner, 0).await.unwrap()
282        else {
283            unreachable!()
284        };
285        (CatalogManifestStore::new(authority), backing, lease.proof())
286    }
287
288    fn entry(name: &str) -> CatalogManifestEntry {
289        CatalogManifestEntry {
290            canonical_name: name.to_string(),
291            kind: CatalogObjectKind::Source,
292            ddl: format!("CREATE SOURCE {name} (k BIGINT)"),
293        }
294    }
295
296    #[tokio::test]
297    async fn empty_inventory_is_a_real_sealed_manifest() {
298        let (store, _, proof) = store().await;
299        assert!(store.load().await.unwrap().is_none());
300        assert_eq!(
301            store
302                .seal(&CatalogManifest::default(), &proof)
303                .await
304                .unwrap(),
305            CatalogSealOutcome::Created
306        );
307        assert_eq!(
308            store.load().await.unwrap(),
309            Some(CatalogManifest::default())
310        );
311    }
312
313    #[tokio::test]
314    async fn concurrent_identical_seal_is_idempotent() {
315        let (store, _, proof) = store().await;
316        let store = Arc::new(store);
317        let manifest = CatalogManifest::new(vec![entry("events")]).unwrap();
318        let (left, right) =
319            tokio::join!(store.seal(&manifest, &proof), store.seal(&manifest, &proof));
320        let outcomes = [left.unwrap(), right.unwrap()];
321        assert!(outcomes.contains(&CatalogSealOutcome::Created));
322        assert!(outcomes.contains(&CatalogSealOutcome::ExistingIdentical));
323        assert_eq!(store.load().await.unwrap(), Some(manifest));
324    }
325
326    #[tokio::test]
327    async fn divergent_second_inventory_fails_closed() {
328        let (store, _, proof) = store().await;
329        let winner = CatalogManifest::new(vec![entry("winner")]).unwrap();
330        let loser = CatalogManifest::new(vec![entry("loser")]).unwrap();
331        store.seal(&winner, &proof).await.unwrap();
332        assert!(matches!(
333            store.seal(&loser, &proof).await,
334            Err(CatalogManifestError::Conflict)
335        ));
336        assert_eq!(store.load().await.unwrap(), Some(winner));
337    }
338
339    #[test]
340    fn duplicate_or_noncanonical_entries_are_rejected() {
341        let duplicate = vec![entry("events"), entry("events")];
342        assert!(matches!(
343            CatalogManifest::new(duplicate),
344            Err(CatalogManifestError::Invalid(_))
345        ));
346        assert!(matches!(
347            CatalogManifest::new(vec![entry(" events")]),
348            Err(CatalogManifestError::Invalid(_))
349        ));
350    }
351
352    #[test]
353    fn manifest_bounds_are_enforced_before_durable_writes() {
354        let too_many = (0..=MAX_CATALOG_MANIFEST_ENTRIES)
355            .map(|index| entry(&format!("source_{index}")))
356            .collect();
357        assert!(matches!(
358            CatalogManifest::new(too_many),
359            Err(CatalogManifestError::Invalid(_))
360        ));
361
362        let oversized = CatalogManifest::new(vec![CatalogManifestEntry {
363            canonical_name: "events".into(),
364            kind: CatalogObjectKind::Source,
365            ddl: "x".repeat(MAX_CATALOG_MANIFEST_BYTES),
366        }]);
367        assert!(matches!(oversized, Err(CatalogManifestError::Invalid(_))));
368    }
369
370    #[tokio::test]
371    async fn missing_sealed_manifest_blob_fails_closed() {
372        let (store, backing, proof) = store().await;
373        let manifest = CatalogManifest::new(vec![entry("events")]).unwrap();
374        let (_, reference) = manifest.encode_and_reference().unwrap();
375        store.seal(&manifest, &proof).await.unwrap();
376        backing.delete(&reference.object_path()).await.unwrap();
377
378        let error = store.load().await.unwrap_err();
379        assert!(matches!(&error, CatalogManifestError::Invalid(_)));
380        assert!(error.to_string().contains("is missing"));
381    }
382
383    #[tokio::test]
384    async fn tampered_sealed_manifest_blob_fails_closed() {
385        let (store, backing, proof) = store().await;
386        let manifest = CatalogManifest::new(vec![entry("events")]).unwrap();
387        let (mut encoded, reference) = manifest.encode_and_reference().unwrap();
388        store.seal(&manifest, &proof).await.unwrap();
389        let offset = encoded
390            .windows(b"events".len())
391            .position(|window| window == b"events")
392            .unwrap();
393        encoded[offset] = b'x';
394        backing
395            .put(
396                &reference.object_path(),
397                PutPayload::from(Bytes::from(encoded)),
398            )
399            .await
400            .unwrap();
401
402        let error = store.load().await.unwrap_err();
403        assert!(matches!(&error, CatalogManifestError::Invalid(_)));
404        assert!(error
405            .to_string()
406            .contains("does not match its sealed reference"));
407    }
408}
laminar_core/cluster/control/catalog_manifest.rs

laminar_core/cluster/control/
catalog_manifest.rs
