laminar_core/cluster/control/catalog_manifest/
mod.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 entry.catalog_generation == 0 {
44            return Err(CatalogManifestError::Invalid(format!(
45                "catalog manifest entry '{}' has a zero object generation",
46                entry.canonical_name
47            )));
48        }
49        if !names.insert(entry.canonical_name.as_str()) {
50            return Err(CatalogManifestError::Invalid(format!(
51                "catalog manifest repeats canonical name '{}'",
52                entry.canonical_name
53            )));
54        }
55    }
56    Ok(())
57}
58
59/// One catalog object's defining DDL.
60#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
61#[serde(deny_unknown_fields)]
62pub struct CatalogManifestEntry {
63    /// Canonical catalog identifier.
64    pub canonical_name: String,
65    /// Exact namespace owner.
66    pub kind: CatalogObjectKind,
67    /// Durable catalog-object incarnation. The initial sealed inventory uses generation one.
68    #[serde(
69        default = "initial_catalog_generation",
70        skip_serializing_if = "is_initial_catalog_generation"
71    )]
72    pub catalog_generation: u64,
73    /// Exact DDL text replayed on every node.
74    pub ddl: String,
75}
76
77const fn initial_catalog_generation() -> u64 {
78    1
79}
80
81const fn is_initial_catalog_generation(generation: &u64) -> bool {
82    *generation == initial_catalog_generation()
83}
84
85/// The complete ordered catalog sealed for one cluster control namespace.
86#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
87#[serde(deny_unknown_fields)]
88pub struct CatalogManifest {
89    /// DDL entries in dependency-safe creation order. An empty inventory is valid and sealed.
90    pub entries: Vec<CatalogManifestEntry>,
91}
92
93impl CatalogManifest {
94    /// Construct and validate a complete inventory.
95    ///
96    /// # Errors
97    /// Rejects empty/non-canonical names, empty DDL, duplicate identifiers, or an inventory that
98    /// exceeds the durable cardinality or encoded-size limits.
99    pub fn new(entries: Vec<CatalogManifestEntry>) -> Result<Self, CatalogManifestError> {
100        let manifest = Self { entries };
101        manifest.encode_and_reference()?;
102        Ok(manifest)
103    }
104
105    pub(super) fn validate(&self) -> Result<(), CatalogManifestError> {
106        validate_entries(&self.entries)?;
107        Ok(())
108    }
109
110    pub(super) fn encode_and_reference(
111        &self,
112    ) -> Result<(Vec<u8>, CatalogManifestRef), CatalogManifestError> {
113        self.validate()?;
114        let encoded = serde_json::to_vec(self)?;
115        if encoded.len() > MAX_CATALOG_MANIFEST_BYTES {
116            return Err(CatalogManifestError::Invalid(format!(
117                "encoded catalog manifest is {} bytes; maximum is {MAX_CATALOG_MANIFEST_BYTES}",
118                encoded.len()
119            )));
120        }
121        let encoded_len = u64::try_from(encoded.len()).map_err(|_| {
122            CatalogManifestError::Invalid("encoded catalog manifest length overflow".into())
123        })?;
124        let entry_count = u32::try_from(self.entries.len()).map_err(|_| {
125            CatalogManifestError::Invalid("catalog manifest entry count overflow".into())
126        })?;
127        let digest = Sha256::digest(&encoded);
128        let mut sha256 = String::with_capacity(64);
129        for byte in digest {
130            write!(&mut sha256, "{byte:02x}").expect("writing to a String cannot fail");
131        }
132        let reference = CatalogManifestRef {
133            version: CATALOG_MANIFEST_FORMAT_VERSION,
134            sha256,
135            encoded_len,
136            entry_count,
137        };
138        reference.validate()?;
139        Ok((encoded, reference))
140    }
141}
142
143/// Small immutable reference carried by every leader-lease renewal.
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
145#[serde(deny_unknown_fields)]
146pub struct CatalogManifestRef {
147    /// Canonical manifest encoding version.
148    pub version: u16,
149    /// Lowercase hexadecimal SHA-256 of the exact encoded manifest.
150    pub sha256: String,
151    /// Exact encoded object length.
152    pub encoded_len: u64,
153    /// Exact inventory cardinality.
154    pub entry_count: u32,
155}
156
157impl CatalogManifestRef {
158    pub(super) fn validate(&self) -> Result<(), CatalogManifestError> {
159        if self.version != CATALOG_MANIFEST_FORMAT_VERSION {
160            return Err(CatalogManifestError::Invalid(format!(
161                "unsupported catalog manifest version {}",
162                self.version
163            )));
164        }
165        if self.sha256.len() != 64
166            || !self
167                .sha256
168                .bytes()
169                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
170        {
171            return Err(CatalogManifestError::Invalid(
172                "catalog manifest SHA-256 must be 64 lowercase hexadecimal characters".into(),
173            ));
174        }
175        if self.encoded_len == 0
176            || self.encoded_len
177                > u64::try_from(MAX_CATALOG_MANIFEST_BYTES)
178                    .expect("catalog manifest byte limit fits u64")
179        {
180            return Err(CatalogManifestError::Invalid(format!(
181                "catalog manifest encoded length {} is outside 1..={MAX_CATALOG_MANIFEST_BYTES}",
182                self.encoded_len
183            )));
184        }
185        if !matches!(
186            usize::try_from(self.entry_count),
187            Ok(count) if count <= MAX_CATALOG_MANIFEST_ENTRIES
188        ) {
189            return Err(CatalogManifestError::Invalid(format!(
190                "catalog manifest entry count {} exceeds {MAX_CATALOG_MANIFEST_ENTRIES}",
191                self.entry_count
192            )));
193        }
194        Ok(())
195    }
196
197    pub(super) fn object_path(&self) -> object_store::path::Path {
198        object_store::path::Path::from(format!("{CATALOG_MANIFEST_PREFIX}{}.json", self.sha256))
199    }
200}
201
202/// Result of attempting to seal the immutable inventory.
203#[derive(Debug, Clone, Copy, PartialEq, Eq)]
204pub enum CatalogSealOutcome {
205    /// This caller appended the first lease record carrying the inventory.
206    Created,
207    /// The authority already carries the exact same canonical inventory.
208    ExistingIdentical,
209}
210
211/// Catalog view over the same append-only authority used for leader fencing.
212pub struct CatalogManifestStore {
213    authority: Arc<LeaderLeaseStore>,
214}
215
216impl std::fmt::Debug for CatalogManifestStore {
217    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
218        formatter
219            .debug_struct("CatalogManifestStore")
220            .finish_non_exhaustive()
221    }
222}
223
224/// Errors loading or sealing the catalog inventory.
225#[derive(Debug, thiserror::Error)]
226pub enum CatalogManifestError {
227    /// Shared leader authority failed.
228    #[error("leader lease authority: {0}")]
229    Authority(#[from] LeaseError),
230    /// JSON serialization or decoding failure.
231    #[error("JSON: {0}")]
232    Json(#[from] serde_json::Error),
233    /// The stored or proposed inventory is malformed.
234    #[error("invalid sealed catalog manifest: {0}")]
235    Invalid(String),
236    /// Another writer sealed a different complete inventory.
237    #[error("cluster catalog is already sealed with a different inventory")]
238    Conflict,
239    /// The supplied proof no longer owns the durable leader term.
240    #[error("catalog seal was fenced by a different durable leader term")]
241    Fenced,
242}
243
244impl CatalogManifestStore {
245    /// Share the exact append-only authority used by the leader lease manager.
246    #[must_use]
247    pub fn new(authority: Arc<LeaderLeaseStore>) -> Self {
248        Self { authority }
249    }
250
251    /// Load the sealed catalog, or `None` before the first successful seal.
252    ///
253    /// # Errors
254    /// Fails on object-store I/O, malformed JSON, or an invalid inventory.
255    pub async fn load(&self) -> Result<Option<CatalogManifest>, CatalogManifestError> {
256        let Some(reference) = self
257            .authority
258            .load()
259            .await?
260            .and_then(|lease| lease.catalog_manifest)
261        else {
262            return Ok(None);
263        };
264        self.authority
265            .load_catalog_manifest(&reference)
266            .await
267            .map(Some)
268    }
269
270    /// CAS-append the first inventory under an exact leader proof.
271    ///
272    /// A concurrent exact inventory is idempotent. Any different winner fails closed.
273    ///
274    /// # Errors
275    /// Fails for an invalid proposal, divergent winner, or object-store I/O.
276    pub async fn seal(
277        &self,
278        manifest: &CatalogManifest,
279        proof: &LeaderProof,
280    ) -> Result<CatalogSealOutcome, CatalogManifestError> {
281        Box::pin(self.authority.seal_catalog(proof, manifest)).await
282    }
283}
284
285#[cfg(test)]
286mod tests;
laminar_core/cluster/control/catalog_manifest/mod.rs

laminar_core/cluster/control/catalog_manifest/
mod.rs
