lodestone-proto 0.0.16

//! Pod-incarnation-bound runtime material delivery.
//!
//! Kubernetes Secret objects are transport only.  The trusted init container
//! mounts the role Secret directory and selects the exact bundle path expanded
//! from its immutable Pod UID; the serving container never mounts that Secret.
//! The init process refuses to materialize files unless both Kubernetes object
//! digests and the immutable node role match.  The bundle carries no
//! node-allocation authority.

use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
use std::path::{Component, Path};

use lodestone_types::{KubernetesObjectUidDigest, NodeRole};

pub const RUNTIME_MATERIAL_FORMAT_VERSION: u16 = 1;
pub const MAX_RUNTIME_MATERIAL_BYTES: usize = 1024 * 1024;
pub const MAX_RUNTIME_MATERIAL_FILES: usize = 32;
pub const MAX_RUNTIME_MATERIAL_FILE_BYTES: usize = 512 * 1024;
pub const MAX_RUNTIME_MATERIAL_FILE_NAME_BYTES: usize = 128;
const RUNTIME_MATERIAL_MAGIC: &[u8; 8] = b"LDP7MAT1";
const RUNTIME_MATERIAL_DIGEST_BYTES: usize = 32;
const RUNTIME_MATERIAL_FIXED_HEADER_BYTES: usize = 8 + 2 + 1 + 32 + 32 + 2;
const MAX_RAW_KUBERNETES_UID_BYTES: usize = 256;
const MATERIAL_STAGING_DIRECTORY: &str = ".p7-material-staging-v1";
pub const RUNTIME_MATERIAL_CURRENT_DIRECTORY: &str = "current";

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RuntimeMaterialFileV1 {
    pub name: String,
    pub contents: Vec<u8>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RuntimeMaterialBundleV1 {
    pub format_version: u16,
    pub cluster_uid_digest: KubernetesObjectUidDigest,
    pub pod_uid_digest: KubernetesObjectUidDigest,
    pub role: NodeRole,
    pub files: Vec<RuntimeMaterialFileV1>,
}

impl RuntimeMaterialBundleV1 {
    pub fn validate(&self) -> Result<(), RuntimeMaterialError> {
        if self.format_version != RUNTIME_MATERIAL_FORMAT_VERSION {
            return Err(RuntimeMaterialError::UnsupportedVersion(self.format_version));
        }
        if self.files.is_empty() || self.files.len() > MAX_RUNTIME_MATERIAL_FILES {
            return Err(RuntimeMaterialError::FileCount(self.files.len()));
        }
        let mut previous: Option<&str> = None;
        let mut aggregate = RUNTIME_MATERIAL_FIXED_HEADER_BYTES
            .checked_add(RUNTIME_MATERIAL_DIGEST_BYTES)
            .ok_or(RuntimeMaterialError::SizeOverflow)?;
        for file in &self.files {
            if !valid_file_name(&file.name) {
                return Err(RuntimeMaterialError::InvalidFileName);
            }
            if file.contents.is_empty() || file.contents.len() > MAX_RUNTIME_MATERIAL_FILE_BYTES {
                return Err(RuntimeMaterialError::FileSize(file.contents.len()));
            }
            if previous.is_some_and(|name| name >= file.name.as_str()) {
                return Err(RuntimeMaterialError::FileOrder);
            }
            previous = Some(&file.name);
            aggregate = aggregate
                .checked_add(2)
                .and_then(|value| value.checked_add(4))
                .and_then(|value| value.checked_add(file.name.len()))
                .and_then(|value| value.checked_add(file.contents.len()))
                .ok_or(RuntimeMaterialError::SizeOverflow)?;
        }
        if aggregate > MAX_RUNTIME_MATERIAL_BYTES {
            return Err(RuntimeMaterialError::MessageTooLarge(aggregate));
        }
        Ok(())
    }
}

pub fn kubernetes_cluster_uid_digest(uid: &str) -> Result<KubernetesObjectUidDigest, RuntimeMaterialError> {
    kubernetes_uid_digest(b"lodestone/p7/kubernetes-cluster-uid/v1", uid)
}

pub fn kubernetes_pod_uid_digest(uid: &str) -> Result<KubernetesObjectUidDigest, RuntimeMaterialError> {
    kubernetes_uid_digest(b"lodestone/p7/kubernetes-pod-uid/v1", uid)
}

fn kubernetes_uid_digest(domain: &[u8], uid: &str) -> Result<KubernetesObjectUidDigest, RuntimeMaterialError> {
    if uid.is_empty() || uid.len() > MAX_RAW_KUBERNETES_UID_BYTES || !uid.bytes().all(|byte| byte.is_ascii_graphic()) {
        return Err(RuntimeMaterialError::InvalidKubernetesUid);
    }
    let mut hasher = blake3::Hasher::new();
    hasher.update(domain);
    hasher.update(&(uid.len() as u64).to_be_bytes());
    hasher.update(uid.as_bytes());
    KubernetesObjectUidDigest::new(*hasher.finalize().as_bytes())
        .map_err(|_| RuntimeMaterialError::InvalidKubernetesUid)
}

pub fn encode_runtime_material(bundle: &RuntimeMaterialBundleV1) -> Result<Vec<u8>, RuntimeMaterialError> {
    bundle.validate()?;
    let mut encoded = Vec::with_capacity(
        MAX_RUNTIME_MATERIAL_BYTES.min(
            RUNTIME_MATERIAL_FIXED_HEADER_BYTES
                + bundle
                    .files
                    .iter()
                    .map(|file| 6 + file.name.len() + file.contents.len())
                    .sum::<usize>()
                + RUNTIME_MATERIAL_DIGEST_BYTES,
        ),
    );
    encoded.extend_from_slice(RUNTIME_MATERIAL_MAGIC);
    encoded.extend_from_slice(&bundle.format_version.to_be_bytes());
    encoded.push(role_tag(bundle.role));
    encoded.extend_from_slice(bundle.cluster_uid_digest.as_bytes());
    encoded.extend_from_slice(bundle.pod_uid_digest.as_bytes());
    encoded.extend_from_slice(
        &u16::try_from(bundle.files.len())
            .map_err(|_| RuntimeMaterialError::SizeOverflow)?
            .to_be_bytes(),
    );
    for file in &bundle.files {
        encoded.extend_from_slice(
            &u16::try_from(file.name.len())
                .map_err(|_| RuntimeMaterialError::SizeOverflow)?
                .to_be_bytes(),
        );
        encoded.extend_from_slice(
            &u32::try_from(file.contents.len())
                .map_err(|_| RuntimeMaterialError::SizeOverflow)?
                .to_be_bytes(),
        );
        encoded.extend_from_slice(file.name.as_bytes());
        encoded.extend_from_slice(&file.contents);
    }
    let digest = material_digest(&encoded);
    encoded.extend_from_slice(&digest);
    if encoded.len() > MAX_RUNTIME_MATERIAL_BYTES {
        return Err(RuntimeMaterialError::MessageTooLarge(encoded.len()));
    }
    Ok(encoded)
}

pub fn decode_runtime_material(bytes: &[u8]) -> Result<RuntimeMaterialBundleV1, RuntimeMaterialError> {
    if bytes.len() < RUNTIME_MATERIAL_FIXED_HEADER_BYTES + RUNTIME_MATERIAL_DIGEST_BYTES
        || bytes.len() > MAX_RUNTIME_MATERIAL_BYTES
    {
        return Err(RuntimeMaterialError::MessageSize(bytes.len()));
    }
    let (payload, supplied_digest) = bytes.split_at(bytes.len() - RUNTIME_MATERIAL_DIGEST_BYTES);
    if material_digest(payload).as_slice() != supplied_digest {
        return Err(RuntimeMaterialError::DigestMismatch);
    }
    let mut decoder = MaterialDecoder::new(payload);
    if decoder.take(8)? != RUNTIME_MATERIAL_MAGIC {
        return Err(RuntimeMaterialError::Magic);
    }
    let format_version = decoder.u16()?;
    if format_version != RUNTIME_MATERIAL_FORMAT_VERSION {
        return Err(RuntimeMaterialError::UnsupportedVersion(format_version));
    }
    let role = role_from_tag(decoder.u8()?)?;
    let cluster_uid_digest =
        KubernetesObjectUidDigest::new(decoder.array_32()?).map_err(|_| RuntimeMaterialError::InvalidKubernetesUid)?;
    let pod_uid_digest =
        KubernetesObjectUidDigest::new(decoder.array_32()?).map_err(|_| RuntimeMaterialError::InvalidKubernetesUid)?;
    let file_count = usize::from(decoder.u16()?);
    if file_count == 0 || file_count > MAX_RUNTIME_MATERIAL_FILES {
        return Err(RuntimeMaterialError::FileCount(file_count));
    }
    let mut files = Vec::with_capacity(file_count);
    for _ in 0..file_count {
        let name_length = usize::from(decoder.u16()?);
        let content_length = usize::try_from(decoder.u32()?).map_err(|_| RuntimeMaterialError::SizeOverflow)?;
        if name_length == 0
            || name_length > MAX_RUNTIME_MATERIAL_FILE_NAME_BYTES
            || content_length == 0
            || content_length > MAX_RUNTIME_MATERIAL_FILE_BYTES
        {
            return Err(RuntimeMaterialError::InvalidFileShape);
        }
        let name = std::str::from_utf8(decoder.take(name_length)?)
            .map_err(|_| RuntimeMaterialError::InvalidFileName)?
            .to_owned();
        let contents = decoder.take(content_length)?.to_vec();
        files.push(RuntimeMaterialFileV1 { name, contents });
    }
    if !decoder.is_empty() {
        return Err(RuntimeMaterialError::TrailingBytes);
    }
    let bundle = RuntimeMaterialBundleV1 {
        format_version,
        cluster_uid_digest,
        pod_uid_digest,
        role,
        files,
    };
    bundle.validate()?;
    if encode_runtime_material(&bundle)? != bytes {
        return Err(RuntimeMaterialError::NonCanonical);
    }
    Ok(bundle)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RuntimeMaterializationReport {
    pub files: usize,
    pub bytes: usize,
    pub already_current: bool,
}

/// Materializes one exact pod-bound bundle into `<output>/current`.
///
/// The function is synchronous and bounded; async binaries must call it from a
/// bounded blocking task.  A crash leaves only the fixed staging directory or
/// a fully renamed current directory, both reconstructible from the Secret.
pub fn materialize_runtime_material(
    encoded: &[u8],
    expected_cluster_uid_digest: KubernetesObjectUidDigest,
    expected_pod_uid_digest: KubernetesObjectUidDigest,
    expected_role: NodeRole,
    required_entrypoint: &str,
    output_directory: &Path,
) -> Result<RuntimeMaterializationReport, RuntimeMaterialError> {
    if !valid_file_name(required_entrypoint) || !safe_output_directory(output_directory) {
        return Err(RuntimeMaterialError::InvalidMaterializationPath);
    }
    let bundle = decode_runtime_material(encoded)?;
    if bundle.cluster_uid_digest != expected_cluster_uid_digest
        || bundle.pod_uid_digest != expected_pod_uid_digest
        || bundle.role != expected_role
    {
        return Err(RuntimeMaterialError::IdentityMismatch);
    }
    if !bundle.files.iter().any(|file| file.name == required_entrypoint) {
        return Err(RuntimeMaterialError::MissingEntrypoint);
    }
    fs::create_dir_all(output_directory).map_err(RuntimeMaterialError::Io)?;
    reject_symlink(output_directory)?;
    fs::set_permissions(output_directory, fs::Permissions::from_mode(0o700)).map_err(RuntimeMaterialError::Io)?;

    let current = output_directory.join(RUNTIME_MATERIAL_CURRENT_DIRECTORY);
    let staging = output_directory.join(MATERIAL_STAGING_DIRECTORY);
    if current_matches(&current, &bundle)? {
        remove_material_directory(&staging)?;
        return Ok(materialization_report(&bundle, true));
    }
    remove_material_directory(&current)?;
    remove_material_directory(&staging)?;
    fs::create_dir(&staging).map_err(RuntimeMaterialError::Io)?;
    fs::set_permissions(&staging, fs::Permissions::from_mode(0o700)).map_err(RuntimeMaterialError::Io)?;
    for file in &bundle.files {
        let path = staging.join(&file.name);
        let mut output = OpenOptions::new()
            .write(true)
            .create_new(true)
            .mode(0o600)
            .open(path)
            .map_err(RuntimeMaterialError::Io)?;
        output
            .write_all(&file.contents)
            .and_then(|()| output.sync_all())
            .map_err(RuntimeMaterialError::Io)?;
    }
    sync_directory(&staging)?;
    fs::rename(&staging, &current).map_err(RuntimeMaterialError::Io)?;
    sync_directory(output_directory)?;
    Ok(materialization_report(&bundle, false))
}

fn materialization_report(bundle: &RuntimeMaterialBundleV1, already_current: bool) -> RuntimeMaterializationReport {
    RuntimeMaterializationReport {
        files: bundle.files.len(),
        bytes: bundle.files.iter().map(|file| file.contents.len()).sum(),
        already_current,
    }
}

fn current_matches(current: &Path, bundle: &RuntimeMaterialBundleV1) -> Result<bool, RuntimeMaterialError> {
    let metadata = match fs::symlink_metadata(current) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(error) => return Err(RuntimeMaterialError::Io(error)),
    };
    if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
        return Err(RuntimeMaterialError::InvalidMaterializationPath);
    }
    let mut observed = Vec::with_capacity(bundle.files.len());
    let mut observed_bytes = 0_usize;
    for entry in fs::read_dir(current).map_err(RuntimeMaterialError::Io)? {
        if observed.len() >= MAX_RUNTIME_MATERIAL_FILES {
            return Err(RuntimeMaterialError::FileCount(observed.len().saturating_add(1)));
        }
        let entry = entry.map_err(RuntimeMaterialError::Io)?;
        let file_type = entry.file_type().map_err(RuntimeMaterialError::Io)?;
        if !file_type.is_file() || file_type.is_symlink() {
            return Err(RuntimeMaterialError::InvalidMaterializationPath);
        }
        let metadata = entry.metadata().map_err(RuntimeMaterialError::Io)?;
        let length = usize::try_from(metadata.len()).map_err(|_| RuntimeMaterialError::SizeOverflow)?;
        observed_bytes = observed_bytes
            .checked_add(length)
            .ok_or(RuntimeMaterialError::SizeOverflow)?;
        if length == 0 || length > MAX_RUNTIME_MATERIAL_FILE_BYTES || observed_bytes > MAX_RUNTIME_MATERIAL_BYTES {
            return Err(RuntimeMaterialError::FileSize(length));
        }
        let name = entry
            .file_name()
            .into_string()
            .map_err(|_| RuntimeMaterialError::InvalidMaterializationPath)?;
        if !valid_file_name(&name) {
            return Err(RuntimeMaterialError::InvalidMaterializationPath);
        }
        let contents = fs::read(entry.path()).map_err(RuntimeMaterialError::Io)?;
        observed.push((name, contents));
    }
    observed.sort_by(|left, right| left.0.cmp(&right.0));
    Ok(observed.len() == bundle.files.len()
        && observed
            .iter()
            .zip(&bundle.files)
            .all(|((name, contents), expected)| name == &expected.name && contents == &expected.contents))
}

fn remove_material_directory(path: &Path) -> Result<(), RuntimeMaterialError> {
    let metadata = match fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
        Err(error) => return Err(RuntimeMaterialError::Io(error)),
    };
    if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
        return Err(RuntimeMaterialError::InvalidMaterializationPath);
    }
    fs::remove_dir_all(path).map_err(RuntimeMaterialError::Io)
}

fn reject_symlink(path: &Path) -> Result<(), RuntimeMaterialError> {
    let metadata = fs::symlink_metadata(path).map_err(RuntimeMaterialError::Io)?;
    if !metadata.file_type().is_dir() || metadata.file_type().is_symlink() {
        return Err(RuntimeMaterialError::InvalidMaterializationPath);
    }
    Ok(())
}

fn sync_directory(path: &Path) -> Result<(), RuntimeMaterialError> {
    let directory = OpenOptions::new()
        .read(true)
        .open(path)
        .map_err(RuntimeMaterialError::Io)?;
    directory.sync_all().map_err(RuntimeMaterialError::Io)
}

fn safe_output_directory(path: &Path) -> bool {
    path.is_absolute()
        && path
            .components()
            .all(|component| matches!(component, Component::RootDir | Component::Normal(_)))
        && path
            .components()
            .filter(|component| matches!(component, Component::Normal(_)))
            .count()
            >= 2
}

fn material_digest(payload: &[u8]) -> [u8; 32] {
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone/p7/runtime-material-bundle/v1");
    hasher.update(&(payload.len() as u64).to_be_bytes());
    hasher.update(payload);
    *hasher.finalize().as_bytes()
}

fn valid_file_name(name: &str) -> bool {
    !name.is_empty()
        && name.len() <= MAX_RUNTIME_MATERIAL_FILE_NAME_BYTES
        && name.as_bytes().first().is_some_and(u8::is_ascii_alphanumeric)
        && name.as_bytes().last().is_some_and(u8::is_ascii_alphanumeric)
        && name
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
}

const fn role_tag(role: NodeRole) -> u8 {
    match role {
        NodeRole::MetastoreVoter => 1,
        NodeRole::Placement => 2,
        NodeRole::Orchestrator => 3,
        NodeRole::Gateway => 4,
        NodeRole::Indexer => 5,
        NodeRole::Searcher => 6,
        NodeRole::Compactor => 7,
    }
}

fn role_from_tag(tag: u8) -> Result<NodeRole, RuntimeMaterialError> {
    match tag {
        1 => Ok(NodeRole::MetastoreVoter),
        2 => Ok(NodeRole::Placement),
        3 => Ok(NodeRole::Orchestrator),
        4 => Ok(NodeRole::Gateway),
        5 => Ok(NodeRole::Indexer),
        6 => Ok(NodeRole::Searcher),
        7 => Ok(NodeRole::Compactor),
        _ => Err(RuntimeMaterialError::InvalidRole(tag)),
    }
}

struct MaterialDecoder<'a> {
    remaining: &'a [u8],
}

impl<'a> MaterialDecoder<'a> {
    const fn new(bytes: &'a [u8]) -> Self {
        Self { remaining: bytes }
    }

    fn take(&mut self, length: usize) -> Result<&'a [u8], RuntimeMaterialError> {
        let (head, tail) = self
            .remaining
            .split_at_checked(length)
            .ok_or(RuntimeMaterialError::Truncated)?;
        self.remaining = tail;
        Ok(head)
    }

    fn u8(&mut self) -> Result<u8, RuntimeMaterialError> {
        self.take(1)?.first().copied().ok_or(RuntimeMaterialError::Truncated)
    }

    fn u16(&mut self) -> Result<u16, RuntimeMaterialError> {
        let bytes: [u8; 2] = self.take(2)?.try_into().map_err(|_| RuntimeMaterialError::Truncated)?;
        Ok(u16::from_be_bytes(bytes))
    }

    fn u32(&mut self) -> Result<u32, RuntimeMaterialError> {
        let bytes: [u8; 4] = self.take(4)?.try_into().map_err(|_| RuntimeMaterialError::Truncated)?;
        Ok(u32::from_be_bytes(bytes))
    }

    fn array_32(&mut self) -> Result<[u8; 32], RuntimeMaterialError> {
        self.take(32)?.try_into().map_err(|_| RuntimeMaterialError::Truncated)
    }

    const fn is_empty(&self) -> bool {
        self.remaining.is_empty()
    }
}

#[derive(Debug)]
pub enum RuntimeMaterialError {
    DigestMismatch,
    FileCount(usize),
    FileOrder,
    FileSize(usize),
    InvalidFileName,
    InvalidFileShape,
    InvalidKubernetesUid,
    IdentityMismatch,
    InvalidMaterializationPath,
    InvalidRole(u8),
    Magic,
    MessageSize(usize),
    MessageTooLarge(usize),
    MissingEntrypoint,
    NonCanonical,
    SizeOverflow,
    TrailingBytes,
    Truncated,
    UnsupportedVersion(u16),
    Io(std::io::Error),
}

impl Display for RuntimeMaterialError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        write!(formatter, "{self:?}")
    }
}

impl Error for RuntimeMaterialError {}

#[cfg(test)]
mod tests {
    use super::*;

    fn bundle() -> Result<RuntimeMaterialBundleV1, Box<dyn Error>> {
        Ok(RuntimeMaterialBundleV1 {
            format_version: RUNTIME_MATERIAL_FORMAT_VERSION,
            cluster_uid_digest: kubernetes_cluster_uid_digest("cluster-uid")?,
            pod_uid_digest: kubernetes_pod_uid_digest("pod-uid-1")?,
            role: NodeRole::Indexer,
            files: vec![
                RuntimeMaterialFileV1 {
                    name: "lodestone.json".to_owned(),
                    contents: br#"{"format_version":1}"#.to_vec(),
                },
                RuntimeMaterialFileV1 {
                    name: "node-key.der".to_owned(),
                    contents: vec![7; 48],
                },
            ],
        })
    }

    #[test]
    fn exact_bundle_round_trips_and_binds_pod_incarnation() -> Result<(), Box<dyn Error>> {
        let first = bundle()?;
        let encoded = encode_runtime_material(&first)?;
        assert_eq!(decode_runtime_material(&encoded)?, first);

        let mut replacement = first.clone();
        replacement.pod_uid_digest = kubernetes_pod_uid_digest("pod-uid-2")?;
        assert_ne!(encode_runtime_material(&replacement)?, encoded);
        Ok(())
    }

    #[test]
    fn malformed_or_noncanonical_bundles_fail_closed() -> Result<(), Box<dyn Error>> {
        let encoded = encode_runtime_material(&bundle()?)?;
        for length in 0..encoded.len() {
            assert!(decode_runtime_material(&encoded[..length]).is_err());
        }
        let mut changed = encoded.clone();
        changed[10] = 0xff;
        assert!(matches!(
            decode_runtime_material(&changed),
            Err(RuntimeMaterialError::DigestMismatch)
        ));
        let mut unsorted = bundle()?;
        unsorted.files.reverse();
        assert!(matches!(
            encode_runtime_material(&unsorted),
            Err(RuntimeMaterialError::FileOrder)
        ));
        Ok(())
    }

    #[test]
    fn paths_and_uid_inputs_are_strict() -> Result<(), Box<dyn Error>> {
        let mut invalid = bundle()?;
        invalid.files[0].name = "../lodestone.json".to_owned();
        assert!(matches!(invalid.validate(), Err(RuntimeMaterialError::InvalidFileName)));
        assert!(kubernetes_pod_uid_digest("").is_err());
        assert!(kubernetes_cluster_uid_digest("bad uid").is_err());
        Ok(())
    }

    #[test]
    fn materialization_is_exact_idempotent_and_replaces_partial_staging() -> Result<(), Box<dyn Error>> {
        let root = std::env::temp_dir().join(format!(
            "lodestone-p7-material-{}-{}",
            std::process::id(),
            std::thread::current().name().unwrap_or("unnamed")
        ));
        if root.exists() {
            fs::remove_dir_all(&root)?;
        }
        fs::create_dir_all(&root)?;
        let bundle = bundle()?;
        let encoded = encode_runtime_material(&bundle)?;
        let first = materialize_runtime_material(
            &encoded,
            bundle.cluster_uid_digest,
            bundle.pod_uid_digest,
            bundle.role,
            "lodestone.json",
            &root,
        )?;
        assert!(!first.already_current);
        assert_eq!(fs::read(root.join("current/node-key.der"))?, vec![7; 48]);
        let second = materialize_runtime_material(
            &encoded,
            bundle.cluster_uid_digest,
            bundle.pod_uid_digest,
            bundle.role,
            "lodestone.json",
            &root,
        )?;
        assert!(second.already_current);
        fs::create_dir(root.join(MATERIAL_STAGING_DIRECTORY))?;
        fs::write(root.join(MATERIAL_STAGING_DIRECTORY).join("partial"), b"partial")?;
        let third = materialize_runtime_material(
            &encoded,
            bundle.cluster_uid_digest,
            bundle.pod_uid_digest,
            bundle.role,
            "lodestone.json",
            &root,
        )?;
        assert!(third.already_current);
        assert!(!root.join(MATERIAL_STAGING_DIRECTORY).exists());
        fs::remove_dir_all(root)?;
        Ok(())
    }

    #[test]
    fn materialization_refuses_wrong_pod_role_and_entrypoint() -> Result<(), Box<dyn Error>> {
        let root = std::env::temp_dir().join(format!("lodestone-p7-material-refusal-{}", std::process::id()));
        if root.exists() {
            fs::remove_dir_all(&root)?;
        }
        fs::create_dir_all(&root)?;
        let bundle = bundle()?;
        let encoded = encode_runtime_material(&bundle)?;
        assert!(matches!(
            materialize_runtime_material(
                &encoded,
                bundle.cluster_uid_digest,
                kubernetes_pod_uid_digest("replacement")?,
                bundle.role,
                "lodestone.json",
                &root,
            ),
            Err(RuntimeMaterialError::IdentityMismatch)
        ));
        assert!(matches!(
            materialize_runtime_material(
                &encoded,
                bundle.cluster_uid_digest,
                bundle.pod_uid_digest,
                NodeRole::Searcher,
                "lodestone.json",
                &root,
            ),
            Err(RuntimeMaterialError::IdentityMismatch)
        ));
        assert!(matches!(
            materialize_runtime_material(
                &encoded,
                bundle.cluster_uid_digest,
                bundle.pod_uid_digest,
                bundle.role,
                "rpc.json",
                &root,
            ),
            Err(RuntimeMaterialError::MissingEntrypoint)
        ));
        fs::remove_dir_all(root)?;
        Ok(())
    }
}