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 let Some(binding) = &entry.schema_binding {
50            binding
51                .canonical_bytes()
52                .map_err(|error| CatalogManifestError::Invalid(error.to_string()))?;
53            let matches_direction = match binding.direction {
54                crate::schema_binding::SchemaDirection::Source => matches!(
55                    entry.kind,
56                    CatalogObjectKind::Source
57                        | CatalogObjectKind::Table
58                        | CatalogObjectKind::LookupTable
59                ),
60                crate::schema_binding::SchemaDirection::Sink => {
61                    entry.kind == CatalogObjectKind::Sink
62                }
63            };
64            if !matches_direction {
65                return Err(CatalogManifestError::Invalid(
66                    "schema binding has the wrong catalog direction".into(),
67                ));
68            }
69        }
70        if !names.insert(entry.canonical_name.as_str()) {
71            return Err(CatalogManifestError::Invalid(format!(
72                "catalog manifest repeats canonical name '{}'",
73                entry.canonical_name
74            )));
75        }
76    }
77    Ok(())
78}
79
80/// One catalog object's defining DDL.
81#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
82#[serde(deny_unknown_fields)]
83pub struct CatalogManifestEntry {
84    /// Resolved connector contract. Absent only in legacy or non-connector entries.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub schema_binding: Option<crate::schema_binding::SchemaBinding>,
87    /// Canonical catalog identifier.
88    pub canonical_name: String,
89    /// Exact namespace owner.
90    pub kind: CatalogObjectKind,
91    /// Durable catalog-object incarnation. The initial sealed inventory uses generation one.
92    #[serde(
93        default = "initial_catalog_generation",
94        skip_serializing_if = "is_initial_catalog_generation"
95    )]
96    pub catalog_generation: u64,
97    /// Exact DDL text replayed on every node.
98    pub ddl: String,
99}
100
101const fn initial_catalog_generation() -> u64 {
102    1
103}
104
105const fn is_initial_catalog_generation(generation: &u64) -> bool {
106    *generation == initial_catalog_generation()
107}
108
109/// The complete ordered catalog sealed for one cluster control namespace.
110#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
111#[serde(deny_unknown_fields)]
112pub struct CatalogManifest {
113    /// DDL entries in dependency-safe creation order. An empty inventory is valid and sealed.
114    pub entries: Vec<CatalogManifestEntry>,
115}
116
117impl CatalogManifest {
118    /// Compute the canonical content reference without publishing or changing authority.
119    ///
120    /// # Errors
121    /// Rejects a malformed or oversized inventory.
122    pub fn reference(&self) -> Result<CatalogManifestRef, CatalogManifestError> {
123        self.encode_and_reference().map(|(_, reference)| reference)
124    }
125
126    /// Construct and validate a complete inventory.
127    ///
128    /// # Errors
129    /// Rejects empty/non-canonical names, empty DDL, duplicate identifiers, or an inventory that
130    /// exceeds the durable cardinality or encoded-size limits.
131    pub fn new(entries: Vec<CatalogManifestEntry>) -> Result<Self, CatalogManifestError> {
132        let manifest = Self { entries };
133        manifest.encode_and_reference()?;
134        Ok(manifest)
135    }
136
137    pub(super) fn validate(&self) -> Result<(), CatalogManifestError> {
138        validate_entries(&self.entries)?;
139        Ok(())
140    }
141
142    pub(super) fn encode_and_reference(
143        &self,
144    ) -> Result<(Vec<u8>, CatalogManifestRef), CatalogManifestError> {
145        self.validate()?;
146        // COMPAT: preserve byte-for-byte legacy manifest references. New bindings contain
147        // Arrow metadata maps, which must be canonical across nodes and process restarts.
148        let encoded = if self
149            .entries
150            .iter()
151            .any(|entry| entry.schema_binding.is_some())
152        {
153            let mut value = serde_json::to_value(self)?;
154            value.sort_all_objects();
155            serde_json::to_vec(&value)?
156        } else {
157            serde_json::to_vec(self)?
158        };
159        if encoded.len() > MAX_CATALOG_MANIFEST_BYTES {
160            return Err(CatalogManifestError::Invalid(format!(
161                "encoded catalog manifest is {} bytes; maximum is {MAX_CATALOG_MANIFEST_BYTES}",
162                encoded.len()
163            )));
164        }
165        let encoded_len = u64::try_from(encoded.len()).map_err(|_| {
166            CatalogManifestError::Invalid("encoded catalog manifest length overflow".into())
167        })?;
168        let entry_count = u32::try_from(self.entries.len()).map_err(|_| {
169            CatalogManifestError::Invalid("catalog manifest entry count overflow".into())
170        })?;
171        let digest = Sha256::digest(&encoded);
172        let mut sha256 = String::with_capacity(64);
173        for byte in digest {
174            write!(&mut sha256, "{byte:02x}").expect("writing to a String cannot fail");
175        }
176        let reference = CatalogManifestRef {
177            version: CATALOG_MANIFEST_FORMAT_VERSION,
178            sha256,
179            encoded_len,
180            entry_count,
181        };
182        reference.validate()?;
183        Ok((encoded, reference))
184    }
185}
186
187/// Small immutable reference carried by every leader-lease renewal.
188#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
189#[serde(deny_unknown_fields)]
190pub struct CatalogManifestRef {
191    /// Canonical manifest encoding version.
192    pub version: u16,
193    /// Lowercase hexadecimal SHA-256 of the exact encoded manifest.
194    pub sha256: String,
195    /// Exact encoded object length.
196    pub encoded_len: u64,
197    /// Exact inventory cardinality.
198    pub entry_count: u32,
199}
200
201impl CatalogManifestRef {
202    pub(crate) fn validate(&self) -> Result<(), CatalogManifestError> {
203        if self.version != CATALOG_MANIFEST_FORMAT_VERSION {
204            return Err(CatalogManifestError::Invalid(format!(
205                "unsupported catalog manifest version {}",
206                self.version
207            )));
208        }
209        if self.sha256.len() != 64
210            || !self
211                .sha256
212                .bytes()
213                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
214        {
215            return Err(CatalogManifestError::Invalid(
216                "catalog manifest SHA-256 must be 64 lowercase hexadecimal characters".into(),
217            ));
218        }
219        if self.encoded_len == 0
220            || self.encoded_len
221                > u64::try_from(MAX_CATALOG_MANIFEST_BYTES)
222                    .expect("catalog manifest byte limit fits u64")
223        {
224            return Err(CatalogManifestError::Invalid(format!(
225                "catalog manifest encoded length {} is outside 1..={MAX_CATALOG_MANIFEST_BYTES}",
226                self.encoded_len
227            )));
228        }
229        if !matches!(
230            usize::try_from(self.entry_count),
231            Ok(count) if count <= MAX_CATALOG_MANIFEST_ENTRIES
232        ) {
233            return Err(CatalogManifestError::Invalid(format!(
234                "catalog manifest entry count {} exceeds {MAX_CATALOG_MANIFEST_ENTRIES}",
235                self.entry_count
236            )));
237        }
238        Ok(())
239    }
240
241    pub(super) fn object_path(&self) -> object_store::path::Path {
242        object_store::path::Path::from(format!("{CATALOG_MANIFEST_PREFIX}{}.json", self.sha256))
243    }
244}
245
246/// Result of attempting to seal the immutable inventory.
247#[derive(Debug, Clone, Copy, PartialEq, Eq)]
248pub enum CatalogSealOutcome {
249    /// This caller appended the first lease record carrying the inventory.
250    Created,
251    /// The authority already carries the exact same canonical inventory.
252    ExistingIdentical,
253}
254
255/// Catalog view over the same append-only authority used for leader fencing.
256pub struct CatalogManifestStore {
257    authority: Arc<LeaderLeaseStore>,
258}
259
260impl std::fmt::Debug for CatalogManifestStore {
261    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
262        formatter
263            .debug_struct("CatalogManifestStore")
264            .finish_non_exhaustive()
265    }
266}
267
268/// Errors loading or sealing the catalog inventory.
269#[derive(Debug, thiserror::Error)]
270pub enum CatalogManifestError {
271    /// Shared leader authority failed.
272    #[error("leader lease authority: {0}")]
273    Authority(#[from] LeaseError),
274    /// JSON serialization or decoding failure.
275    #[error("JSON: {0}")]
276    Json(#[from] serde_json::Error),
277    /// The stored or proposed inventory is malformed.
278    #[error("invalid sealed catalog manifest: {0}")]
279    Invalid(String),
280    /// Another writer sealed a different complete inventory.
281    #[error("cluster catalog is already sealed with a different inventory")]
282    Conflict,
283    /// The supplied proof no longer owns the durable leader term.
284    #[error("catalog seal was fenced by a different durable leader term")]
285    Fenced,
286}
287
288impl CatalogManifestStore {
289    /// Read the definitive payload-bound status of a reserved topology request.
290    ///
291    /// # Errors
292    /// Fails on unavailable/corrupt evidence or the bounded read deadline.
293    pub async fn operation_status(
294        &self,
295        operation_id: super::topology::TopologyOperationId,
296    ) -> Result<Option<super::topology::TopologyAdmissionStatus>, super::topology::TopologyError>
297    {
298        self.authority.topology_operation_status(operation_id).await
299    }
300
301    /// Share the exact append-only authority used by the leader lease manager.
302    #[must_use]
303    pub fn new(authority: Arc<LeaderLeaseStore>) -> Self {
304        Self { authority }
305    }
306
307    /// Read explicit logical topology metadata from the same catalog/leader authority.
308    ///
309    /// # Errors
310    /// Fails closed when the authority or referenced catalog/deployment is invalid.
311    pub async fn topology_state(
312        &self,
313    ) -> Result<super::topology::TopologyCatalogState, super::topology::TopologyError> {
314        self.authority.topology_catalog_state().await
315    }
316
317    /// Commit of a topology migration that still owns startup and recovery; `None` once settled.
318    ///
319    /// # Errors
320    /// Fails closed when the authority or referenced topology evidence is invalid.
321    pub async fn live_topology_commit(
322        &self,
323    ) -> Result<Option<super::topology::TopologyCommit>, super::topology::TopologyError> {
324        self.authority.live_topology_commit().await
325    }
326
327    /// Retained topology request identities against the fixed journal capacity.
328    ///
329    /// # Errors
330    /// Fails closed when the authority is invalid or the read exceeds its deadline.
331    pub async fn topology_journal_usage(
332        &self,
333    ) -> Result<super::topology::TopologyJournalUsage, super::topology::TopologyError> {
334        self.authority.topology_journal_usage().await
335    }
336
337    /// Subscription certificates preserved by the newest committed migration.
338    ///
339    /// # Errors
340    /// Fails closed when the authority or referenced migration root is invalid.
341    pub async fn committed_topology_subscriptions(
342        &self,
343    ) -> Result<Vec<crate::checkpoint::OutputDistributionCertificate>, super::topology::TopologyError>
344    {
345        self.authority.committed_topology_subscriptions().await
346    }
347
348    /// Load the catalog and its topology metadata from one immutable authority snapshot.
349    ///
350    /// # Errors
351    /// Fails closed on missing/corrupt content or inconsistent adoption/deployment authority.
352    pub async fn load_with_topology(
353        &self,
354    ) -> Result<
355        Option<(CatalogManifest, super::topology::TopologyCatalogState)>,
356        super::topology::TopologyError,
357    > {
358        self.authority.catalog_with_topology().await
359    }
360
361    /// Load the exact parent catalog while checking the current committed inventory.
362    ///
363    /// # Errors
364    /// Rejects changed authority, invalid retained evidence or the bounded read deadline.
365    pub async fn parent_topology_catalog(
366        &self,
367        expected_current: &CatalogManifestRef,
368    ) -> Result<Option<CatalogManifest>, super::topology::TopologyError> {
369        self.authority
370            .parent_topology_catalog(expected_current)
371            .await
372    }
373
374    /// Load the complete adopted startup catalog while checking the exact current inventory.
375    /// Removed objects remain startup assertions and are never recreated.
376    ///
377    /// # Errors
378    /// Rejects changed authority, invalid retained evidence or the bounded read deadline.
379    pub async fn original_topology_catalog(
380        &self,
381        expected_current: &CatalogManifestRef,
382    ) -> Result<Option<CatalogManifest>, super::topology::TopologyError> {
383        self.authority
384            .original_topology_catalog(expected_current)
385            .await
386    }
387
388    /// Read object names retired by the bounded, retained committed topology journal.
389    ///
390    /// # Errors
391    /// Rejects invalid retained evidence, changed committed authority or the read deadline.
392    pub async fn retired_topology_names(
393        &self,
394    ) -> Result<std::collections::BTreeSet<String>, super::topology::TopologyError> {
395        self.authority.retired_topology_names().await
396    }
397
398    /// Read the latest retired incarnation of each name for safe future-only recreation.
399    ///
400    /// # Errors
401    /// Rejects invalid retained evidence, changed committed authority or the read deadline.
402    pub async fn retired_topology_generations(
403        &self,
404    ) -> Result<std::collections::BTreeMap<String, u64>, super::topology::TopologyError> {
405        self.authority.retired_topology_generations().await
406    }
407
408    /// Explicitly adopt a sealed legacy inventory without changing the processing graph.
409    ///
410    /// Requires a coordinated binary upgrade; this is not runtime migration admission.
411    ///
412    /// # Errors
413    /// Rejects stale leader proof, divergent reference/deployment, invalid data or authority I/O.
414    pub async fn adopt_legacy_topology(
415        &self,
416        proof: &LeaderProof,
417        operation_id: super::topology::TopologyOperationId,
418        expected_manifest: &CatalogManifestRef,
419        expected_deployment: &str,
420    ) -> Result<super::topology::TopologyAdoptionOutcome, super::topology::TopologyError> {
421        self.authority
422            .adopt_legacy_topology(proof, operation_id, expected_manifest, expected_deployment)
423            .await
424    }
425
426    /// Load the sealed catalog, or `None` before the first successful seal.
427    ///
428    /// # Errors
429    /// Fails on object-store I/O, malformed JSON, or an invalid inventory.
430    pub async fn load(&self) -> Result<Option<CatalogManifest>, CatalogManifestError> {
431        self.load_with_topology()
432            .await
433            .map(|snapshot| snapshot.map(|(manifest, _)| manifest))
434            .map_err(|error| match error {
435                super::topology::TopologyError::Authority(error) => {
436                    CatalogManifestError::Authority(error)
437                }
438                super::topology::TopologyError::Catalog(error) => error,
439                error => CatalogManifestError::Invalid(error.to_string()),
440            })
441    }
442
443    /// CAS-append the first inventory under an exact leader proof.
444    ///
445    /// A concurrent exact inventory is idempotent. Any different winner fails closed.
446    ///
447    /// # Errors
448    /// Fails for an invalid proposal, divergent winner, or object-store I/O.
449    pub async fn seal(
450        &self,
451        manifest: &CatalogManifest,
452        proof: &LeaderProof,
453    ) -> Result<CatalogSealOutcome, CatalogManifestError> {
454        Box::pin(self.authority.seal_catalog(proof, manifest)).await
455    }
456}
457
458#[cfg(test)]
459mod tests;
laminar_core/cluster/control/catalog_manifest/mod.rs

laminar_core/cluster/control/catalog_manifest/
mod.rs
