laminar_core/shuffle/
topology.rs

1//! Logical graph identity at the existing ordered-stream boundary.
2
3use std::io;
4use std::num::NonZeroU64;
5
6/// Exact logical topology and catalog digest, independent of assignment and recovery counters.
7/// Installing this transport fence is a trusted control-path action, not an output or Release permit.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub struct ShuffleTopologyFence {
10    version: NonZeroU64,
11    manifest_sha256: [u8; 32],
12}
13
14impl ShuffleTopologyFence {
15    /// Construct a nonzero version and nonzero manifest digest.
16    ///
17    /// # Errors
18    /// Rejects reserved zero identities.
19    pub fn new(version: u64, manifest_sha256: [u8; 32]) -> io::Result<Self> {
20        let version = NonZeroU64::new(version).ok_or_else(|| {
21            io::Error::new(io::ErrorKind::InvalidInput, "zero shuffle topology version")
22        })?;
23        if manifest_sha256 == [0; 32] {
24            return Err(io::Error::new(
25                io::ErrorKind::InvalidInput,
26                "zero shuffle topology digest",
27            ));
28        }
29        Ok(Self {
30            version,
31            manifest_sha256,
32        })
33    }
34
35    /// Logical topology version, never an assignment, recovery or serialization version.
36    #[must_use]
37    pub const fn version(self) -> u64 {
38        self.version.get()
39    }
40
41    /// Exact content identity of the installed catalog.
42    #[must_use]
43    pub const fn manifest_sha256(self) -> [u8; 32] {
44        self.manifest_sha256
45    }
46
47    /// Derive a fence from a validated catalog reference on the control path.
48    ///
49    /// # Errors
50    /// Rejects malformed references and reserved identities.
51    #[cfg(feature = "cluster")]
52    pub fn from_manifest(
53        version: crate::cluster::control::TopologyVersion,
54        manifest: &crate::cluster::control::CatalogManifestRef,
55    ) -> io::Result<Self> {
56        manifest.validate().map_err(io::Error::other)?;
57        let mut digest = [0; 32];
58        for (byte, encoded) in digest
59            .iter_mut()
60            .zip(manifest.sha256.as_bytes().as_chunks::<2>().0)
61        {
62            let encoded = std::str::from_utf8(encoded).map_err(io::Error::other)?;
63            *byte = u8::from_str_radix(encoded, 16).map_err(io::Error::other)?;
64        }
65        Self::new(version.get(), digest)
66    }
67}
laminar_core/shuffle/topology.rs

laminar_core/shuffle/
topology.rs
