kcode_k1_canonical_chain/

lib.rs

use kcode_k1_order_store::OrderStore;
pub use kcode_k1_transaction::{GENESIS_PARENT, REGISTER_AT_TIP, SubsystemId};
use kcode_k1_transaction::{Transaction, build_signed_transaction};
pub use kcode_k1_transaction_store::TxId;
use kcode_k1_transaction_store::{PutOutcome, StoreError, TransactionStore};
use sha2::{Digest, Sha256};
use std::{cmp::Ordering, fmt, fs, path::Path};

pub struct CanonicalChain {
    order: OrderStore,
    store: TransactionStore,
}

pub struct ReplayCursor {
    subsystem: SubsystemId,
    after: Option<TxId>,
}

pub struct ReplayTransaction {
    pub id: TxId,
    pub bytes: Vec<u8>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CommitOutcome {
    Duplicate,
    Extension { id: TxId, subsystem: SubsystemId },
    Reorganization { id: TxId, subsystem: SubsystemId },
}

#[derive(Debug)]
pub enum SubmitError {
    MissingParent,
    Other(String),
}

impl fmt::Display for SubmitError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::MissingParent => formatter.write_str("missing parent"),
            Self::Other(message) => formatter.write_str(message),
        }
    }
}

impl std::error::Error for SubmitError {}

struct Candidate<'a> {
    id: TxId,
    parent: TxId,
    creator: [u8; 32],
    timestamp: u64,
    subsystem: SubsystemId,
    bytes: &'a [u8],
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ForkDecision {
    Incoming,
    Incumbent,
    Duplicate,
    Collision,
}

impl CanonicalChain {
    pub fn open(root: &Path) -> Result<Self, String> {
        prepare_root(root)?;

        let ordering_path = root.join("ordering.dat");
        let store_path = root.join("k1-transaction-store");
        let ordering_type = path_type(&ordering_path)?;
        let store_type = path_type(&store_path)?;

        if ordering_type.is_some_and(|kind| !kind.is_file()) {
            return Err("ordering.dat is not a regular file".to_owned());
        }
        if store_type.is_some_and(|kind| !kind.is_dir()) {
            return Err("k1-transaction-store is not a directory".to_owned());
        }

        let (order, store) = match (ordering_type, store_type) {
            (None, None) => {
                let store = TransactionStore::create(&store_path)
                    .unwrap_or_else(|error| fatal("create-transaction-store", error));
                let order = OrderStore::create(&ordering_path)
                    .unwrap_or_else(|error| fatal("create-order-store", error));
                (order, store)
            }
            (Some(_), Some(_)) => (
                OrderStore::open(&ordering_path)?,
                TransactionStore::open(&store_path)
                    .map_err(|error| format!("transaction store error: {error}"))?,
            ),
            _ => return Err("canonical chain root is incomplete".to_owned()),
        };

        Ok(Self { order, store })
    }

    pub fn submit_validated(&mut self, transaction: &[u8]) -> Result<CommitOutcome, SubmitError> {
        let parsed = Transaction::parse(transaction)
            .map_err(|message| SubmitError::Other(format!("invalid transaction: {message}")))?;
        let candidate = Candidate {
            id: TxId::for_transaction(transaction),
            parent: parsed.parent(),
            creator: *parsed.creator(),
            timestamp: parsed.timestamp(),
            subsystem: parsed.subsystem(),
            bytes: transaction,
        };

        if is_reserved_id(candidate.id) {
            return Err(SubmitError::Other(
                "transaction ID collides with a reserved sentinel".to_owned(),
            ));
        }

        self.submit_candidate(candidate)
    }

    pub fn submit_local<F>(
        &mut self,
        timestamp: u64,
        creator: [u8; 32],
        subsystem: SubsystemId,
        payload: &[u8],
        signer: F,
    ) -> Result<Vec<u8>, String>
    where
        F: FnOnce(&[u8]) -> Result<[u8; 64], String>,
    {
        let parent = self.tip().unwrap_or(GENESIS_PARENT);
        let bytes =
            build_signed_transaction(parent, timestamp, creator, subsystem, payload, signer)?;
        let id = TxId::for_transaction(&bytes);

        if is_reserved_id(id) {
            return Err("transaction ID collides with a reserved sentinel".to_owned());
        }
        if self.order.index_of(id).is_some() {
            return Err("transaction ID collides with a canonical transaction".to_owned());
        }

        self.persist(&bytes, id)
            .map_err(|error| error.to_string())?;
        self.order
            .commit(self.order.entries().len(), id, subsystem)
            .unwrap_or_else(|error| fatal("commit-local-order", error));
        Ok(bytes)
    }

    pub fn contains(&self, id: TxId) -> bool {
        !is_reserved_id(id) && self.order.index_of(id).is_some()
    }

    pub fn tip(&self) -> Option<TxId> {
        self.order.entries().last().map(|entry| entry.0)
    }

    pub fn between_txids(&self, older: TxId, newer: TxId) -> Result<Vec<TxId>, String> {
        let older_index = if older == GENESIS_PARENT {
            -1_i128
        } else {
            self.order
                .index_of(older)
                .map(|index| index as i128)
                .ok_or_else(|| "older boundary is not canonical".to_owned())?
        };
        let newer_index = self
            .order
            .index_of(newer)
            .map(|index| index as i128)
            .ok_or_else(|| "newer boundary is not canonical".to_owned())?;

        if older_index == newer_index {
            return Ok(Vec::new());
        }
        if older_index > newer_index {
            return Err("transaction boundaries are reversed".to_owned());
        }

        let interior = newer_index - older_index - 1;
        if interior <= 128 {
            return Ok(((older_index + 1)..newer_index)
                .map(|index| self.order.entries()[index as usize].0)
                .collect());
        }

        let distance = newer_index - older_index;
        Ok((1_i128..=128)
            .map(|k| self.order.entries()[(older_index + k * distance / 129) as usize].0)
            .collect())
    }

    pub fn get_txn(&self, id: TxId) -> Result<Option<Vec<u8>>, String> {
        if !self.contains(id) {
            return Ok(None);
        }
        Ok(Some(self.canonical_bytes(id)))
    }

    pub fn replay_cursor(
        &self,
        subsystem: SubsystemId,
        after: Option<TxId>,
    ) -> Result<ReplayCursor, String> {
        if let Some(id) = after {
            let index = self
                .order
                .index_of(id)
                .ok_or_else(|| "replay checkpoint is not canonical".to_owned())?;
            if self.order.entries()[index].1 != subsystem {
                return Err("replay checkpoint belongs to another subsystem".to_owned());
            }
        }

        Ok(ReplayCursor { subsystem, after })
    }

    pub fn replay_next(
        &self,
        cursor: &mut ReplayCursor,
    ) -> Result<Option<ReplayTransaction>, String> {
        let start = match cursor.after {
            None => 0,
            Some(id) => {
                self.order
                    .index_of(id)
                    .ok_or_else(|| "replay cursor is no longer canonical".to_owned())?
                    + 1
            }
        };

        for &(id, subsystem) in &self.order.entries()[start..] {
            if subsystem != cursor.subsystem {
                continue;
            }

            let bytes = self.canonical_bytes(id);
            let parsed = Transaction::parse(&bytes)
                .unwrap_or_else(|error| fatal("parse-canonical-transaction", error));
            if TxId::for_transaction(&bytes) != id || parsed.subsystem() != subsystem {
                fatal(
                    "verify-canonical-transaction",
                    "canonical transaction does not match its order record",
                );
            }

            cursor.after = Some(id);
            return Ok(Some(ReplayTransaction { id, bytes }));
        }

        Ok(None)
    }

    fn submit_candidate(&mut self, candidate: Candidate<'_>) -> Result<CommitOutcome, SubmitError> {
        if self.order.index_of(candidate.id).is_some() {
            return if self.canonical_bytes(candidate.id) == candidate.bytes {
                Ok(CommitOutcome::Duplicate)
            } else {
                Err(SubmitError::Other(
                    "transaction ID collision with canonical bytes".to_owned(),
                ))
            };
        }

        let shared_len = if candidate.parent == GENESIS_PARENT {
            0
        } else {
            self.order
                .index_of(candidate.parent)
                .map(|index| index + 1)
                .ok_or(SubmitError::MissingParent)?
        };

        if shared_len == self.order.entries().len() {
            self.persist(candidate.bytes, candidate.id)?;
            self.order
                .commit(shared_len, candidate.id, candidate.subsystem)
                .unwrap_or_else(|error| fatal("commit-extension-order", error));
            return Ok(CommitOutcome::Extension {
                id: candidate.id,
                subsystem: candidate.subsystem,
            });
        }

        let (incumbent_id, incumbent_subsystem) = self.order.entries()[shared_len];
        let incumbent_bytes = self.canonical_bytes(incumbent_id);
        let incumbent = Transaction::parse(&incumbent_bytes)
            .unwrap_or_else(|error| fatal("parse-canonical-incumbent", error));
        if incumbent.subsystem() != incumbent_subsystem || incumbent.parent() != candidate.parent {
            fatal(
                "verify-canonical-incumbent",
                "canonical incumbent does not match its order record or parent",
            );
        }

        match fork_decision(
            &candidate.creator,
            candidate.timestamp,
            candidate.bytes,
            incumbent.creator(),
            incumbent.timestamp(),
            &incumbent_bytes,
        ) {
            ForkDecision::Incumbent => {
                return Err(SubmitError::Other(
                    "fork loses canonical ordering".to_owned(),
                ));
            }
            ForkDecision::Duplicate => return Ok(CommitOutcome::Duplicate),
            ForkDecision::Collision => {
                return Err(SubmitError::Other(
                    "full transaction digest collision".to_owned(),
                ));
            }
            ForkDecision::Incoming => {}
        }

        self.persist(candidate.bytes, candidate.id)?;
        self.order
            .commit(shared_len, candidate.id, candidate.subsystem)
            .unwrap_or_else(|error| fatal("commit-reorganization-order", error));
        Ok(CommitOutcome::Reorganization {
            id: candidate.id,
            subsystem: candidate.subsystem,
        })
    }

    fn persist(&self, bytes: &[u8], expected: TxId) -> Result<(), SubmitError> {
        match self.store.put(bytes) {
            Ok(PutOutcome::Inserted(id)) | Ok(PutOutcome::Duplicate(id)) if id == expected => {
                Ok(())
            }
            Ok(_) => fatal(
                "persist-transaction",
                "transaction store returned an unexpected transaction ID",
            ),
            Err(StoreError::IdCollision(_)) => Err(SubmitError::Other(
                "transaction ID collision with stored bytes".to_owned(),
            )),
            Err(error) => fatal("persist-transaction", error),
        }
    }

    fn canonical_bytes(&self, id: TxId) -> Vec<u8> {
        match self.store.get(id) {
            Ok(Some(bytes)) => bytes,
            Ok(None) => fatal(
                "load-canonical-transaction",
                "canonical transaction bytes are missing",
            ),
            Err(error) => fatal("load-canonical-transaction", error),
        }
    }
}

fn is_reserved_id(id: TxId) -> bool {
    id == GENESIS_PARENT || id == REGISTER_AT_TIP
}

fn prepare_root(root: &Path) -> Result<(), String> {
    match fs::symlink_metadata(root) {
        Ok(metadata) if metadata.file_type().is_dir() => Ok(()),
        Ok(_) => Err("canonical chain root is not a directory".to_owned()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            fs::create_dir_all(root).unwrap_or_else(|error| fatal("create-root", error));
            Ok(())
        }
        Err(error) => Err(format!("cannot inspect canonical chain root: {error}")),
    }
}

fn path_type(path: &Path) -> Result<Option<fs::FileType>, String> {
    match fs::symlink_metadata(path) {
        Ok(metadata) => Ok(Some(metadata.file_type())),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(format!("cannot inspect canonical chain component: {error}")),
    }
}

fn fork_decision(
    incoming_creator: &[u8; 32],
    incoming_timestamp: u64,
    incoming_bytes: &[u8],
    incumbent_creator: &[u8; 32],
    incumbent_timestamp: u64,
    incumbent_bytes: &[u8],
) -> ForkDecision {
    match incoming_creator.cmp(incumbent_creator) {
        Ordering::Less => return ForkDecision::Incoming,
        Ordering::Greater => return ForkDecision::Incumbent,
        Ordering::Equal => {}
    }
    match incoming_timestamp.cmp(&incumbent_timestamp) {
        Ordering::Less => return ForkDecision::Incoming,
        Ordering::Greater => return ForkDecision::Incumbent,
        Ordering::Equal => {}
    }

    let incoming_digest: [u8; 32] = Sha256::digest(incoming_bytes).into();
    let incumbent_digest: [u8; 32] = Sha256::digest(incumbent_bytes).into();
    match incoming_digest.cmp(&incumbent_digest) {
        Ordering::Less => ForkDecision::Incoming,
        Ordering::Greater => ForkDecision::Incumbent,
        Ordering::Equal if incoming_bytes == incumbent_bytes => ForkDecision::Duplicate,
        Ordering::Equal => ForkDecision::Collision,
    }
}

fn fatal(operation: &str, error: impl fmt::Display) -> ! {
    eprintln!("kcode-k1-canonical-chain fatal {operation}: {error}");
    std::process::abort()
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::cell::Cell;
    use std::path::PathBuf;
    use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
    use std::time::{Duration, Instant};

    static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);

    struct TempRoot(PathBuf);

    impl TempRoot {
        fn new(label: &str) -> Self {
            let number = NEXT_ROOT.fetch_add(1, AtomicOrdering::Relaxed);
            let path = std::env::temp_dir().join(format!(
                "kcode-k1-canonical-chain-{}-{number}-{label}",
                std::process::id()
            ));
            let _ = fs::remove_dir_all(&path);
            Self(path)
        }
    }

    impl Drop for TempRoot {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.0);
        }
    }

    fn subsystem(value: u8) -> SubsystemId {
        SubsystemId::from_bytes([value; 20]).unwrap()
    }

    fn transaction(
        parent: TxId,
        creator: u8,
        timestamp: u64,
        subsystem: SubsystemId,
        payload: &[u8],
    ) -> Vec<u8> {
        build_signed_transaction(parent, timestamp, [creator; 32], subsystem, payload, |_| {
            Ok([creator; 64])
        })
        .unwrap()
    }

    #[test]
    fn local_submission_parents_and_signer_error_does_not_mutate() {
        let root = TempRoot::new("local");
        let mut chain = CanonicalChain::open(&root.0).unwrap();
        let calls = Cell::new(0);
        let error = chain.submit_local(0, [0; 32], subsystem(b'a'), b"bad", |_| {
            calls.set(calls.get() + 1);
            Err("signer stopped".to_owned())
        });
        assert_eq!(error.unwrap_err(), "signer stopped");
        assert_eq!((calls.get(), chain.tip()), (1, None));
        assert_eq!(fs::metadata(root.0.join("ordering.dat")).unwrap().len(), 0);

        let first = chain
            .submit_local(1, [1; 32], subsystem(b'a'), b"first", |_| Ok([2; 64]))
            .unwrap();
        let second = chain
            .submit_local(2, [1; 32], subsystem(b'b'), b"second", |_| Ok([3; 64]))
            .unwrap();
        let first_id = TxId::for_transaction(&first);
        assert_eq!(Transaction::parse(&first).unwrap().parent(), GENESIS_PARENT);
        assert_eq!(Transaction::parse(&second).unwrap().parent(), first_id);
        assert_eq!(chain.get_txn(first_id).unwrap(), Some(first));

        drop(chain);
        assert_eq!(
            CanonicalChain::open(&root.0).unwrap().tip(),
            Some(TxId::for_transaction(&second))
        );
    }

    #[test]
    fn remote_reorganization_reopens_and_retains_orphans() {
        let root = TempRoot::new("remote");
        let mut chain = CanonicalChain::open(&root.0).unwrap();
        let owner = subsystem(b'a');
        let first = transaction(GENESIS_PARENT, 20, 1, owner, b"first");
        let first_id = TxId::for_transaction(&first);
        assert!(matches!(
            chain.submit_validated(&first),
            Ok(CommitOutcome::Extension { .. })
        ));
        assert_eq!(
            chain.submit_validated(&first).unwrap(),
            CommitOutcome::Duplicate
        );

        let incumbent = transaction(first_id, 50, 2, owner, b"incumbent");
        let incumbent_id = TxId::for_transaction(&incumbent);
        chain.submit_validated(&incumbent).unwrap();
        let descendant = transaction(incumbent_id, 50, 3, owner, b"descendant");
        let descendant_id = TxId::for_transaction(&descendant);
        chain.submit_validated(&descendant).unwrap();

        let replacement = transaction(first_id, 1, 99, owner, b"replacement");
        let replacement_id = TxId::for_transaction(&replacement);
        assert!(matches!(
            chain.submit_validated(&replacement),
            Ok(CommitOutcome::Reorganization { .. })
        ));
        let loser = transaction(first_id, 250, 0, owner, b"loser");
        let loser_id = TxId::for_transaction(&loser);
        assert!(chain.submit_validated(&loser).is_err());
        let removed_child = transaction(descendant_id, 0, 4, owner, b"removed-parent");
        let removed_child_id = TxId::for_transaction(&removed_child);
        assert!(matches!(
            chain.submit_validated(&removed_child),
            Err(SubmitError::MissingParent)
        ));

        drop(chain);
        let reopened = CanonicalChain::open(&root.0).unwrap();
        assert_eq!(reopened.tip(), Some(replacement_id));
        assert!(!reopened.contains(incumbent_id));
        assert!(!reopened.contains(descendant_id));
        drop(reopened);

        let store = TransactionStore::open(&root.0.join("k1-transaction-store")).unwrap();
        assert!(store.contains(incumbent_id));
        assert!(store.contains(descendant_id));
        assert!(!store.contains(loser_id));
        assert!(!store.contains(removed_child_id));
    }

    #[test]
    fn fork_ranking_uses_timestamp_then_complete_digest() {
        let creator = [1; 32];
        assert_eq!(
            fork_decision(&creator, 1, b"x", &creator, 2, b"y"),
            ForkDecision::Incoming
        );
        let left: [u8; 32] = Sha256::digest(b"left").into();
        let right: [u8; 32] = Sha256::digest(b"right").into();
        assert_eq!(
            fork_decision(&creator, 1, b"left", &creator, 1, b"right"),
            if left < right {
                ForkDecision::Incoming
            } else {
                ForkDecision::Incumbent
            }
        );
    }

    #[test]
    fn queries_replay_and_removed_cursor_behave_canonically() {
        let root = TempRoot::new("queries");
        let mut chain = CanonicalChain::open(&root.0).unwrap();
        let (a, b) = (subsystem(b'a'), subsystem(b'b'));
        let mut parent = GENESIS_PARENT;
        let mut ids = Vec::new();

        for index in 0..140_u64 {
            let owner = if index % 2 == 0 { a } else { b };
            let bytes = transaction(parent, 1, index, owner, &[index as u8]);
            parent = TxId::for_transaction(&bytes);
            ids.push(parent);
            chain.submit_validated(&bytes).unwrap();
        }

        assert_eq!(chain.between_txids(ids[3], ids[10]).unwrap(), ids[4..10]);
        assert_eq!(
            chain
                .between_txids(GENESIS_PARENT, *ids.last().unwrap())
                .unwrap()
                .len(),
            128
        );
        assert!(chain.between_txids(ids[10], ids[3]).is_err());

        let mut cursor = chain.replay_cursor(a, Some(ids[0])).unwrap();
        assert_eq!(chain.replay_next(&mut cursor).unwrap().unwrap().id, ids[2]);
        let replacement = transaction(ids[0], 0, 999, b, b"replacement");
        chain.submit_validated(&replacement).unwrap();
        assert!(chain.replay_next(&mut cursor).is_err());
    }

    #[test]
    fn registration_sentinels_are_reserved() {
        let root = TempRoot::new("sentinels");
        let chain = CanonicalChain::open(&root.0).unwrap();

        assert_eq!(GENESIS_PARENT.into_bytes(), [0xff; 12]);
        assert_eq!(
            REGISTER_AT_TIP.into_bytes(),
            [
                0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xfe
            ]
        );
        assert!(is_reserved_id(GENESIS_PARENT));
        assert!(is_reserved_id(REGISTER_AT_TIP));
        assert!(!is_reserved_id(TxId::from_bytes([0; 12])));
        assert!(!chain.contains(GENESIS_PARENT));
        assert!(!chain.contains(REGISTER_AT_TIP));
        assert!(
            chain
                .replay_cursor(subsystem(b'a'), Some(REGISTER_AT_TIP))
                .is_err()
        );
        assert!(
            chain
                .between_txids(REGISTER_AT_TIP, GENESIS_PARENT)
                .is_err()
        );
    }

    #[test]
    fn root_shapes_remain_compatible() {
        let empty = TempRoot::new("empty");
        fs::create_dir_all(&empty.0).unwrap();
        fs::write(empty.0.join("extra"), []).unwrap();
        drop(CanonicalChain::open(&empty.0).unwrap());

        let mixed = TempRoot::new("mixed");
        fs::create_dir_all(&mixed.0).unwrap();
        fs::write(mixed.0.join("ordering.dat"), []).unwrap();
        assert!(CanonicalChain::open(&mixed.0).is_err());

        let malformed = TempRoot::new("malformed");
        drop(CanonicalChain::open(&malformed.0).unwrap());
        fs::write(malformed.0.join("ordering.dat"), [0; 31]).unwrap();
        assert!(CanonicalChain::open(&malformed.0).is_err());
    }

    #[test]
    fn opens_million_record_fixture_under_five_seconds() {
        let root = TempRoot::new("million");
        drop(CanonicalChain::open(&root.0).unwrap());
        let count = 1_000_000_u64;
        let owner = subsystem(b'm');
        let mut bytes = Vec::with_capacity(count as usize * 32);

        for value in 0..count {
            let mut id = [0_u8; 12];
            id[..8].copy_from_slice(&value.to_le_bytes());
            bytes.extend_from_slice(&id);
            bytes.extend_from_slice(owner.as_bytes());
        }
        fs::write(root.0.join("ordering.dat"), bytes).unwrap();

        let started = Instant::now();
        let chain = CanonicalChain::open(&root.0).unwrap();
        assert!(started.elapsed() < Duration::from_secs(5));
        assert_eq!(chain.order.entries().len(), count as usize);
        let mut last = [0_u8; 12];
        last[..8].copy_from_slice(&(count - 1).to_le_bytes());
        assert!(chain.contains(TxId::from_bytes(last)));
    }
}