use kcode_k1_transaction::Transaction;
pub use kcode_k1_transaction::{GENESIS_PARENT, SubsystemId};
pub use kcode_k1_transaction_store::TxId;
use kcode_k1_transaction_store::{PutOutcome, StoreError, TransactionStore};
use sha2::{Digest, Sha256};
use std::cmp::Ordering;
use std::collections::HashMap;
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::Path;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::{Duration, Instant};
type Order = Vec<(TxId, SubsystemId)>;
type Indexes = HashMap<TxId, usize>;
pub trait Subsystem: Send + Sync + 'static {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String>;
fn reorg(&self) -> Result<(), String>;
}
#[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 {}
pub struct K1TxnOrdering {
state: Mutex<State>,
store: TransactionStore,
}
struct State {
file: File,
order: Order,
indexes: Indexes,
subscribers: HashMap<SubsystemId, SubscriberState>,
reopen_required: bool,
}
struct SubscriberState {
handler: Arc<dyn Subsystem>,
latest: Option<usize>,
in_commission: bool,
}
struct Candidate<'a> {
id: TxId,
parent: TxId,
creator: [u8; 32],
timestamp: u64,
subsystem: SubsystemId,
payload: &'a [u8],
bytes: &'a [u8],
}
#[derive(Debug, Eq, PartialEq)]
enum ForkDecision {
Incoming,
Incumbent,
Duplicate,
Collision,
}
impl K1TxnOrdering {
pub fn open(root: &Path) -> Result<Self, String> {
let started = Instant::now();
let result = Self::open_inner(root);
let elapsed = started.elapsed();
if elapsed > Duration::from_millis(100) {
let outcome = if result.is_ok() { "ready" } else { "error" };
eprintln!(
"{{\"module\":\"kcode-k1-txn-ordering\",\"operation\":\"open\",\"elapsed_microseconds\":{},\"outcome\":\"{}\"}}",
elapsed.as_micros(),
outcome
);
}
result
}
pub fn register_subsystem(
&self,
subsystem: SubsystemId,
after: Option<TxId>,
handler: Arc<dyn Subsystem>,
) -> Result<(), String> {
let mut state = self.lock_state();
if state.reopen_required {
return Err(reopen_required_message());
}
if state
.subscribers
.get(&subsystem)
.is_some_and(|subscriber| subscriber.in_commission)
{
return Err("subsystem is already registered and active".to_owned());
}
let (start, latest) = match after {
None => (0, None),
Some(id) => {
let index = state
.indexes
.get(&id)
.copied()
.ok_or_else(|| "registration checkpoint is not canonical".to_owned())?;
if state.order[index].1 != subsystem {
return Err("registration checkpoint belongs to another subsystem".to_owned());
}
(index + 1, Some(index))
}
};
let mut subscriber = SubscriberState {
handler,
latest,
in_commission: false,
};
for index in start..state.order.len() {
let (id, entry_subsystem) = state.order[index];
if entry_subsystem != subsystem {
continue;
}
let bytes = match self.store_get(&mut state, id) {
Ok(Some(bytes)) => bytes,
Ok(None) => {
state.subscribers.insert(subsystem, subscriber);
return Err("canonical transaction bytes are missing".to_owned());
}
Err(message) => {
state.subscribers.insert(subsystem, subscriber);
return Err(message);
}
};
let transaction = match Transaction::parse(&bytes) {
Ok(transaction) => transaction,
Err(message) => {
state.subscribers.insert(subsystem, subscriber);
return Err(format!("canonical transaction is corrupt: {message}"));
}
};
if transaction.subsystem() != subsystem {
state.subscribers.insert(subsystem, subscriber);
return Err("canonical transaction has the wrong subsystem".to_owned());
}
if let Err(message) = subscriber.handler.submit_txn(id, transaction.payload()) {
state.subscribers.insert(subsystem, subscriber);
return Err(format!("subsystem replay callback failed: {message}"));
}
subscriber.latest = Some(index);
}
subscriber.in_commission = true;
state.subscribers.insert(subsystem, subscriber);
Ok(())
}
pub fn submit_txn(&self, transaction: &[u8]) -> Result<(), 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(),
payload: parsed.payload(),
bytes: transaction,
};
if candidate.id == GENESIS_PARENT {
return Err(SubmitError::Other(
"transaction ID collides with the genesis sentinel".to_owned(),
));
}
let mut state = self.lock_state();
if state.reopen_required {
return Err(SubmitError::Other(reopen_required_message()));
}
self.submit_candidate(&mut state, candidate)
}
pub fn contains(&self, id: TxId) -> bool {
id != GENESIS_PARENT && self.lock_state().indexes.contains_key(&id)
}
pub fn tip(&self) -> Option<TxId> {
self.lock_state().order.last().map(|entry| entry.0)
}
pub fn between_txids(&self, older: TxId, newer: TxId) -> Result<Vec<TxId>, String> {
let state = self.lock_state();
let older_index = if older == GENESIS_PARENT {
-1_i128
} else {
state
.indexes
.get(&older)
.map(|index| *index as i128)
.ok_or_else(|| "older boundary is not canonical".to_owned())?
};
let newer_index = state
.indexes
.get(&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| state.order[index as usize].0)
.collect());
}
let distance = newer_index - older_index;
Ok((1_i128..=128)
.map(|k| {
let index = older_index + k * distance / 129;
state.order[index as usize].0
})
.collect())
}
pub fn get_txn(&self, id: TxId) -> Result<Option<Vec<u8>>, String> {
{
let state = self.lock_state();
if id == GENESIS_PARENT || !state.indexes.contains_key(&id) {
return Ok(None);
}
}
match self.store.get(id) {
Ok(Some(bytes)) => Ok(Some(bytes)),
Ok(None) => Err("canonical transaction bytes are missing".to_owned()),
Err(error) => {
if store_requires_reopen(&error) {
self.lock_state().reopen_required = true;
}
Err(store_error_message(error))
}
}
}
fn open_inner(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 (mut file, store) = match (ordering_type, store_type) {
(None, None) => {
let store = TransactionStore::create(&store_path).map_err(store_error_message)?;
let file = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&ordering_path)
.map_err(|error| format!("cannot create ordering.dat: {error}"))?;
(file, store)
}
(Some(_), Some(_)) => {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&ordering_path)
.map_err(|error| format!("cannot open ordering.dat: {error}"))?;
let store = TransactionStore::open(&store_path).map_err(store_error_message)?;
(file, store)
}
_ => return Err("ordering root is incomplete".to_owned()),
};
let (order, indexes) = reconstruct_order(&mut file)?;
Ok(Self {
state: Mutex::new(State {
file,
order,
indexes,
subscribers: HashMap::new(),
reopen_required: false,
}),
store,
})
}
fn submit_candidate(
&self,
state: &mut State,
candidate: Candidate<'_>,
) -> Result<(), SubmitError> {
if state.indexes.contains_key(&candidate.id) {
return match self.store_get(state, candidate.id) {
Ok(Some(bytes)) if bytes == candidate.bytes => Ok(()),
Ok(Some(_)) => Err(SubmitError::Other(
"transaction ID collision with canonical bytes".to_owned(),
)),
Ok(None) => Err(SubmitError::Other(
"canonical transaction bytes are missing".to_owned(),
)),
Err(message) => Err(SubmitError::Other(message)),
};
}
let shared_len = if candidate.parent == GENESIS_PARENT {
0
} else {
match state.indexes.get(&candidate.parent) {
Some(index) => index + 1,
None => return Err(SubmitError::MissingParent),
}
};
if shared_len == state.order.len() {
self.extend(state, candidate)
} else {
self.replace_fork(state, shared_len, candidate)
}
}
fn extend(&self, state: &mut State, candidate: Candidate<'_>) -> Result<(), SubmitError> {
self.persist_candidate(state, &candidate)?;
let index = state.order.len();
append_record(state, index, candidate.id, candidate.subsystem)
.map_err(SubmitError::Other)?;
state.indexes.insert(candidate.id, index);
state.order.push((candidate.id, candidate.subsystem));
if let Some(message) = deliver_live(
state,
candidate.subsystem,
candidate.id,
candidate.payload,
index,
) {
return Err(SubmitError::Other(format!(
"transaction committed; {message}"
)));
}
Ok(())
}
fn replace_fork(
&self,
state: &mut State,
shared_len: usize,
candidate: Candidate<'_>,
) -> Result<(), SubmitError> {
let (incumbent_id, incumbent_subsystem) = state.order[shared_len];
let incumbent_bytes = match self.store_get(state, incumbent_id) {
Ok(Some(bytes)) => bytes,
Ok(None) => {
return Err(SubmitError::Other(
"canonical incumbent bytes are missing".to_owned(),
));
}
Err(message) => return Err(SubmitError::Other(message)),
};
let incumbent = Transaction::parse(&incumbent_bytes).map_err(|message| {
SubmitError::Other(format!("canonical incumbent is corrupt: {message}"))
})?;
if incumbent.subsystem() != incumbent_subsystem {
return Err(SubmitError::Other(
"canonical incumbent has the wrong subsystem".to_owned(),
));
}
if incumbent.parent() != candidate.parent {
return Err(SubmitError::Other(
"canonical incumbent has the wrong parent".to_owned(),
));
}
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(()),
ForkDecision::Collision => {
return Err(SubmitError::Other(
"full transaction digest collision".to_owned(),
));
}
ForkDecision::Incoming => {}
}
self.persist_candidate(state, &candidate)?;
truncate_records(state, shared_len).map_err(SubmitError::Other)?;
append_record(state, shared_len, candidate.id, candidate.subsystem)
.map_err(SubmitError::Other)?;
replace_memory(state, shared_len, candidate.id, candidate.subsystem);
let mut errors = notify_reorg(state, shared_len);
if let Some(message) = deliver_live(
state,
candidate.subsystem,
candidate.id,
candidate.payload,
shared_len,
) {
errors.push(message);
}
if errors.is_empty() {
Ok(())
} else {
Err(SubmitError::Other(format!(
"transaction committed; {}",
errors.join("; ")
)))
}
}
fn persist_candidate(
&self,
state: &mut State,
candidate: &Candidate<'_>,
) -> Result<(), SubmitError> {
match self.store.put(candidate.bytes) {
Ok(PutOutcome::Inserted(id)) | Ok(PutOutcome::Duplicate(id)) if id == candidate.id => {
Ok(())
}
Ok(_) => Err(SubmitError::Other(
"transaction store returned an unexpected ID".to_owned(),
)),
Err(error) => {
if store_requires_reopen(&error) {
state.reopen_required = true;
}
Err(SubmitError::Other(store_error_message(error)))
}
}
}
fn store_get(&self, state: &mut State, id: TxId) -> Result<Option<Vec<u8>>, String> {
match self.store.get(id) {
Ok(bytes) => Ok(bytes),
Err(error) => {
if store_requires_reopen(&error) {
state.reopen_required = true;
}
Err(store_error_message(error))
}
}
}
fn lock_state(&self) -> MutexGuard<'_, State> {
self.state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
}
fn prepare_root(root: &Path) -> Result<(), String> {
match fs::symlink_metadata(root) {
Ok(metadata) if metadata.file_type().is_dir() => Ok(()),
Ok(_) => Err("ordering root is not a directory".to_owned()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => fs::create_dir_all(root)
.map_err(|error| format!("cannot create ordering root: {error}")),
Err(error) => Err(format!("cannot inspect ordering 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 ordering component: {error}")),
}
}
fn reconstruct_order(file: &mut File) -> Result<(Order, Indexes), String> {
file.seek(SeekFrom::Start(0))
.map_err(|error| format!("cannot seek ordering.dat: {error}"))?;
let mut bytes = Vec::new();
file.read_to_end(&mut bytes)
.map_err(|error| format!("cannot read ordering.dat: {error}"))?;
if bytes.len() % 32 != 0 {
return Err("ordering.dat length is not a multiple of 32".to_owned());
}
let mut order = Vec::with_capacity(bytes.len() / 32);
let mut indexes = HashMap::with_capacity(bytes.len() / 32);
for chunk in bytes.chunks_exact(32) {
let id = TxId::from_bytes(chunk[..12].try_into().expect("fixed transaction ID range"));
let subsystem =
SubsystemId::from_bytes(chunk[12..].try_into().expect("fixed subsystem ID range"))
.map_err(|message| {
format!("ordering.dat contains an invalid subsystem: {message}")
})?;
if id == GENESIS_PARENT {
return Err("ordering.dat contains the genesis sentinel".to_owned());
}
let index = order.len();
if indexes.insert(id, index).is_some() {
return Err("ordering.dat contains a duplicate transaction ID".to_owned());
}
order.push((id, subsystem));
}
Ok((order, indexes))
}
fn append_record(
state: &mut State,
record_index: usize,
id: TxId,
subsystem: SubsystemId,
) -> Result<(), String> {
let expected_offset = record_offset(record_index)?;
let actual_offset = state
.file
.metadata()
.map_err(|error| format!("cannot inspect ordering.dat: {error}"))?
.len();
if actual_offset != expected_offset {
state.reopen_required = true;
return Err("ordering.dat changed outside this instance; reopen required".to_owned());
}
state
.file
.seek(SeekFrom::Start(expected_offset))
.map_err(|error| format!("cannot seek ordering.dat: {error}"))?;
let record = order_record(id, subsystem);
match state.file.write(&record) {
Ok(32) => {}
Ok(_) | Err(_) => {
state.reopen_required = true;
return Err("ordering append outcome is ambiguous; reopen required".to_owned());
}
}
if state.file.sync_data().is_err() {
state.reopen_required = true;
return Err("ordering append synchronization is ambiguous; reopen required".to_owned());
}
Ok(())
}
fn truncate_records(state: &mut State, records: usize) -> Result<(), String> {
let length = record_offset(records)?;
if state.file.set_len(length).is_err() {
state.reopen_required = true;
return Err("ordering truncation outcome is ambiguous; reopen required".to_owned());
}
if state.file.sync_data().is_err() {
state.reopen_required = true;
return Err("ordering truncation synchronization is ambiguous; reopen required".to_owned());
}
Ok(())
}
fn record_offset(records: usize) -> Result<u64, String> {
let records =
u64::try_from(records).map_err(|_| "ordering.dat offset exceeds u64".to_owned())?;
records
.checked_mul(32)
.ok_or_else(|| "ordering.dat offset exceeds u64".to_owned())
}
fn order_record(id: TxId, subsystem: SubsystemId) -> [u8; 32] {
let mut record = [0_u8; 32];
record[..12].copy_from_slice(id.as_bytes());
record[12..].copy_from_slice(subsystem.as_bytes());
record
}
fn replace_memory(state: &mut State, shared_len: usize, id: TxId, subsystem: SubsystemId) {
for (removed_id, _) in state.order.drain(shared_len..) {
state.indexes.remove(&removed_id);
}
state.indexes.insert(id, shared_len);
state.order.push((id, subsystem));
}
fn notify_reorg(state: &mut State, removed_start: usize) -> Vec<String> {
let mut errors = Vec::new();
for subscriber in state.subscribers.values_mut() {
let affected = subscriber.in_commission
&& subscriber
.latest
.is_some_and(|latest| latest >= removed_start);
if !affected {
continue;
}
let result = subscriber.handler.reorg();
subscriber.in_commission = false;
if let Err(message) = result {
errors.push(format!("reorg callback failed: {message}"));
}
}
errors
}
fn deliver_live(
state: &mut State,
subsystem: SubsystemId,
id: TxId,
payload: &[u8],
index: usize,
) -> Option<String> {
let subscriber = state.subscribers.get_mut(&subsystem)?;
if !subscriber.in_commission {
return None;
}
match subscriber.handler.submit_txn(id, payload) {
Ok(()) => {
subscriber.latest = Some(index);
None
}
Err(message) => {
subscriber.in_commission = false;
Some(format!("subsystem callback failed: {message}"))
}
}
}
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();
digest_decision(
incoming_digest,
incumbent_digest,
incoming_bytes == incumbent_bytes,
)
}
fn digest_decision(incoming: [u8; 32], incumbent: [u8; 32], equal_bytes: bool) -> ForkDecision {
match incoming.cmp(&incumbent) {
Ordering::Less => ForkDecision::Incoming,
Ordering::Greater => ForkDecision::Incumbent,
Ordering::Equal if equal_bytes => ForkDecision::Duplicate,
Ordering::Equal => ForkDecision::Collision,
}
}
fn store_requires_reopen(error: &StoreError) -> bool {
matches!(
error,
StoreError::OutcomeUnknown(_) | StoreError::ReopenRequired
)
}
fn store_error_message(error: StoreError) -> String {
format!("transaction store error: {error}")
}
fn reopen_required_message() -> String {
"instance requires reopening before further mutation".to_owned()
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering as AtomicOrdering};
static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
struct TempRoot {
path: PathBuf,
}
impl TempRoot {
fn new(label: &str) -> Self {
let sequence = NEXT_ROOT.fetch_add(1, AtomicOrdering::Relaxed);
let path = std::env::temp_dir().join(format!(
"kcode-k1-txn-ordering-{}-{}-{}",
std::process::id(),
sequence,
label
));
let _ = fs::remove_dir_all(&path);
Self { path }
}
}
impl Drop for TempRoot {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.path);
}
}
struct RecordingSubsystem {
submissions: Mutex<Vec<(TxId, Vec<u8>)>>,
reorgs: AtomicUsize,
fail_submit: AtomicBool,
fail_reorg: AtomicBool,
}
impl RecordingSubsystem {
fn new() -> Self {
Self {
submissions: Mutex::new(Vec::new()),
reorgs: AtomicUsize::new(0),
fail_submit: AtomicBool::new(false),
fail_reorg: AtomicBool::new(false),
}
}
fn submission_count(&self) -> usize {
self.submissions.lock().unwrap().len()
}
}
impl Subsystem for RecordingSubsystem {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
self.submissions
.lock()
.unwrap()
.push((id, payload.to_vec()));
if self.fail_submit.load(AtomicOrdering::Relaxed) {
Err("submit failure".to_owned())
} else {
Ok(())
}
}
fn reorg(&self) -> Result<(), String> {
self.reorgs.fetch_add(1, AtomicOrdering::Relaxed);
if self.fail_reorg.load(AtomicOrdering::Relaxed) {
Err("reorg failure".to_owned())
} else {
Ok(())
}
}
}
fn id(value: u64) -> TxId {
let mut bytes = [0_u8; 12];
bytes[..4].copy_from_slice(b"KTO!");
bytes[4..].copy_from_slice(&value.to_be_bytes());
TxId::from_bytes(bytes)
}
fn subsystem(value: u8) -> SubsystemId {
SubsystemId::from_bytes([value; 20]).unwrap()
}
fn transaction(
parent: TxId,
creator: [u8; 32],
timestamp: u64,
subsystem: SubsystemId,
payload: &[u8],
signature: u8,
) -> Vec<u8> {
let mut bytes = Vec::with_capacity(136 + payload.len());
bytes.extend_from_slice(parent.as_bytes());
bytes.extend_from_slice(×tamp.to_le_bytes());
bytes.extend_from_slice(&creator);
bytes.extend_from_slice(subsystem.as_bytes());
bytes.extend_from_slice(payload);
bytes.extend_from_slice(&[signature; 64]);
bytes
}
fn write_order(root: &Path, entries: &[(TxId, SubsystemId)]) {
let mut file = OpenOptions::new()
.write(true)
.truncate(true)
.open(root.join("ordering.dat"))
.unwrap();
for (id, subsystem) in entries {
file.write_all(&order_record(*id, *subsystem)).unwrap();
}
file.sync_data().unwrap();
}
#[test]
fn creates_reopens_and_validates_ordering_roots() {
let root = TempRoot::new("root");
let ordering = K1TxnOrdering::open(&root.path).unwrap();
assert!(root.path.join("ordering.dat").is_file());
assert!(root.path.join("k1-transaction-store").is_dir());
drop(ordering);
assert!(K1TxnOrdering::open(&root.path).is_ok());
let mixed = TempRoot::new("mixed");
fs::create_dir_all(&mixed.path).unwrap();
File::create(mixed.path.join("ordering.dat")).unwrap();
assert!(K1TxnOrdering::open(&mixed.path).is_err());
let mut malformed = OpenOptions::new()
.write(true)
.truncate(true)
.open(root.path.join("ordering.dat"))
.unwrap();
malformed.write_all(&[0_u8; 31]).unwrap();
malformed.sync_data().unwrap();
drop(malformed);
assert!(K1TxnOrdering::open(&root.path).is_err());
write_order(
&root.path,
&[(id(1), subsystem(b'a')), (id(1), subsystem(b'b'))],
);
assert!(K1TxnOrdering::open(&root.path).is_err());
let mut invalid_subsystem = [0_u8; 32];
invalid_subsystem[..12].copy_from_slice(id(2).as_bytes());
invalid_subsystem[12..].fill(0xff);
fs::write(root.path.join("ordering.dat"), invalid_subsystem).unwrap();
assert!(K1TxnOrdering::open(&root.path).is_err());
}
#[test]
fn provides_canonical_queries_and_even_sampling() {
let root = TempRoot::new("queries");
drop(K1TxnOrdering::open(&root.path).unwrap());
let entries: Vec<_> = (0_u64..260)
.map(|value| (id(value), subsystem(b'q')))
.collect();
write_order(&root.path, &entries);
let ordering = K1TxnOrdering::open(&root.path).unwrap();
assert!(ordering.contains(id(0)));
assert!(!ordering.contains(GENESIS_PARENT));
assert_eq!(ordering.tip(), Some(id(259)));
assert_eq!(ordering.get_txn(id(9999)).unwrap(), None);
assert!(ordering.get_txn(id(0)).is_err());
let short = ordering.between_txids(id(10), id(20)).unwrap();
assert_eq!(short, (11_u64..20).map(id).collect::<Vec<_>>());
assert!(ordering.between_txids(id(10), id(10)).unwrap().is_empty());
assert!(ordering.between_txids(id(20), id(10)).is_err());
assert!(ordering.between_txids(id(9999), id(10)).is_err());
let sampled = ordering.between_txids(GENESIS_PARENT, id(200)).unwrap();
assert_eq!(sampled.len(), 128);
for (offset, actual) in sampled.iter().enumerate() {
let k = offset as i128 + 1;
let expected = -1_i128 + k * 201 / 129;
assert_eq!(*actual, id(expected as u64));
}
}
#[test]
fn submits_replays_reorganizes_and_retains_removed_bytes() {
let root = TempRoot::new("workflow");
let ordering = K1TxnOrdering::open(&root.path).unwrap();
let subsystem_a = subsystem(b'a');
let subsystem_b = subsystem(b'b');
let handler_a = Arc::new(RecordingSubsystem::new());
let handler_b = Arc::new(RecordingSubsystem::new());
ordering
.register_subsystem(subsystem_a, None, handler_a.clone())
.unwrap();
let first = transaction(GENESIS_PARENT, [20; 32], 10, subsystem_a, b"first", 1);
let first_id = TxId::for_transaction(&first);
ordering.submit_txn(&first).unwrap();
ordering.submit_txn(&first).unwrap();
assert_eq!(handler_a.submission_count(), 1);
let missing = transaction(id(9999), [1; 32], 1, subsystem_a, b"missing", 2);
assert!(matches!(
ordering.submit_txn(&missing),
Err(SubmitError::MissingParent)
));
assert!(!ordering.contains(TxId::for_transaction(&missing)));
let second = transaction(first_id, [50; 32], 20, subsystem_b, b"second", 3);
let second_id = TxId::for_transaction(&second);
ordering.submit_txn(&second).unwrap();
ordering
.register_subsystem(subsystem_b, None, handler_b.clone())
.unwrap();
assert_eq!(handler_b.submission_count(), 1);
let third = transaction(second_id, [50; 32], 30, subsystem_a, b"third", 4);
let third_id = TxId::for_transaction(&third);
ordering.submit_txn(&third).unwrap();
assert_eq!(handler_a.submission_count(), 2);
let replacement = transaction(first_id, [1; 32], 100, subsystem_a, b"replacement", 5);
let replacement_id = TxId::for_transaction(&replacement);
ordering.submit_txn(&replacement).unwrap();
assert!(ordering.contains(first_id));
assert!(ordering.contains(replacement_id));
assert!(!ordering.contains(second_id));
assert!(!ordering.contains(third_id));
assert_eq!(handler_a.reorgs.load(AtomicOrdering::Relaxed), 1);
assert_eq!(handler_b.reorgs.load(AtomicOrdering::Relaxed), 1);
assert_eq!(handler_a.submission_count(), 2);
ordering
.register_subsystem(subsystem_a, Some(first_id), handler_a.clone())
.unwrap();
assert_eq!(handler_a.submission_count(), 3);
ordering
.register_subsystem(subsystem_b, None, handler_b.clone())
.unwrap();
let extension = transaction(replacement_id, [2; 32], 200, subsystem_b, b"extension", 6);
let extension_id = TxId::for_transaction(&extension);
ordering.submit_txn(&extension).unwrap();
assert!(ordering.contains(extension_id));
assert_eq!(handler_b.submission_count(), 2);
let loser = transaction(first_id, [250; 32], 1, subsystem_b, b"loser", 7);
let loser_id = TxId::for_transaction(&loser);
assert!(matches!(
ordering.submit_txn(&loser),
Err(SubmitError::Other(_))
));
assert!(!ordering.contains(loser_id));
assert_eq!(
ordering.get_txn(replacement_id).unwrap().unwrap(),
replacement
);
assert_eq!(ordering.get_txn(second_id).unwrap(), None);
drop(ordering);
let store = TransactionStore::open(&root.path.join("k1-transaction-store")).unwrap();
assert!(store.contains(second_id));
assert!(store.contains(third_id));
assert!(!store.contains(loser_id));
}
#[test]
fn callback_failure_blocks_until_reregistration() {
let root = TempRoot::new("callback");
let ordering = K1TxnOrdering::open(&root.path).unwrap();
let subsystem_c = subsystem(b'c');
let handler = Arc::new(RecordingSubsystem::new());
handler.fail_submit.store(true, AtomicOrdering::Relaxed);
ordering
.register_subsystem(subsystem_c, None, handler.clone())
.unwrap();
let first = transaction(GENESIS_PARENT, [1; 32], 1, subsystem_c, b"first", 1);
let first_id = TxId::for_transaction(&first);
assert!(matches!(
ordering.submit_txn(&first),
Err(SubmitError::Other(_))
));
assert!(ordering.contains(first_id));
let second = transaction(first_id, [1; 32], 2, subsystem_c, b"second", 2);
ordering.submit_txn(&second).unwrap();
assert_eq!(handler.submission_count(), 1);
handler.fail_submit.store(false, AtomicOrdering::Relaxed);
ordering
.register_subsystem(subsystem_c, None, handler.clone())
.unwrap();
assert_eq!(handler.submission_count(), 3);
assert!(
ordering
.register_subsystem(subsystem_c, None, handler.clone())
.is_err()
);
let wrong = subsystem(b'd');
assert!(
ordering
.register_subsystem(wrong, Some(first_id), Arc::new(RecordingSubsystem::new()))
.is_err()
);
}
#[test]
fn fork_ranking_uses_creator_timestamp_then_digest() {
let low = [1_u8; 32];
let high = [2_u8; 32];
assert_eq!(
fork_decision(&low, 10, b"x", &high, 1, b"y"),
ForkDecision::Incoming
);
assert_eq!(
fork_decision(&high, 1, b"x", &low, 10, b"y"),
ForkDecision::Incumbent
);
assert_eq!(
fork_decision(&low, 1, b"x", &low, 2, b"y"),
ForkDecision::Incoming
);
assert_eq!(
fork_decision(&low, 2, b"x", &low, 1, b"y"),
ForkDecision::Incumbent
);
assert_eq!(
fork_decision(&low, 1, b"same", &low, 1, b"same"),
ForkDecision::Duplicate
);
let left: [u8; 32] = Sha256::digest(b"left").into();
let right: [u8; 32] = Sha256::digest(b"right").into();
let expected = if left < right {
ForkDecision::Incoming
} else {
ForkDecision::Incumbent
};
assert_eq!(fork_decision(&low, 1, b"left", &low, 1, b"right"), expected);
assert_eq!(
digest_decision([4; 32], [4; 32], false),
ForkDecision::Collision
);
}
#[test]
fn ambiguous_order_change_leaves_an_orphan_and_requires_reopen() {
let root = TempRoot::new("ambiguous");
let ordering = K1TxnOrdering::open(&root.path).unwrap();
let bytes = transaction(GENESIS_PARENT, [1; 32], 1, subsystem(b'e'), b"orphan", 1);
let transaction_id = TxId::for_transaction(&bytes);
let mut external = OpenOptions::new()
.append(true)
.open(root.path.join("ordering.dat"))
.unwrap();
external.write_all(&[0]).unwrap();
external.sync_data().unwrap();
assert!(matches!(
ordering.submit_txn(&bytes),
Err(SubmitError::Other(_))
));
assert!(!ordering.contains(transaction_id));
assert!(
ordering
.register_subsystem(subsystem(b'e'), None, Arc::new(RecordingSubsystem::new()))
.is_err()
);
drop(ordering);
let store = TransactionStore::open(&root.path.join("k1-transaction-store")).unwrap();
assert!(store.contains(transaction_id));
}
#[test]
fn million_entry_open_and_scan_fixture() {
let root = TempRoot::new("million");
drop(K1TxnOrdering::open(&root.path).unwrap());
let count = 1_000_000_u64;
let entry_subsystem = subsystem(b'm');
let mut bytes = Vec::with_capacity(count as usize * 32);
for value in 0..count {
bytes.extend_from_slice(id(value).as_bytes());
bytes.extend_from_slice(entry_subsystem.as_bytes());
}
let mut file = OpenOptions::new()
.write(true)
.truncate(true)
.open(root.path.join("ordering.dat"))
.unwrap();
file.write_all(&bytes).unwrap();
file.sync_data().unwrap();
drop(file);
let started = Instant::now();
let ordering = K1TxnOrdering::open(&root.path).unwrap();
assert!(started.elapsed() < Duration::from_secs(5));
let scan_started = Instant::now();
let matching = ordering
.lock_state()
.order
.iter()
.filter(|entry| entry.1 == entry_subsystem)
.count();
assert_eq!(matching, count as usize);
assert!(scan_started.elapsed() < Duration::from_secs(1));
assert!(ordering.contains(id(0)));
assert!(ordering.contains(id(count - 1)));
assert_eq!(
ordering
.between_txids(GENESIS_PARENT, id(count - 1))
.unwrap()
.len(),
128
);
}
}