tryaudex_core/

ha.rs

use serde::{Deserialize, Serialize};

/// High availability backend type.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum HaBackend {
    Redis,
    Etcd,
}

impl std::fmt::Display for HaBackend {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Redis => write!(f, "redis"),
            Self::Etcd => write!(f, "etcd"),
        }
    }
}

/// Configuration for high availability mode.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HaConfig {
    /// Backend: "redis" or "etcd"
    pub backend: HaBackend,
    /// Connection URL(s).
    /// - Redis: "redis://localhost:6379" or "redis+sentinel://host:26379/mymaster"
    /// - Etcd: "http://localhost:2379" (comma-separated for cluster)
    pub endpoints: Vec<String>,
    /// Password for backend authentication
    pub password: Option<String>,
    /// TLS enabled
    #[serde(default)]
    pub tls: bool,
    /// Key prefix for all HA keys (default: "audex:")
    #[serde(default = "default_prefix")]
    pub prefix: String,
    /// Leader election configuration
    #[serde(default)]
    pub leader: LeaderConfig,
    /// Distributed rate limiting configuration
    #[serde(default)]
    pub rate_limit: DistributedRateLimitConfig,
    /// Audit replication configuration
    #[serde(default)]
    pub audit_replication: AuditReplicationConfig,
}

fn default_prefix() -> String {
    "audex:".to_string()
}

impl HaConfig {
    /// Return the backend password, preferring the `AUDEX_HA_PASSWORD` environment
    /// variable over the config file value.
    ///
    /// # Security note (R3-L15)
    /// Storing a plaintext password in the config file is discouraged. Set the
    /// `AUDEX_HA_PASSWORD` environment variable instead; it takes precedence over
    /// the `password` field in the config so that secrets can be injected at
    /// runtime (e.g. via a secrets manager or container environment) without
    /// persisting them on disk.
    pub fn resolve_password(&self) -> Option<String> {
        std::env::var("AUDEX_HA_PASSWORD")
            .ok()
            .filter(|v| !v.is_empty())
            .or_else(|| self.password.clone())
    }
}

impl Default for HaConfig {
    fn default() -> Self {
        Self {
            backend: HaBackend::Redis,
            endpoints: vec!["redis://localhost:6379".to_string()],
            password: None,
            tls: false,
            prefix: default_prefix(),
            leader: LeaderConfig::default(),
            rate_limit: DistributedRateLimitConfig::default(),
            audit_replication: AuditReplicationConfig::default(),
        }
    }
}

/// Leader election configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LeaderConfig {
    /// Unique node ID for this instance (auto-generated if not set)
    pub node_id: Option<String>,
    /// Lease TTL in seconds (default: 15)
    #[serde(default = "default_lease_ttl")]
    pub lease_ttl: u64,
    /// Renewal interval in seconds (default: 5, should be < lease_ttl/3)
    #[serde(default = "default_renewal_interval")]
    pub renewal_interval: u64,
}

fn default_lease_ttl() -> u64 {
    15
}

fn default_renewal_interval() -> u64 {
    5
}

impl LeaderConfig {
    /// Validate that `renewal_interval` is less than `lease_ttl / 3`.
    ///
    /// If renewal takes longer than a third of the TTL, network hiccups can
    /// cause the lease to expire before the next renewal succeeds, leading to
    /// split-brain where two nodes both believe they are leader.
    pub fn validate(&self) -> crate::error::Result<()> {
        if self.lease_ttl == 0 {
            return Err(crate::error::AvError::InvalidPolicy(
                "HA leader lease_ttl must be > 0".to_string(),
            ));
        }
        if self.renewal_interval == 0 {
            return Err(crate::error::AvError::InvalidPolicy(
                "HA leader renewal_interval must be > 0".to_string(),
            ));
        }
        // Use multiplication instead of division to avoid integer truncation.
        // e.g. lease_ttl=7, renewal_interval=2: 2*3=6 < 7 passes correctly,
        // whereas 2 >= 7/3=2 incorrectly fails.
        //
        // R6-M66: use checked_mul to prevent u64 overflow. A crafted
        // renewal_interval near u64::MAX/2 would wrap to a small value
        // and silently pass the comparison.
        let triple = self.renewal_interval.checked_mul(3).ok_or_else(|| {
            crate::error::AvError::InvalidPolicy(format!(
                "HA leader renewal_interval ({}) overflows when multiplied by 3",
                self.renewal_interval
            ))
        })?;
        if triple >= self.lease_ttl {
            return Err(crate::error::AvError::InvalidPolicy(format!(
                "HA leader renewal_interval ({}) must be < lease_ttl/3 ({}/3 = {}). \
                 Otherwise network hiccups can cause lease loss and split-brain.",
                self.renewal_interval,
                self.lease_ttl,
                self.lease_ttl / 3
            )));
        }
        Ok(())
    }
}

impl Default for LeaderConfig {
    fn default() -> Self {
        Self {
            node_id: None,
            lease_ttl: default_lease_ttl(),
            renewal_interval: default_renewal_interval(),
        }
    }
}

/// Distributed rate limiting configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DistributedRateLimitConfig {
    /// Enable distributed rate limiting (default: true)
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Sliding window size in seconds (default: 3600 = 1 hour)
    #[serde(default = "default_window")]
    pub window_seconds: u64,
    /// Key expiry in seconds (default: 7200 = 2 hours)
    #[serde(default = "default_expiry")]
    pub expiry_seconds: u64,
}

fn default_true() -> bool {
    true
}

fn default_window() -> u64 {
    3600
}

fn default_expiry() -> u64 {
    7200
}

impl Default for DistributedRateLimitConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            window_seconds: default_window(),
            expiry_seconds: default_expiry(),
        }
    }
}

/// Audit replication configuration.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuditReplicationConfig {
    /// Enable audit replication across nodes (default: true)
    #[serde(default = "default_true")]
    pub enabled: bool,
    /// Stream/channel name for audit events (default: "audex:audit:stream")
    pub stream: Option<String>,
    /// Max stream length before trimming (default: 10000)
    #[serde(default = "default_max_stream")]
    pub max_stream_length: u64,
    /// Consumer group name (auto-generated from node_id if not set)
    pub consumer_group: Option<String>,
}

fn default_max_stream() -> u64 {
    10000
}

impl Default for AuditReplicationConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            stream: None,
            max_stream_length: default_max_stream(),
            consumer_group: None,
        }
    }
}

/// Leader election state.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LeaderState {
    pub node_id: String,
    pub is_leader: bool,
    pub leader_node: Option<String>,
    pub lease_expires_at: Option<String>,
    pub last_heartbeat: Option<String>,
}

/// Redis key generators for HA operations.
pub struct HaKeys {
    prefix: String,
}

impl HaKeys {
    pub fn new(prefix: &str) -> Self {
        // Sanitize prefix: only allow alphanumeric, colon, hyphen, underscore, dot.
        // Reject characters that could be used for Redis protocol injection
        // (newlines, spaces, null bytes).
        let sanitized: String = prefix
            .chars()
            .filter(|c| {
                c.is_ascii_alphanumeric() || *c == ':' || *c == '-' || *c == '_' || *c == '.'
            })
            .collect();
        if sanitized != prefix {
            tracing::warn!(
                original = prefix,
                sanitized = %sanitized,
                "HA key prefix contained invalid characters and was sanitized"
            );
        }
        Self { prefix: sanitized }
    }

    /// Sanitize a user-provided key component (identity, node_id) to prevent
    /// Redis protocol injection via newlines or control characters.
    fn sanitize_component(s: &str) -> String {
        s.chars()
            .filter(|c| {
                c.is_ascii_alphanumeric()
                    || *c == ':'
                    || *c == '-'
                    || *c == '_'
                    || *c == '.'
                    || *c == '@'
            })
            .collect()
    }

    /// Leader election lock key.
    pub fn leader_lock(&self) -> String {
        format!("{}leader:lock", self.prefix)
    }

    /// Leader heartbeat key.
    pub fn leader_heartbeat(&self) -> String {
        format!("{}leader:heartbeat", self.prefix)
    }

    /// Rate limit counter key for an identity.
    pub fn rate_limit(&self, identity: &str) -> String {
        format!(
            "{}ratelimit:{}",
            self.prefix,
            Self::sanitize_component(identity)
        )
    }

    /// Rate limit sorted set key for sliding window.
    pub fn rate_limit_window(&self, identity: &str) -> String {
        format!(
            "{}ratelimit:window:{}",
            self.prefix,
            Self::sanitize_component(identity)
        )
    }

    /// Audit replication stream key.
    pub fn audit_stream(&self) -> String {
        format!("{}audit:stream", self.prefix)
    }

    /// Session registry key (tracks active sessions across nodes).
    pub fn session_registry(&self) -> String {
        format!("{}sessions:active", self.prefix)
    }

    /// Node registry key.
    pub fn node_registry(&self) -> String {
        format!("{}nodes:active", self.prefix)
    }

    /// Per-node session count key.
    pub fn node_sessions(&self, node_id: &str) -> String {
        format!(
            "{}nodes:{}:sessions",
            self.prefix,
            Self::sanitize_component(node_id)
        )
    }
}

/// Redis commands for leader election using SET NX EX (atomic lock).
pub struct LeaderElectionCommands;

impl LeaderElectionCommands {
    /// Attempt to acquire leadership.
    /// Uses SET key value NX EX ttl for atomic acquire.
    pub fn try_acquire(key: &str, node_id: &str, ttl_secs: u64) -> Vec<String> {
        vec![
            "SET".to_string(),
            key.to_string(),
            node_id.to_string(),
            "NX".to_string(),
            "EX".to_string(),
            ttl_secs.to_string(),
        ]
    }

    /// Renew leadership lease (only if we still hold it).
    /// Uses a Lua script for atomic check-and-renew.
    /// ARGV[2] is lease TTL in seconds (matching `HaConfig::lease_ttl`).
    pub fn renew_lua() -> &'static str {
        r#"
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("EXPIRE", KEYS[1], ARGV[2])
else
    return 0
end
"#
    }

    /// Release leadership (only if we hold it).
    pub fn release_lua() -> &'static str {
        r#"
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end
"#
    }

    /// Get current leader.
    pub fn get_leader(key: &str) -> Vec<String> {
        vec!["GET".to_string(), key.to_string()]
    }
}

/// Redis commands for distributed rate limiting using sliding window.
pub struct DistributedRateLimitCommands;

impl DistributedRateLimitCommands {
    /// Record a session and check rate limit.
    /// Uses ZADD + ZREMRANGEBYSCORE + ZCARD in a pipeline for atomic sliding window.
    pub fn check_and_record_lua() -> &'static str {
        r#"
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])
local expiry = tonumber(ARGV[4])
local member = ARGV[5]

-- Remove expired entries
redis.call("ZREMRANGEBYSCORE", key, 0, now - window)

-- Count current entries in window
local count = redis.call("ZCARD", key)

if count >= limit then
    return {0, count}
end

-- Add new entry
redis.call("ZADD", key, now, member)
redis.call("EXPIRE", key, expiry)

return {1, count + 1}
"#
    }

    /// Get current count for an identity.
    pub fn get_count(key: &str, window_start: u64) -> Vec<String> {
        vec![
            "ZCOUNT".to_string(),
            key.to_string(),
            window_start.to_string(),
            "+inf".to_string(),
        ]
    }
}

/// Redis commands for audit stream replication.
pub struct AuditReplicationCommands;

impl AuditReplicationCommands {
    /// Publish an audit entry to the stream.
    /// XADD key MAXLEN ~ max_length * data
    pub fn publish_fields(entry_json: &str) -> Vec<(&str, &str)> {
        vec![("data", entry_json)]
    }

    /// Create consumer group (idempotent with MKSTREAM).
    pub fn create_group(stream_key: &str, group: &str) -> Vec<String> {
        vec![
            "XGROUP".to_string(),
            "CREATE".to_string(),
            stream_key.to_string(),
            group.to_string(),
            "0".to_string(),
            "MKSTREAM".to_string(),
        ]
    }

    /// Read new entries from the stream for a consumer.
    pub fn read_new(stream_key: &str, group: &str, consumer: &str, count: u64) -> Vec<String> {
        vec![
            "XREADGROUP".to_string(),
            "GROUP".to_string(),
            group.to_string(),
            consumer.to_string(),
            "COUNT".to_string(),
            count.to_string(),
            "STREAMS".to_string(),
            stream_key.to_string(),
            ">".to_string(),
        ]
    }

    /// Acknowledge processed entries.
    pub fn ack(stream_key: &str, group: &str, ids: &[&str]) -> Vec<String> {
        let mut cmd = vec![
            "XACK".to_string(),
            stream_key.to_string(),
            group.to_string(),
        ];
        for id in ids {
            cmd.push(id.to_string());
        }
        cmd
    }
}

/// Etcd key-value operations for HA (alternative to Redis).
pub struct EtcdOperations;

impl EtcdOperations {
    /// Leader election uses etcd leases with keep-alive.
    /// PUT key value --lease=LEASE_ID
    pub fn acquire_lock_endpoint() -> &'static str {
        "/v3/kv/put"
    }

    /// Create a lease.
    pub fn grant_lease_endpoint() -> &'static str {
        "/v3/lease/grant"
    }

    /// Keep lease alive.
    pub fn keepalive_endpoint() -> &'static str {
        "/v3/lease/keepalive"
    }

    /// Watch for leader changes.
    pub fn watch_endpoint() -> &'static str {
        "/v3/watch"
    }

    /// Format a lease grant request body.
    pub fn lease_grant_body(ttl: u64) -> String {
        format!("{{\"TTL\":{}}}", ttl)
    }

    /// Format a put-if-not-exists request (using transactions).
    pub fn txn_create_body(key: &str, value: &str, lease_id: u64) -> String {
        let key_b64 = base64_encode(key);
        let value_b64 = base64_encode(value);
        format!(
            r#"{{"compare":[{{"result":"EQUAL","target":"CREATE","key":"{}"}}],"success":[{{"request_put":{{"key":"{}","value":"{}","lease":{}}}}}],"failure":[]}}"#,
            key_b64, key_b64, value_b64, lease_id
        )
    }
}

/// Simple base64 encoding for etcd API (no padding).
fn base64_encode(input: &str) -> String {
    use std::io::Write;
    let mut buf = Vec::new();
    {
        let mut encoder = Base64Writer::new(&mut buf);
        encoder.write_all(input.as_bytes()).ok();
    }
    String::from_utf8(buf).unwrap_or_default()
}

/// Minimal base64 encoder (avoid external dependency).
struct Base64Writer<'a> {
    out: &'a mut Vec<u8>,
    buf: [u8; 3],
    len: usize,
}

const B64_CHARS: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

impl<'a> Base64Writer<'a> {
    fn new(out: &'a mut Vec<u8>) -> Self {
        Self {
            out,
            buf: [0; 3],
            len: 0,
        }
    }

    fn flush_buf(&mut self) {
        if self.len == 0 {
            return;
        }
        let b = self.buf;
        self.out.push(B64_CHARS[(b[0] >> 2) as usize]);
        self.out
            .push(B64_CHARS[((b[0] & 0x03) << 4 | b[1] >> 4) as usize]);
        if self.len > 1 {
            self.out
                .push(B64_CHARS[((b[1] & 0x0f) << 2 | b[2] >> 6) as usize]);
        } else {
            self.out.push(b'=');
        }
        if self.len > 2 {
            self.out.push(B64_CHARS[(b[2] & 0x3f) as usize]);
        } else {
            self.out.push(b'=');
        }
        self.buf = [0; 3];
        self.len = 0;
    }
}

impl<'a> std::io::Write for Base64Writer<'a> {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        for &byte in data {
            self.buf[self.len] = byte;
            self.len += 1;
            if self.len == 3 {
                self.flush_buf();
            }
        }
        Ok(data.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.flush_buf();
        Ok(())
    }
}

impl<'a> Drop for Base64Writer<'a> {
    fn drop(&mut self) {
        self.flush_buf();
    }
}

/// Generate a unique node ID for this instance.
///
/// R6-M65: the raw hostname may contain characters unsafe for use as
/// Redis/etcd keys (newlines, spaces, null bytes, non-ASCII). Sanitize
/// through the same allowlist used for all key components so the node ID
/// is always a safe key suffix.
pub fn generate_node_id() -> String {
    let raw_hostname = std::fs::read_to_string("/etc/hostname")
        .map(|h| h.trim().to_string())
        .unwrap_or_else(|_| "unknown".to_string());
    let hostname = HaKeys::sanitize_component(&raw_hostname);
    // If sanitization stripped everything, fall back to "unknown".
    let hostname = if hostname.is_empty() {
        "unknown".to_string()
    } else {
        hostname
    };
    let id = uuid::Uuid::new_v4();
    format!("{}-{}", hostname, id)
}

/// Cluster status information.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterStatus {
    pub node_id: String,
    pub is_leader: bool,
    pub leader_node: Option<String>,
    pub active_nodes: Vec<NodeInfo>,
    pub total_sessions: u64,
    pub backend: String,
}

/// Information about an active node in the cluster.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeInfo {
    pub node_id: String,
    pub address: String,
    pub sessions: u64,
    pub last_seen: String,
}

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

    #[test]
    fn test_ha_config_default() {
        let config = HaConfig::default();
        assert_eq!(config.backend, HaBackend::Redis);
        assert_eq!(config.endpoints.len(), 1);
        assert_eq!(config.prefix, "audex:");
        assert!(!config.tls);
    }

    #[test]
    fn test_ha_config_deserialize_redis() {
        let toml_str = r#"
backend = "redis"
endpoints = ["redis://redis-1:6379", "redis://redis-2:6379"]
password = "secret"
tls = true
prefix = "myapp:audex:"

[leader]
node_id = "server-1"
lease_ttl = 30
renewal_interval = 10

[rate_limit]
enabled = true
window_seconds = 1800

[audit_replication]
enabled = true
max_stream_length = 50000
"#;
        let config: HaConfig = toml::from_str(toml_str).unwrap();
        assert_eq!(config.backend, HaBackend::Redis);
        assert_eq!(config.endpoints.len(), 2);
        assert_eq!(config.password.as_deref(), Some("secret"));
        assert!(config.tls);
        assert_eq!(config.leader.lease_ttl, 30);
        assert_eq!(config.rate_limit.window_seconds, 1800);
        assert_eq!(config.audit_replication.max_stream_length, 50000);
    }

    #[test]
    fn test_ha_config_deserialize_etcd() {
        let toml_str = r#"
backend = "etcd"
endpoints = ["http://etcd-1:2379", "http://etcd-2:2379", "http://etcd-3:2379"]

[leader]
lease_ttl = 10
"#;
        let config: HaConfig = toml::from_str(toml_str).unwrap();
        assert_eq!(config.backend, HaBackend::Etcd);
        assert_eq!(config.endpoints.len(), 3);
        assert_eq!(config.leader.lease_ttl, 10);
    }

    #[test]
    fn test_ha_keys() {
        let keys = HaKeys::new("audex:");
        assert_eq!(keys.leader_lock(), "audex:leader:lock");
        assert_eq!(keys.leader_heartbeat(), "audex:leader:heartbeat");
        assert_eq!(
            keys.rate_limit("user@corp.com"),
            "audex:ratelimit:user@corp.com"
        );
        assert_eq!(keys.audit_stream(), "audex:audit:stream");
        assert_eq!(keys.session_registry(), "audex:sessions:active");
        assert_eq!(keys.node_registry(), "audex:nodes:active");
    }

    #[test]
    fn test_ha_keys_custom_prefix() {
        let keys = HaKeys::new("prod:audex:");
        assert_eq!(keys.leader_lock(), "prod:audex:leader:lock");
        assert_eq!(keys.rate_limit("alice"), "prod:audex:ratelimit:alice");
    }

    #[test]
    fn test_leader_election_commands() {
        let cmd = LeaderElectionCommands::try_acquire("audex:leader:lock", "node-1", 15);
        assert_eq!(cmd[0], "SET");
        assert_eq!(cmd[2], "node-1");
        assert_eq!(cmd[3], "NX");
        assert_eq!(cmd[4], "EX");
        assert_eq!(cmd[5], "15");
    }

    #[test]
    fn test_leader_election_lua_scripts() {
        let renew = LeaderElectionCommands::renew_lua();
        assert!(renew.contains("EXPIRE"));
        assert!(renew.contains("GET"));

        let release = LeaderElectionCommands::release_lua();
        assert!(release.contains("DEL"));
    }

    #[test]
    fn test_rate_limit_lua_script() {
        let lua = DistributedRateLimitCommands::check_and_record_lua();
        assert!(lua.contains("ZREMRANGEBYSCORE"));
        assert!(lua.contains("ZADD"));
        assert!(lua.contains("ZCARD"));
        assert!(lua.contains("EXPIRE"));
    }

    #[test]
    fn test_audit_replication_commands() {
        let fields = AuditReplicationCommands::publish_fields("{\"event\":\"test\"}");
        assert_eq!(fields[0].0, "data");

        let group = AuditReplicationCommands::create_group("audex:audit:stream", "node-1-group");
        assert_eq!(group[0], "XGROUP");
        assert_eq!(group[5], "MKSTREAM");

        let ack = AuditReplicationCommands::ack("stream", "group", &["1-1", "1-2"]);
        assert_eq!(ack[0], "XACK");
        assert_eq!(ack.len(), 5);
    }

    #[test]
    fn test_etcd_operations() {
        let body = EtcdOperations::lease_grant_body(15);
        assert!(body.contains("\"TTL\":15"));

        let txn = EtcdOperations::txn_create_body("mykey", "myval", 12345);
        assert!(txn.contains("request_put"));
        assert!(txn.contains("12345"));
    }

    #[test]
    fn test_base64_encode() {
        assert_eq!(base64_encode("hello"), "aGVsbG8=");
        assert_eq!(base64_encode("ab"), "YWI=");
        assert_eq!(base64_encode("abc"), "YWJj");
    }

    #[test]
    fn test_generate_node_id() {
        let id = generate_node_id();
        assert!(!id.is_empty());
        // Should contain a hyphen-separated structure
        assert!(id.contains('-'));
    }

    #[test]
    fn test_cluster_status_serialization() {
        let status = ClusterStatus {
            node_id: "node-1".to_string(),
            is_leader: true,
            leader_node: Some("node-1".to_string()),
            active_nodes: vec![NodeInfo {
                node_id: "node-1".to_string(),
                address: "10.0.0.1:8080".to_string(),
                sessions: 5,
                last_seen: "2026-01-01T00:00:00Z".to_string(),
            }],
            total_sessions: 5,
            backend: "redis".to_string(),
        };
        let json = serde_json::to_string(&status).unwrap();
        assert!(json.contains("node-1"));
        assert!(json.contains("is_leader"));
    }

    #[test]
    fn test_ha_backend_display() {
        assert_eq!(HaBackend::Redis.to_string(), "redis");
        assert_eq!(HaBackend::Etcd.to_string(), "etcd");
    }

    #[test]
    fn test_sanitize_component_strips_unsafe_chars() {
        // R6-M65: verify the sanitizer used by generate_node_id.
        assert_eq!(HaKeys::sanitize_component("host\nname"), "hostname");
        assert_eq!(HaKeys::sanitize_component("host\rname"), "hostname");
        assert_eq!(HaKeys::sanitize_component("host name"), "hostname");
        assert_eq!(HaKeys::sanitize_component("host\0name"), "hostname");
        assert_eq!(
            HaKeys::sanitize_component("good-host.local"),
            "good-host.local"
        );
        assert_eq!(HaKeys::sanitize_component(""), "");
    }

    #[test]
    fn test_leader_config_validate_overflow() {
        // R6-M66: renewal_interval near u64::MAX should fail, not wrap.
        let cfg = LeaderConfig {
            node_id: None,
            lease_ttl: 15,
            renewal_interval: u64::MAX / 2,
        };
        let err = cfg.validate().unwrap_err();
        let msg = format!("{}", err);
        assert!(
            msg.contains("overflows"),
            "expected overflow error, got: {}",
            msg
        );
    }

    #[test]
    fn test_leader_config_validate_normal() {
        // Normal valid config passes.
        let cfg = LeaderConfig {
            node_id: None,
            lease_ttl: 15,
            renewal_interval: 4,
        };
        assert!(cfg.validate().is_ok());

        // Boundary: renewal * 3 == lease_ttl should fail.
        let cfg = LeaderConfig {
            node_id: None,
            lease_ttl: 15,
            renewal_interval: 5,
        };
        assert!(cfg.validate().is_err());
    }
}