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)));
}
}