polyc-eventlog-host 2026.9.0

//! The reversible name one logical partition takes on durable storage.
//!
//! The storage backend accepts a narrow alphabet in a partition name. A
//! conversation id is namespaced as `{namespace}:{id}`, so it does not fit that
//! alphabet. The codec here escapes a logical name into the alphabet, and it
//! recovers the logical name again.
//!
//! # Why the codec must be injective
//!
//! The previous mapping replaced every illegal byte with `_`. That map is
//! many-to-one, so `a:b` and `a_b` reached one partition and shared one
//! history in the source of truth (#1748). The journal is the durable
//! authority, and two conversations must never share one. This codec is
//! reversible, so distinct logical names always take distinct physical names.
//!
//! # The alphabet
//!
//! The physical alphabet is `[a-z0-9_-]`. It holds no uppercase letter on
//! purpose: a case-insensitive volume treats `A` and `a` as one directory, so
//! passing uppercase through would not be injective everywhere this runs.
//!
//! - `a-z`, `0-9` and `-` pass through unchanged.
//! - `_` becomes `__`.
//! - Every other byte becomes `_` plus two lowercase hex digits.
//!
//! ```text
//! conv-web:cafe  ->  conv-web_3acafe
//! conv-web_cafe  ->  conv-web__cafe
//! conv-A         ->  conv-_41
//! ```
//!
//! Worst-case growth is three times the logical length.
//!
//! # Canonical form
//!
//! One logical name has exactly one physical spelling. [`decode`] proves that:
//! it decodes, encodes the result again, and requires the two to match byte for
//! byte. So `_41` decodes, but `_61` does not — `a` encodes as `a`. A physical
//! name that fails this test is corruption, not an older spelling. This
//! deployment carries no legacy names.

/// The longest suffix the storage backend appends to a partition name.
///
/// The backend lays one partition down as several sibling directory
/// components: `{name}_data`, `{name}_offsets-metadata`,
/// `{name}_offsets-blobs`, and `{name}__eventcount_checkpoint`. The last is the
/// longest, so it sets how much of a directory name the codec may spend.
const LONGEST_BACKEND_SUFFIX: usize = "__eventcount_checkpoint".len();

/// The longest single directory component the supported filesystems hold.
const MAX_NAME_BYTES: usize = 255;

/// The longest encoded partition name that storage can hold.
///
/// The adapter enforces this before it changes the filesystem, so an
/// over-long name is a typed refusal rather than a failed directory create
/// part-way through a commit. Callers bound the LOGICAL id separately, at
/// ingress, which is what gives a caller a useful error.
pub const MAX_ENCODED_PARTITION_BYTES: usize = MAX_NAME_BYTES - LONGEST_BACKEND_SUFFIX;

/// The lowercase hex digits an escape uses.
const HEX_DIGITS: &[u8; 16] = b"0123456789abcdef";

/// Why a partition name is not usable.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum PartitionNameError {
    /// The logical name held no bytes.
    #[error("a partition name holds at least one byte")]
    Empty,
    /// The encoded name is longer than storage holds.
    #[error("the encoded partition name takes {actual} bytes, and storage holds {limit}")]
    TooLong {
        /// Length of the encoded name.
        actual: usize,
        /// The most this adapter accepts.
        limit: usize,
    },
    /// The logical name reaches an internal namespace.
    #[error("`{0}` names an internal namespace that a caller does not address")]
    Reserved(String),
    /// The physical name is not the canonical encoding of any logical name.
    #[error("the physical name `{0}` is not a canonical encoding")]
    NotCanonical(String),
}

/// One partition name in its physical, storage-facing form.
///
/// The type exists so that an encoded name cannot be encoded twice. The codec
/// is deliberately not idempotent: `:` encodes to `_3a`, and `_3a` encodes to
/// `__3a`. A double encode would therefore address a partition that no writer
/// ever wrote to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EncodedPartitionName(String);

impl EncodedPartitionName {
    /// Returns the physical name, for the storage call that needs it.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    /// Consumes this name and returns the physical string it holds.
    #[must_use]
    pub fn into_string(self) -> String {
        self.0
    }
}

impl std::fmt::Display for EncodedPartitionName {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// One partition, in both the names this host needs for it.
///
/// A command handler needs both at once. Storage addresses the encoded name,
/// and an observer notification reports the logical name — a projector keys its
/// own state on the id it was given, never on a storage spelling. Carrying both
/// in one value is what keeps a handler from reaching for the wrong one, and
/// what makes a second encode of an encoded name unrepresentable.
///
/// Built once, at this host's public API, so the codec runs once per command.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PartitionRef {
    /// The name every caller, observer, and projection knows the partition by.
    logical: String,
    /// The name storage holds it under.
    encoded: EncodedPartitionName,
}

impl PartitionRef {
    /// Builds both names for `logical`.
    ///
    /// # Errors
    ///
    /// As [`encode`].
    pub fn new(logical: &str) -> Result<Self, PartitionNameError> {
        Ok(Self {
            logical: logical.to_owned(),
            encoded: encode(logical)?,
        })
    }

    /// The name callers and observers know this partition by.
    #[must_use]
    pub fn logical(&self) -> &str {
        &self.logical
    }

    /// The encoded name as a string, for a storage call that takes `&str`.
    #[must_use]
    pub fn storage_key(&self) -> &str {
        self.encoded.as_str()
    }
}

impl std::fmt::Display for PartitionRef {
    /// Shows the logical name. Logs and errors name the partition a person
    /// asked for, not the spelling storage chose for it.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.logical)
    }
}

/// Encodes `logical` into the physical name storage holds it under.
///
/// # Errors
///
/// Returns [`PartitionNameError::Empty`] for an empty name,
/// [`PartitionNameError::Reserved`] for a name in an internal namespace, and
/// [`PartitionNameError::TooLong`] when the encoded form is longer than
/// [`MAX_ENCODED_PARTITION_BYTES`].
pub fn encode(logical: &str) -> Result<EncodedPartitionName, PartitionNameError> {
    if logical.starts_with(crate::REPAIR_STAGE_PREFIX)
        || logical.starts_with(crate::REWRITE_STAGE_PREFIX)
    {
        return Err(PartitionNameError::Reserved(logical.to_owned()));
    }
    encode_any_namespace(logical)
}

/// Encodes `logical` without the reserved-namespace refusal.
///
/// Two callers need the codec without the reserved-namespace refusal:
/// [`decode`]'s canonical-form proof, which re-encodes a name it has just
/// decoded, and the duplicate report in partition discovery. A caller-supplied
/// name never reaches this function; [`encode`] is the boundary that refuses
/// one. Repair stages are minted already canonical and used directly as a
/// storage key, so they do not pass through here at all.
///
/// # Errors
///
/// As [`encode`], less [`PartitionNameError::Reserved`].
pub(crate) fn encode_any_namespace(
    logical: &str,
) -> Result<EncodedPartitionName, PartitionNameError> {
    if logical.is_empty() {
        return Err(PartitionNameError::Empty);
    }
    let mut encoded = String::with_capacity(logical.len());
    for byte in logical.bytes() {
        match byte {
            b'a'..=b'z' | b'0'..=b'9' | b'-' => encoded.push(char::from(byte)),
            b'_' => encoded.push_str("__"),
            other => {
                encoded.push('_');
                encoded.push(char::from(HEX_DIGITS[usize::from(other >> 4)]));
                encoded.push(char::from(HEX_DIGITS[usize::from(other & 0x0f)]));
            }
        }
    }
    if encoded.len() > MAX_ENCODED_PARTITION_BYTES {
        return Err(PartitionNameError::TooLong {
            actual: encoded.len(),
            limit: MAX_ENCODED_PARTITION_BYTES,
        });
    }
    Ok(EncodedPartitionName(encoded))
}

/// Recovers the logical name `physical` encodes.
///
/// # Errors
///
/// Returns [`PartitionNameError::NotCanonical`] when `physical` is not the
/// exact encoding of one logical name. A truncated escape, an uppercase hex
/// digit, an escape of a byte that needs none, and bytes that do not form UTF-8
/// all fail this way.
pub fn decode(physical: &str) -> Result<String, PartitionNameError> {
    let not_canonical = || PartitionNameError::NotCanonical(physical.to_owned());
    if physical.is_empty() {
        return Err(PartitionNameError::Empty);
    }

    let bytes = physical.as_bytes();
    let mut logical = Vec::with_capacity(bytes.len());
    let mut index = 0;
    while index < bytes.len() {
        match bytes[index] {
            b'_' => {
                let next = *bytes.get(index + 1).ok_or_else(not_canonical)?;
                if next == b'_' {
                    logical.push(b'_');
                    index += 2;
                } else {
                    let low = *bytes.get(index + 2).ok_or_else(not_canonical)?;
                    let high = hex_value(next).ok_or_else(not_canonical)?;
                    let low = hex_value(low).ok_or_else(not_canonical)?;
                    logical.push((high << 4) | low);
                    index += 3;
                }
            }
            byte => {
                logical.push(byte);
                index += 1;
            }
        }
    }

    let logical = String::from_utf8(logical).map_err(|_| not_canonical())?;

    // The canonical-form proof. Every physical name has exactly one logical
    // name, and every logical name exactly one physical name. Re-encoding is
    // what rejects a spelling that decodes but that this codec would never
    // have written, such as `_61` for `a`.
    let round_trip = encode_any_namespace(&logical).map_err(|_| not_canonical())?;
    if round_trip.as_str() != physical {
        return Err(not_canonical());
    }
    Ok(logical)
}

/// The value of one lowercase hex digit, or [`None`] for any other byte.
///
/// Uppercase is not a hex digit here. An escape carries one spelling, and the
/// canonical check in [`decode`] would refuse `_3A` in any case; refusing it
/// at the digit keeps the reason precise.
const fn hex_value(byte: u8) -> Option<u8> {
    match byte {
        b'0'..=b'9' => Some(byte - b'0'),
        b'a'..=b'f' => Some(byte - b'a' + 10),
        _ => None,
    }
}

#[cfg(test)]
mod tests {
    use super::{MAX_ENCODED_PARTITION_BYTES, PartitionNameError, decode, encode};

    /// Encoding then decoding returns the name that went in.
    ///
    /// The cases walk the alphabet's edges: the bytes that pass through, the
    /// one that doubles, the namespace separator this defect was found on, and
    /// a multi-byte character.
    #[test]
    fn every_logical_name_survives_the_round_trip() {
        for logical in [
            "conv-web:cafe",
            "conv-web_cafe",
            "conv-a",
            "conv-A",
            "conv-app:persona-1:standup",
            "conv-proj/alpha",
            "conv-a..b",
            "conv-café",
            "conv-_",
            "conv-__",
            "mem-persona-1",
            "audit-immutable",
        ] {
            let encoded = encode(logical).expect("the name encodes");
            assert_eq!(
                decode(encoded.as_str()).expect("the name decodes"),
                logical,
                "round trip for {logical}"
            );
        }
    }

    /// The pair that shared one history now takes two names (#1748).
    #[test]
    fn a_colon_and_an_underscore_take_different_names() {
        let colon = encode("conv-web:cafe").expect("encodes");
        let underscore = encode("conv-web_cafe").expect("encodes");
        assert_eq!(colon.as_str(), "conv-web_3acafe");
        assert_eq!(underscore.as_str(), "conv-web__cafe");
        assert_ne!(colon, underscore);
    }

    /// The physical alphabet holds no uppercase letter.
    ///
    /// A case-insensitive volume folds `A` onto `a`, so an encoding that
    /// passed uppercase through would not be injective on every filesystem
    /// this runs on.
    #[test]
    fn the_physical_alphabet_is_lowercase() {
        let encoded = encode("conv-Alpha").expect("encodes");
        assert_eq!(encoded.as_str(), "conv-_41lpha");
        assert!(
            !encoded.as_str().bytes().any(|b| b.is_ascii_uppercase()),
            "no uppercase reaches storage: {encoded}"
        );
    }

    /// The prefix every conversation partition carries survives encoding.
    ///
    /// Readers strip `conv-` from the LOGICAL name the listing returns, so
    /// this is a readability property and not what protects them. It is worth
    /// pinning anyway: a physical listing is what a person reads during an
    /// incident.
    #[test]
    fn the_conversation_prefix_stays_readable() {
        assert!(
            encode("conv-web:cafe")
                .expect("encodes")
                .as_str()
                .starts_with("conv-")
        );
    }

    /// A spelling this codec would never have written is corruption.
    #[test]
    fn a_noncanonical_spelling_is_refused() {
        for physical in [
            "conv-_61",   // `a`, escaped though it needs no escape
            "conv-_3A",   // uppercase hex digit
            "conv-_3",    // truncated escape
            "conv-_",     // an escape with nothing after it
            "conv-web:x", // a byte the alphabet does not hold
            "conv-_2d",   // `-`, escaped though it passes through
        ] {
            assert!(
                matches!(
                    decode(physical),
                    Err(PartitionNameError::NotCanonical(_) | PartitionNameError::Empty)
                ),
                "{physical} is not a canonical name"
            );
        }
    }

    /// Bytes that do not form UTF-8 do not name a partition.
    #[test]
    fn invalid_utf8_is_refused() {
        assert!(matches!(
            decode("conv-_ff"),
            Err(PartitionNameError::NotCanonical(_))
        ));
    }

    /// The adapter refuses a name longer than storage holds, and it accepts
    /// one exactly at the limit.
    #[test]
    fn the_length_limit_binds_at_the_adapter() {
        let at_limit = "a".repeat(MAX_ENCODED_PARTITION_BYTES);
        assert_eq!(
            encode(&at_limit)
                .expect("a name at the limit encodes")
                .as_str()
                .len(),
            MAX_ENCODED_PARTITION_BYTES
        );

        let over = "a".repeat(MAX_ENCODED_PARTITION_BYTES + 1);
        assert!(matches!(
            encode(&over),
            Err(PartitionNameError::TooLong { .. })
        ));

        // Escaping grows a name, so a short logical name can still cross the
        // limit. Three bytes per escape is the worst case.
        let escaped = ":".repeat(MAX_ENCODED_PARTITION_BYTES / 3 + 1);
        assert!(matches!(
            encode(&escaped),
            Err(PartitionNameError::TooLong { .. })
        ));
    }

    /// A caller cannot address the internal repair-stage namespace.
    ///
    /// The old sanitizer passed this prefix through unchanged, and partition
    /// discovery hides it. So a caller-supplied id could mint a partition that
    /// the fleet listing, retention, and both projections could not see.
    #[test]
    fn the_internal_namespace_is_refused_at_the_boundary() {
        assert!(matches!(
            encode("state-repair-stage-beef"),
            Err(PartitionNameError::Reserved(_))
        ));
    }

    /// An empty name addresses nothing.
    #[test]
    fn an_empty_name_is_refused() {
        assert!(matches!(encode(""), Err(PartitionNameError::Empty)));
        assert!(matches!(decode(""), Err(PartitionNameError::Empty)));
    }
}