laminar_core/cluster/control/topology/
source_initialization.rs

1//! Connector-owned, globally sealed initial cursors. These are requirements, not intake permits.
2
3use super::TopologyError;
4use crate::checkpoint::ConnectorCheckpoint;
5use serde::{Deserialize, Serialize};
6
7/// Maximum physical input channels in one new source's initial inventory.
8pub const MAX_TOPOLOGY_SOURCE_CHANNELS: usize = 4096;
9
10/// One new catalog incarnation's exact position before any participant consumes input.
11/// The existing connector cursor encoding preserves never-read partitions without inventing a
12/// consumed offset, checkpoint attempt or assignment ownership. Target installation must validate
13/// this cursor through the connector and adopt only its assigned channels before releasing intake.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(deny_unknown_fields)]
16pub struct TopologySourceInitialization {
17    /// Canonical source identity from the independently certified candidate.
18    pub name: String,
19    /// New catalog incarnation, distinct from the source's connector cursor format.
20    pub catalog_generation: u64,
21    /// Exact definition/schema/connector/dependency contract certified by every participant.
22    pub compatibility_sha256: String,
23    /// Complete global input inventory and numeric initial cursors supplied by the connector.
24    pub checkpoint: ConnectorCheckpoint,
25}
26
27impl TopologySourceInitialization {
28    pub(crate) fn validate(&self) -> Result<(), TopologyError> {
29        let checkpoint = &self.checkpoint;
30        let channels = checkpoint.input_channels.as_deref().ok_or_else(|| {
31            TopologyError::Invalid("source initialization has no complete input inventory".into())
32        })?;
33        if self.name.is_empty()
34            || self.catalog_generation == 0
35            || self.compatibility_sha256.len() != 64
36            || !self
37                .compatibility_sha256
38                .bytes()
39                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
40            || checkpoint.source_assignment_version.is_some()
41            || channels.is_empty()
42            || channels.len() > MAX_TOPOLOGY_SOURCE_CHANNELS
43            || !channels.windows(2).all(|p| p[0] < p[1])
44            || channels.iter().any(|c| {
45                c.is_empty()
46                    || c.len() > 1024
47                    || c == crate::checkpoint::SINGLETON_WATERMARK_CHANNEL
48            })
49            || checkpoint.offsets.is_empty()
50            || checkpoint.offsets.len() > MAX_TOPOLOGY_SOURCE_CHANNELS
51            || checkpoint.metadata.len() > 64
52            || checkpoint
53                .metadata
54                .get("connector")
55                .is_none_or(String::is_empty)
56            || checkpoint
57                .offsets
58                .iter()
59                .chain(&checkpoint.metadata)
60                .any(|(key, value)| key.is_empty() || key.len() > 1024 || value.len() > 4096)
61        {
62            return Err(TopologyError::Invalid(
63                "source initialization requires a bounded, unowned, complete connector cursor"
64                    .into(),
65            ));
66        }
67        Ok(())
68    }
69}
laminar_core/cluster/control/topology/source_initialization.rs

laminar_core/cluster/control/topology/
source_initialization.rs
