lodestone-proto 0.0.16

//! Bounded rollout projections.
//!
//! These values carry no membership or lifecycle authority.  They bind a
//! Kubernetes Downward API projection to the two exact P6 incarnations that
//! the current voter must submit through the existing P6 membership method.

use std::error::Error;
use std::fmt::{self, Display, Formatter};

use lodestone_types::NodeToken;
use serde::{Deserialize, Serialize};

pub const VOTER_REPLACEMENT_DIRECTIVE_FORMAT_VERSION: u16 = 1;
pub const MAX_VOTER_REPLACEMENT_DIRECTIVE_BYTES: usize = 4 * 1024;

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct VoterReplacementDirectiveV1 {
    pub format_version: u16,
    pub remove: NodeToken,
    pub candidate: NodeToken,
    pub semantic_digest: [u8; 32],
}

impl VoterReplacementDirectiveV1 {
    pub fn validate(&self) -> Result<(), RolloutWireError> {
        if self.format_version != VOTER_REPLACEMENT_DIRECTIVE_FORMAT_VERSION {
            return Err(RolloutWireError::UnsupportedVersion(self.format_version));
        }
        if self.remove == self.candidate
            || self.remove.node_id == self.candidate.node_id
            || self.remove.recovery_era != self.candidate.recovery_era
        {
            return Err(RolloutWireError::InvalidReplacement);
        }
        if self.semantic_digest != self.derived_semantic_digest()? {
            return Err(RolloutWireError::SemanticDigestMismatch);
        }
        Ok(())
    }

    pub fn derive_semantic_digest(&mut self) -> Result<(), RolloutWireError> {
        self.semantic_digest = self.derived_semantic_digest()?;
        Ok(())
    }

    fn derived_semantic_digest(&self) -> Result<[u8; 32], RolloutWireError> {
        let canonical = serde_json::to_vec(&(self.format_version, self.remove, self.candidate))
            .map_err(|error| RolloutWireError::Codec(error.to_string()))?;
        let mut hasher = blake3::Hasher::new();
        hasher.update(b"lodestone/p7/voter-replacement-directive/v1");
        hasher.update(&canonical);
        Ok(*hasher.finalize().as_bytes())
    }
}

pub fn encode_voter_replacement_directive(
    directive: &VoterReplacementDirectiveV1,
) -> Result<Vec<u8>, RolloutWireError> {
    directive.validate()?;
    let encoded = serde_json::to_vec(directive).map_err(|error| RolloutWireError::Codec(error.to_string()))?;
    if encoded.len() > MAX_VOTER_REPLACEMENT_DIRECTIVE_BYTES {
        return Err(RolloutWireError::MessageTooLarge(encoded.len()));
    }
    Ok(encoded)
}

pub fn decode_voter_replacement_directive(bytes: &[u8]) -> Result<VoterReplacementDirectiveV1, RolloutWireError> {
    if bytes.len() > MAX_VOTER_REPLACEMENT_DIRECTIVE_BYTES {
        return Err(RolloutWireError::MessageTooLarge(bytes.len()));
    }
    let directive: VoterReplacementDirectiveV1 =
        serde_json::from_slice(bytes).map_err(|error| RolloutWireError::Codec(error.to_string()))?;
    directive.validate()?;
    if serde_json::to_vec(&directive).map_err(|error| RolloutWireError::Codec(error.to_string()))? != bytes {
        return Err(RolloutWireError::NonCanonical);
    }
    Ok(directive)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RolloutWireError {
    Codec(String),
    InvalidReplacement,
    MessageTooLarge(usize),
    NonCanonical,
    SemanticDigestMismatch,
    UnsupportedVersion(u16),
}

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

impl Error for RolloutWireError {}

#[cfg(test)]
mod tests {
    use super::*;
    use lodestone_types::{NodeId, NodeIncarnation, RecoveryEra};

    fn token(node: u64) -> Result<NodeToken, Box<dyn Error>> {
        Ok(NodeToken::new(
            RecoveryEra::new(7)?,
            NodeId::new(node),
            NodeIncarnation::new(1)?,
        )?)
    }

    fn directive() -> Result<VoterReplacementDirectiveV1, Box<dyn Error>> {
        let mut directive = VoterReplacementDirectiveV1 {
            format_version: VOTER_REPLACEMENT_DIRECTIVE_FORMAT_VERSION,
            remove: token(1)?,
            candidate: token(2)?,
            semantic_digest: [0; 32],
        };
        directive.derive_semantic_digest()?;
        Ok(directive)
    }

    #[test]
    fn exact_voter_directive_round_trips() -> Result<(), Box<dyn Error>> {
        let directive = directive()?;
        let encoded = encode_voter_replacement_directive(&directive)?;
        assert_eq!(decode_voter_replacement_directive(&encoded)?, directive);
        Ok(())
    }

    #[test]
    fn substitution_future_version_and_noncanonical_inputs_fail_closed() -> Result<(), Box<dyn Error>> {
        let directive = directive()?;
        let mut substituted = directive.clone();
        substituted.candidate = token(3)?;
        let substituted = serde_json::to_vec(&substituted)?;
        assert!(matches!(
            decode_voter_replacement_directive(&substituted),
            Err(RolloutWireError::SemanticDigestMismatch)
        ));

        let mut future = directive.clone();
        future.format_version = 2;
        assert!(matches!(
            decode_voter_replacement_directive(&serde_json::to_vec(&future)?),
            Err(RolloutWireError::UnsupportedVersion(2))
        ));

        let mut noncanonical = encode_voter_replacement_directive(&directive)?;
        noncanonical.push(b'\n');
        assert!(matches!(
            decode_voter_replacement_directive(&noncanonical),
            Err(RolloutWireError::NonCanonical)
        ));
        Ok(())
    }
}