1use kcode_k1_order_store::OrderStore;
2pub use kcode_k1_transaction::{GENESIS_PARENT, SubsystemId};
3use kcode_k1_transaction::{Transaction, build_signed_transaction};
4pub use kcode_k1_transaction_store::TxId;
5use kcode_k1_transaction_store::{PutOutcome, StoreError, TransactionStore};
6use sha2::{Digest, Sha256};
7use std::{cmp::Ordering, fmt, fs, path::Path};
8
9pub struct CanonicalChain {
10 order: OrderStore,
11 store: TransactionStore,
12}
13
14pub struct ReplayCursor {
15 subsystem: SubsystemId,
16 after: Option<TxId>,
17}
18
19pub struct ReplayTransaction {
20 pub id: TxId,
21 pub bytes: Vec<u8>,
22}
23
24#[derive(Clone, Copy, Debug, Eq, PartialEq)]
25pub enum CommitOutcome {
26 Duplicate,
27 Extension { id: TxId, subsystem: SubsystemId },
28 Reorganization { id: TxId, subsystem: SubsystemId },
29}
30
31#[derive(Debug)]
32pub enum SubmitError {
33 MissingParent,
34 Other(String),
35}
36
37impl fmt::Display for SubmitError {
38 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
39 match self {
40 Self::MissingParent => formatter.write_str("missing parent"),
41 Self::Other(message) => formatter.write_str(message),
42 }
43 }
44}
45
46impl std::error::Error for SubmitError {}
47
48struct Candidate<'a> {
49 id: TxId,
50 parent: TxId,
51 creator: [u8; 32],
52 timestamp: u64,
53 subsystem: SubsystemId,
54 bytes: &'a [u8],
55}
56
57#[derive(Clone, Copy, Debug, Eq, PartialEq)]
58enum ForkDecision {
59 Incoming,
60 Incumbent,
61 Duplicate,
62 Collision,
63}
64
65impl CanonicalChain {
66 pub fn open(root: &Path) -> Result<Self, String> {
67 prepare_root(root)?;
68
69 let ordering_path = root.join("ordering.dat");
70 let store_path = root.join("k1-transaction-store");
71 let ordering_type = path_type(&ordering_path)?;
72 let store_type = path_type(&store_path)?;
73
74 if ordering_type.is_some_and(|kind| !kind.is_file()) {
75 return Err("ordering.dat is not a regular file".to_owned());
76 }
77 if store_type.is_some_and(|kind| !kind.is_dir()) {
78 return Err("k1-transaction-store is not a directory".to_owned());
79 }
80
81 let (order, store) = match (ordering_type, store_type) {
82 (None, None) => {
83 let store = TransactionStore::create(&store_path)
84 .unwrap_or_else(|error| fatal("create-transaction-store", error));
85 let order = OrderStore::create(&ordering_path)
86 .unwrap_or_else(|error| fatal("create-order-store", error));
87 (order, store)
88 }
89 (Some(_), Some(_)) => (
90 OrderStore::open(&ordering_path)?,
91 TransactionStore::open(&store_path)
92 .map_err(|error| format!("transaction store error: {error}"))?,
93 ),
94 _ => return Err("canonical chain root is incomplete".to_owned()),
95 };
96
97 Ok(Self { order, store })
98 }
99
100 pub fn submit_validated(&mut self, transaction: &[u8]) -> Result<CommitOutcome, SubmitError> {
101 let parsed = Transaction::parse(transaction)
102 .map_err(|message| SubmitError::Other(format!("invalid transaction: {message}")))?;
103 let candidate = Candidate {
104 id: TxId::for_transaction(transaction),
105 parent: parsed.parent(),
106 creator: *parsed.creator(),
107 timestamp: parsed.timestamp(),
108 subsystem: parsed.subsystem(),
109 bytes: transaction,
110 };
111
112 if candidate.id == GENESIS_PARENT {
113 return Err(SubmitError::Other(
114 "transaction ID collides with the genesis sentinel".to_owned(),
115 ));
116 }
117
118 self.submit_candidate(candidate)
119 }
120
121 pub fn submit_local<F>(
122 &mut self,
123 timestamp: u64,
124 creator: [u8; 32],
125 subsystem: SubsystemId,
126 payload: &[u8],
127 signer: F,
128 ) -> Result<Vec<u8>, String>
129 where
130 F: FnOnce(&[u8]) -> Result<[u8; 64], String>,
131 {
132 let parent = self.tip().unwrap_or(GENESIS_PARENT);
133 let bytes =
134 build_signed_transaction(parent, timestamp, creator, subsystem, payload, signer)?;
135 let id = TxId::for_transaction(&bytes);
136
137 if id == GENESIS_PARENT {
138 return Err("transaction ID collides with the genesis sentinel".to_owned());
139 }
140 if self.order.index_of(id).is_some() {
141 return Err("transaction ID collides with a canonical transaction".to_owned());
142 }
143
144 self.persist(&bytes, id)
145 .map_err(|error| error.to_string())?;
146 self.order
147 .commit(self.order.entries().len(), id, subsystem)
148 .unwrap_or_else(|error| fatal("commit-local-order", error));
149 Ok(bytes)
150 }
151
152 pub fn contains(&self, id: TxId) -> bool {
153 id != GENESIS_PARENT && self.order.index_of(id).is_some()
154 }
155
156 pub fn tip(&self) -> Option<TxId> {
157 self.order.entries().last().map(|entry| entry.0)
158 }
159
160 pub fn between_txids(&self, older: TxId, newer: TxId) -> Result<Vec<TxId>, String> {
161 let older_index = if older == GENESIS_PARENT {
162 -1_i128
163 } else {
164 self.order
165 .index_of(older)
166 .map(|index| index as i128)
167 .ok_or_else(|| "older boundary is not canonical".to_owned())?
168 };
169 let newer_index = self
170 .order
171 .index_of(newer)
172 .map(|index| index as i128)
173 .ok_or_else(|| "newer boundary is not canonical".to_owned())?;
174
175 if older_index == newer_index {
176 return Ok(Vec::new());
177 }
178 if older_index > newer_index {
179 return Err("transaction boundaries are reversed".to_owned());
180 }
181
182 let interior = newer_index - older_index - 1;
183 if interior <= 128 {
184 return Ok(((older_index + 1)..newer_index)
185 .map(|index| self.order.entries()[index as usize].0)
186 .collect());
187 }
188
189 let distance = newer_index - older_index;
190 Ok((1_i128..=128)
191 .map(|k| self.order.entries()[(older_index + k * distance / 129) as usize].0)
192 .collect())
193 }
194
195 pub fn get_txn(&self, id: TxId) -> Result<Option<Vec<u8>>, String> {
196 if !self.contains(id) {
197 return Ok(None);
198 }
199 Ok(Some(self.canonical_bytes(id)))
200 }
201
202 pub fn replay_cursor(
203 &self,
204 subsystem: SubsystemId,
205 after: Option<TxId>,
206 ) -> Result<ReplayCursor, String> {
207 if let Some(id) = after {
208 let index = self
209 .order
210 .index_of(id)
211 .ok_or_else(|| "replay checkpoint is not canonical".to_owned())?;
212 if self.order.entries()[index].1 != subsystem {
213 return Err("replay checkpoint belongs to another subsystem".to_owned());
214 }
215 }
216
217 Ok(ReplayCursor { subsystem, after })
218 }
219
220 pub fn replay_next(
221 &self,
222 cursor: &mut ReplayCursor,
223 ) -> Result<Option<ReplayTransaction>, String> {
224 let start = match cursor.after {
225 None => 0,
226 Some(id) => {
227 self.order
228 .index_of(id)
229 .ok_or_else(|| "replay cursor is no longer canonical".to_owned())?
230 + 1
231 }
232 };
233
234 for &(id, subsystem) in &self.order.entries()[start..] {
235 if subsystem != cursor.subsystem {
236 continue;
237 }
238
239 let bytes = self.canonical_bytes(id);
240 let parsed = Transaction::parse(&bytes)
241 .unwrap_or_else(|error| fatal("parse-canonical-transaction", error));
242 if TxId::for_transaction(&bytes) != id || parsed.subsystem() != subsystem {
243 fatal(
244 "verify-canonical-transaction",
245 "canonical transaction does not match its order record",
246 );
247 }
248
249 cursor.after = Some(id);
250 return Ok(Some(ReplayTransaction { id, bytes }));
251 }
252
253 Ok(None)
254 }
255
256 fn submit_candidate(&mut self, candidate: Candidate<'_>) -> Result<CommitOutcome, SubmitError> {
257 if self.order.index_of(candidate.id).is_some() {
258 return if self.canonical_bytes(candidate.id) == candidate.bytes {
259 Ok(CommitOutcome::Duplicate)
260 } else {
261 Err(SubmitError::Other(
262 "transaction ID collision with canonical bytes".to_owned(),
263 ))
264 };
265 }
266
267 let shared_len = if candidate.parent == GENESIS_PARENT {
268 0
269 } else {
270 self.order
271 .index_of(candidate.parent)
272 .map(|index| index + 1)
273 .ok_or(SubmitError::MissingParent)?
274 };
275
276 if shared_len == self.order.entries().len() {
277 self.persist(candidate.bytes, candidate.id)?;
278 self.order
279 .commit(shared_len, candidate.id, candidate.subsystem)
280 .unwrap_or_else(|error| fatal("commit-extension-order", error));
281 return Ok(CommitOutcome::Extension {
282 id: candidate.id,
283 subsystem: candidate.subsystem,
284 });
285 }
286
287 let (incumbent_id, incumbent_subsystem) = self.order.entries()[shared_len];
288 let incumbent_bytes = self.canonical_bytes(incumbent_id);
289 let incumbent = Transaction::parse(&incumbent_bytes)
290 .unwrap_or_else(|error| fatal("parse-canonical-incumbent", error));
291 if incumbent.subsystem() != incumbent_subsystem || incumbent.parent() != candidate.parent {
292 fatal(
293 "verify-canonical-incumbent",
294 "canonical incumbent does not match its order record or parent",
295 );
296 }
297
298 match fork_decision(
299 &candidate.creator,
300 candidate.timestamp,
301 candidate.bytes,
302 incumbent.creator(),
303 incumbent.timestamp(),
304 &incumbent_bytes,
305 ) {
306 ForkDecision::Incumbent => {
307 return Err(SubmitError::Other(
308 "fork loses canonical ordering".to_owned(),
309 ));
310 }
311 ForkDecision::Duplicate => return Ok(CommitOutcome::Duplicate),
312 ForkDecision::Collision => {
313 return Err(SubmitError::Other(
314 "full transaction digest collision".to_owned(),
315 ));
316 }
317 ForkDecision::Incoming => {}
318 }
319
320 self.persist(candidate.bytes, candidate.id)?;
321 self.order
322 .commit(shared_len, candidate.id, candidate.subsystem)
323 .unwrap_or_else(|error| fatal("commit-reorganization-order", error));
324 Ok(CommitOutcome::Reorganization {
325 id: candidate.id,
326 subsystem: candidate.subsystem,
327 })
328 }
329
330 fn persist(&self, bytes: &[u8], expected: TxId) -> Result<(), SubmitError> {
331 match self.store.put(bytes) {
332 Ok(PutOutcome::Inserted(id)) | Ok(PutOutcome::Duplicate(id)) if id == expected => {
333 Ok(())
334 }
335 Ok(_) => fatal(
336 "persist-transaction",
337 "transaction store returned an unexpected transaction ID",
338 ),
339 Err(StoreError::IdCollision(_)) => Err(SubmitError::Other(
340 "transaction ID collision with stored bytes".to_owned(),
341 )),
342 Err(error) => fatal("persist-transaction", error),
343 }
344 }
345
346 fn canonical_bytes(&self, id: TxId) -> Vec<u8> {
347 match self.store.get(id) {
348 Ok(Some(bytes)) => bytes,
349 Ok(None) => fatal(
350 "load-canonical-transaction",
351 "canonical transaction bytes are missing",
352 ),
353 Err(error) => fatal("load-canonical-transaction", error),
354 }
355 }
356}
357
358fn prepare_root(root: &Path) -> Result<(), String> {
359 match fs::symlink_metadata(root) {
360 Ok(metadata) if metadata.file_type().is_dir() => Ok(()),
361 Ok(_) => Err("canonical chain root is not a directory".to_owned()),
362 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
363 fs::create_dir_all(root).unwrap_or_else(|error| fatal("create-root", error));
364 Ok(())
365 }
366 Err(error) => Err(format!("cannot inspect canonical chain root: {error}")),
367 }
368}
369
370fn path_type(path: &Path) -> Result<Option<fs::FileType>, String> {
371 match fs::symlink_metadata(path) {
372 Ok(metadata) => Ok(Some(metadata.file_type())),
373 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
374 Err(error) => Err(format!("cannot inspect canonical chain component: {error}")),
375 }
376}
377
378fn fork_decision(
379 incoming_creator: &[u8; 32],
380 incoming_timestamp: u64,
381 incoming_bytes: &[u8],
382 incumbent_creator: &[u8; 32],
383 incumbent_timestamp: u64,
384 incumbent_bytes: &[u8],
385) -> ForkDecision {
386 match incoming_creator.cmp(incumbent_creator) {
387 Ordering::Less => return ForkDecision::Incoming,
388 Ordering::Greater => return ForkDecision::Incumbent,
389 Ordering::Equal => {}
390 }
391 match incoming_timestamp.cmp(&incumbent_timestamp) {
392 Ordering::Less => return ForkDecision::Incoming,
393 Ordering::Greater => return ForkDecision::Incumbent,
394 Ordering::Equal => {}
395 }
396
397 let incoming_digest: [u8; 32] = Sha256::digest(incoming_bytes).into();
398 let incumbent_digest: [u8; 32] = Sha256::digest(incumbent_bytes).into();
399 match incoming_digest.cmp(&incumbent_digest) {
400 Ordering::Less => ForkDecision::Incoming,
401 Ordering::Greater => ForkDecision::Incumbent,
402 Ordering::Equal if incoming_bytes == incumbent_bytes => ForkDecision::Duplicate,
403 Ordering::Equal => ForkDecision::Collision,
404 }
405}
406
407fn fatal(operation: &str, error: impl fmt::Display) -> ! {
408 eprintln!("kcode-k1-canonical-chain fatal {operation}: {error}");
409 std::process::abort()
410}
411
412#[cfg(test)]
413mod tests {
414 use super::*;
415 use std::cell::Cell;
416 use std::path::PathBuf;
417 use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
418 use std::time::{Duration, Instant};
419
420 static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
421
422 struct TempRoot(PathBuf);
423
424 impl TempRoot {
425 fn new(label: &str) -> Self {
426 let number = NEXT_ROOT.fetch_add(1, AtomicOrdering::Relaxed);
427 let path = std::env::temp_dir().join(format!(
428 "kcode-k1-canonical-chain-{}-{number}-{label}",
429 std::process::id()
430 ));
431 let _ = fs::remove_dir_all(&path);
432 Self(path)
433 }
434 }
435
436 impl Drop for TempRoot {
437 fn drop(&mut self) {
438 let _ = fs::remove_dir_all(&self.0);
439 }
440 }
441
442 fn subsystem(value: u8) -> SubsystemId {
443 SubsystemId::from_bytes([value; 20]).unwrap()
444 }
445
446 fn transaction(
447 parent: TxId,
448 creator: u8,
449 timestamp: u64,
450 subsystem: SubsystemId,
451 payload: &[u8],
452 ) -> Vec<u8> {
453 build_signed_transaction(parent, timestamp, [creator; 32], subsystem, payload, |_| {
454 Ok([creator; 64])
455 })
456 .unwrap()
457 }
458
459 #[test]
460 fn local_submission_parents_and_signer_error_does_not_mutate() {
461 let root = TempRoot::new("local");
462 let mut chain = CanonicalChain::open(&root.0).unwrap();
463 let calls = Cell::new(0);
464 let error = chain.submit_local(0, [0; 32], subsystem(b'a'), b"bad", |_| {
465 calls.set(calls.get() + 1);
466 Err("signer stopped".to_owned())
467 });
468 assert_eq!(error.unwrap_err(), "signer stopped");
469 assert_eq!((calls.get(), chain.tip()), (1, None));
470 assert_eq!(fs::metadata(root.0.join("ordering.dat")).unwrap().len(), 0);
471
472 let first = chain
473 .submit_local(1, [1; 32], subsystem(b'a'), b"first", |_| Ok([2; 64]))
474 .unwrap();
475 let second = chain
476 .submit_local(2, [1; 32], subsystem(b'b'), b"second", |_| Ok([3; 64]))
477 .unwrap();
478 let first_id = TxId::for_transaction(&first);
479 assert_eq!(Transaction::parse(&first).unwrap().parent(), GENESIS_PARENT);
480 assert_eq!(Transaction::parse(&second).unwrap().parent(), first_id);
481 assert_eq!(chain.get_txn(first_id).unwrap(), Some(first));
482
483 drop(chain);
484 assert_eq!(
485 CanonicalChain::open(&root.0).unwrap().tip(),
486 Some(TxId::for_transaction(&second))
487 );
488 }
489
490 #[test]
491 fn remote_reorganization_reopens_and_retains_orphans() {
492 let root = TempRoot::new("remote");
493 let mut chain = CanonicalChain::open(&root.0).unwrap();
494 let owner = subsystem(b'a');
495 let first = transaction(GENESIS_PARENT, 20, 1, owner, b"first");
496 let first_id = TxId::for_transaction(&first);
497 assert!(matches!(
498 chain.submit_validated(&first),
499 Ok(CommitOutcome::Extension { .. })
500 ));
501 assert_eq!(
502 chain.submit_validated(&first).unwrap(),
503 CommitOutcome::Duplicate
504 );
505
506 let incumbent = transaction(first_id, 50, 2, owner, b"incumbent");
507 let incumbent_id = TxId::for_transaction(&incumbent);
508 chain.submit_validated(&incumbent).unwrap();
509 let descendant = transaction(incumbent_id, 50, 3, owner, b"descendant");
510 let descendant_id = TxId::for_transaction(&descendant);
511 chain.submit_validated(&descendant).unwrap();
512
513 let replacement = transaction(first_id, 1, 99, owner, b"replacement");
514 let replacement_id = TxId::for_transaction(&replacement);
515 assert!(matches!(
516 chain.submit_validated(&replacement),
517 Ok(CommitOutcome::Reorganization { .. })
518 ));
519 let loser = transaction(first_id, 250, 0, owner, b"loser");
520 let loser_id = TxId::for_transaction(&loser);
521 assert!(chain.submit_validated(&loser).is_err());
522 let removed_child = transaction(descendant_id, 0, 4, owner, b"removed-parent");
523 let removed_child_id = TxId::for_transaction(&removed_child);
524 assert!(matches!(
525 chain.submit_validated(&removed_child),
526 Err(SubmitError::MissingParent)
527 ));
528
529 drop(chain);
530 let reopened = CanonicalChain::open(&root.0).unwrap();
531 assert_eq!(reopened.tip(), Some(replacement_id));
532 assert!(!reopened.contains(incumbent_id));
533 assert!(!reopened.contains(descendant_id));
534 drop(reopened);
535
536 let store = TransactionStore::open(&root.0.join("k1-transaction-store")).unwrap();
537 assert!(store.contains(incumbent_id));
538 assert!(store.contains(descendant_id));
539 assert!(!store.contains(loser_id));
540 assert!(!store.contains(removed_child_id));
541 }
542
543 #[test]
544 fn fork_ranking_uses_timestamp_then_complete_digest() {
545 let creator = [1; 32];
546 assert_eq!(
547 fork_decision(&creator, 1, b"x", &creator, 2, b"y"),
548 ForkDecision::Incoming
549 );
550 let left: [u8; 32] = Sha256::digest(b"left").into();
551 let right: [u8; 32] = Sha256::digest(b"right").into();
552 assert_eq!(
553 fork_decision(&creator, 1, b"left", &creator, 1, b"right"),
554 if left < right {
555 ForkDecision::Incoming
556 } else {
557 ForkDecision::Incumbent
558 }
559 );
560 }
561
562 #[test]
563 fn queries_replay_and_removed_cursor_behave_canonically() {
564 let root = TempRoot::new("queries");
565 let mut chain = CanonicalChain::open(&root.0).unwrap();
566 let (a, b) = (subsystem(b'a'), subsystem(b'b'));
567 let mut parent = GENESIS_PARENT;
568 let mut ids = Vec::new();
569
570 for index in 0..140_u64 {
571 let owner = if index % 2 == 0 { a } else { b };
572 let bytes = transaction(parent, 1, index, owner, &[index as u8]);
573 parent = TxId::for_transaction(&bytes);
574 ids.push(parent);
575 chain.submit_validated(&bytes).unwrap();
576 }
577
578 assert_eq!(chain.between_txids(ids[3], ids[10]).unwrap(), ids[4..10]);
579 assert_eq!(
580 chain
581 .between_txids(GENESIS_PARENT, *ids.last().unwrap())
582 .unwrap()
583 .len(),
584 128
585 );
586 assert!(chain.between_txids(ids[10], ids[3]).is_err());
587
588 let mut cursor = chain.replay_cursor(a, Some(ids[0])).unwrap();
589 assert_eq!(chain.replay_next(&mut cursor).unwrap().unwrap().id, ids[2]);
590 let replacement = transaction(ids[0], 0, 999, b, b"replacement");
591 chain.submit_validated(&replacement).unwrap();
592 assert!(chain.replay_next(&mut cursor).is_err());
593 }
594
595 #[test]
596 fn root_shapes_remain_compatible() {
597 let empty = TempRoot::new("empty");
598 fs::create_dir_all(&empty.0).unwrap();
599 fs::write(empty.0.join("extra"), []).unwrap();
600 drop(CanonicalChain::open(&empty.0).unwrap());
601
602 let mixed = TempRoot::new("mixed");
603 fs::create_dir_all(&mixed.0).unwrap();
604 fs::write(mixed.0.join("ordering.dat"), []).unwrap();
605 assert!(CanonicalChain::open(&mixed.0).is_err());
606
607 let malformed = TempRoot::new("malformed");
608 drop(CanonicalChain::open(&malformed.0).unwrap());
609 fs::write(malformed.0.join("ordering.dat"), [0; 31]).unwrap();
610 assert!(CanonicalChain::open(&malformed.0).is_err());
611 }
612
613 #[test]
614 fn opens_million_record_fixture_under_five_seconds() {
615 let root = TempRoot::new("million");
616 drop(CanonicalChain::open(&root.0).unwrap());
617 let count = 1_000_000_u64;
618 let owner = subsystem(b'm');
619 let mut bytes = Vec::with_capacity(count as usize * 32);
620
621 for value in 0..count {
622 let mut id = [0_u8; 12];
623 id[..8].copy_from_slice(&value.to_le_bytes());
624 bytes.extend_from_slice(&id);
625 bytes.extend_from_slice(owner.as_bytes());
626 }
627 fs::write(root.0.join("ordering.dat"), bytes).unwrap();
628
629 let started = Instant::now();
630 let chain = CanonicalChain::open(&root.0).unwrap();
631 assert!(started.elapsed() < Duration::from_secs(5));
632 assert_eq!(chain.order.entries().len(), count as usize);
633 let mut last = [0_u8; 12];
634 last[..8].copy_from_slice(&(count - 1).to_le_bytes());
635 assert!(chain.contains(TxId::from_bytes(last)));
636 }
637}