1use kcode_k1_order_store::OrderStore;
2pub use kcode_k1_transaction::{GENESIS_PARENT, REGISTER_AT_TIP, 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}
45impl std::error::Error for SubmitError {}
46
47struct Candidate<'a> {
48 id: TxId,
49 parent: TxId,
50 creator: [u8; 32],
51 timestamp: u64,
52 subsystem: SubsystemId,
53 bytes: &'a [u8],
54}
55#[derive(Clone, Copy, Debug, Eq, PartialEq)]
56enum ForkDecision {
57 Incoming,
58 Incumbent,
59 Duplicate,
60 Collision,
61}
62
63impl CanonicalChain {
64 pub fn open(root: &Path) -> Result<Self, String> {
65 prepare_root(root)?;
66 let ordering_path = root.join("ordering.dat");
67 let store_path = root.join("k1-transaction-store");
68 let ordering_type = path_type(&ordering_path)?;
69 let store_type = path_type(&store_path)?;
70 if ordering_type.is_some_and(|kind| !kind.is_file()) {
71 return Err("ordering.dat is not a regular file".to_owned());
72 }
73 if store_type.is_some_and(|kind| !kind.is_dir()) {
74 return Err("k1-transaction-store is not a directory".to_owned());
75 }
76 let (order, store) = match (ordering_type, store_type) {
77 (None, None) => {
78 let store = TransactionStore::create(&store_path)
79 .unwrap_or_else(|error| fatal("create-transaction-store", error));
80 let order = OrderStore::create(&ordering_path)
81 .unwrap_or_else(|error| fatal("create-order-store", error));
82 (order, store)
83 }
84 (Some(_), Some(_)) => (
85 OrderStore::open(&ordering_path)?,
86 TransactionStore::open(&store_path)
87 .map_err(|error| format!("transaction store error: {error}"))?,
88 ),
89 _ => return Err("canonical chain root is incomplete".to_owned()),
90 };
91 Ok(Self { order, store })
92 }
93 pub fn submit_validated(&mut self, transaction: &[u8]) -> Result<CommitOutcome, SubmitError> {
94 let parsed = Transaction::parse(transaction)
95 .map_err(|message| SubmitError::Other(format!("invalid transaction: {message}")))?;
96 let candidate = Candidate {
97 id: TxId::for_transaction(transaction),
98 parent: parsed.parent(),
99 creator: *parsed.creator(),
100 timestamp: parsed.timestamp(),
101 subsystem: parsed.subsystem(),
102 bytes: transaction,
103 };
104 if is_reserved_id(candidate.id) {
105 return Err(SubmitError::Other(
106 "transaction ID collides with a reserved sentinel".to_owned(),
107 ));
108 }
109 self.submit_candidate(candidate)
110 }
111 pub fn submit_local<F>(
112 &mut self,
113 timestamp: u64,
114 creator: [u8; 32],
115 subsystem: SubsystemId,
116 payload: &[u8],
117 signer: F,
118 ) -> Result<Vec<u8>, String>
119 where
120 F: FnOnce(&[u8]) -> Result<[u8; 64], String>,
121 {
122 let parent = self.tip().unwrap_or(GENESIS_PARENT);
123 let bytes =
124 build_signed_transaction(parent, timestamp, creator, subsystem, payload, signer)?;
125 let id = TxId::for_transaction(&bytes);
126 if is_reserved_id(id) {
127 return Err("transaction ID collides with a reserved sentinel".to_owned());
128 }
129 if self.order.index_of(id).is_some() {
130 return Err("transaction ID collides with a canonical transaction".to_owned());
131 }
132 self.persist(&bytes, id)
133 .map_err(|error| error.to_string())?;
134 self.order
135 .commit(self.order.entries().len(), id, subsystem)
136 .unwrap_or_else(|error| fatal("commit-local-order", error));
137 Ok(bytes)
138 }
139 pub fn contains(&self, id: TxId) -> bool {
140 !is_reserved_id(id) && self.order.index_of(id).is_some()
141 }
142 pub fn tip(&self) -> Option<TxId> {
143 self.order.entries().last().map(|entry| entry.0)
144 }
145 pub fn between_txids(&self, older: TxId, newer: TxId) -> Result<Vec<TxId>, String> {
146 let older_index = if older == GENESIS_PARENT {
147 -1_i128
148 } else {
149 self.order
150 .index_of(older)
151 .map(|index| index as i128)
152 .ok_or_else(|| "older boundary is not canonical".to_owned())?
153 };
154 let newer_index = self
155 .order
156 .index_of(newer)
157 .map(|index| index as i128)
158 .ok_or_else(|| "newer boundary is not canonical".to_owned())?;
159 if older_index == newer_index {
160 return Ok(Vec::new());
161 }
162 if older_index > newer_index {
163 return Err("transaction boundaries are reversed".to_owned());
164 }
165 let interior = newer_index - older_index - 1;
166 if interior <= 128 {
167 return Ok(((older_index + 1)..newer_index)
168 .map(|index| self.order.entries()[index as usize].0)
169 .collect());
170 }
171 let distance = newer_index - older_index;
172 Ok((1_i128..=128)
173 .map(|k| self.order.entries()[(older_index + k * distance / 129) as usize].0)
174 .collect())
175 }
176 pub fn get_txn(&self, id: TxId) -> Result<Option<Vec<u8>>, String> {
177 if !self.contains(id) {
178 return Ok(None);
179 }
180 Ok(Some(self.canonical_bytes(id)))
181 }
182 pub fn replay_cursor(
183 &self,
184 subsystem: SubsystemId,
185 after: Option<TxId>,
186 ) -> Result<ReplayCursor, String> {
187 if let Some(id) = after {
188 let index = self
189 .order
190 .index_of(id)
191 .ok_or_else(|| "replay checkpoint is not canonical".to_owned())?;
192 if self.order.entries()[index].1 != subsystem {
193 return Err("replay checkpoint belongs to another subsystem".to_owned());
194 }
195 }
196 Ok(ReplayCursor { subsystem, after })
197 }
198 pub fn replay_next(
199 &self,
200 cursor: &mut ReplayCursor,
201 ) -> Result<Option<ReplayTransaction>, String> {
202 let start = match cursor.after {
203 None => 0,
204 Some(id) => {
205 self.order
206 .index_of(id)
207 .ok_or_else(|| "replay cursor is no longer canonical".to_owned())?
208 + 1
209 }
210 };
211 for &(id, subsystem) in &self.order.entries()[start..] {
212 if subsystem != cursor.subsystem {
213 continue;
214 }
215 let bytes = self.canonical_bytes(id);
216 let parsed = Transaction::parse(&bytes)
217 .unwrap_or_else(|error| fatal("parse-canonical-transaction", error));
218 if TxId::for_transaction(&bytes) != id || parsed.subsystem() != subsystem {
219 fatal(
220 "verify-canonical-transaction",
221 "canonical transaction does not match its order record",
222 );
223 }
224 cursor.after = Some(id);
225 return Ok(Some(ReplayTransaction { id, bytes }));
226 }
227 Ok(None)
228 }
229 fn submit_candidate(&mut self, candidate: Candidate<'_>) -> Result<CommitOutcome, SubmitError> {
230 if self.order.index_of(candidate.id).is_some() {
231 return if self.canonical_bytes(candidate.id) == candidate.bytes {
232 Ok(CommitOutcome::Duplicate)
233 } else {
234 Err(SubmitError::Other(
235 "transaction ID collision with canonical bytes".to_owned(),
236 ))
237 };
238 }
239 let shared_len = if candidate.parent == GENESIS_PARENT {
240 0
241 } else {
242 self.order
243 .index_of(candidate.parent)
244 .map(|index| index + 1)
245 .ok_or(SubmitError::MissingParent)?
246 };
247 if shared_len == self.order.entries().len() {
248 self.persist(candidate.bytes, candidate.id)?;
249 self.order
250 .commit(shared_len, candidate.id, candidate.subsystem)
251 .unwrap_or_else(|error| fatal("commit-extension-order", error));
252 return Ok(CommitOutcome::Extension {
253 id: candidate.id,
254 subsystem: candidate.subsystem,
255 });
256 }
257 let (incumbent_id, incumbent_subsystem) = self.order.entries()[shared_len];
258 let incumbent_bytes = self.canonical_bytes(incumbent_id);
259 let incumbent = Transaction::parse(&incumbent_bytes)
260 .unwrap_or_else(|error| fatal("parse-canonical-incumbent", error));
261 if incumbent.subsystem() != incumbent_subsystem || incumbent.parent() != candidate.parent {
262 fatal(
263 "verify-canonical-incumbent",
264 "canonical incumbent does not match its order record or parent",
265 );
266 }
267 match fork_decision(
268 &candidate.creator,
269 candidate.timestamp,
270 candidate.bytes,
271 incumbent.creator(),
272 incumbent.timestamp(),
273 &incumbent_bytes,
274 ) {
275 ForkDecision::Incumbent => {
276 return Err(SubmitError::Other(
277 "fork loses canonical ordering".to_owned(),
278 ));
279 }
280 ForkDecision::Duplicate => return Ok(CommitOutcome::Duplicate),
281 ForkDecision::Collision => {
282 return Err(SubmitError::Other(
283 "full transaction digest collision".to_owned(),
284 ));
285 }
286 ForkDecision::Incoming => {}
287 }
288 self.persist(candidate.bytes, candidate.id)?;
289 self.order
290 .commit(shared_len, candidate.id, candidate.subsystem)
291 .unwrap_or_else(|error| fatal("commit-reorganization-order", error));
292 Ok(CommitOutcome::Reorganization {
293 id: candidate.id,
294 subsystem: candidate.subsystem,
295 })
296 }
297 fn persist(&self, bytes: &[u8], expected: TxId) -> Result<(), SubmitError> {
298 match self.store.put(bytes) {
299 Ok(PutOutcome::Inserted(id)) | Ok(PutOutcome::Duplicate(id)) if id == expected => {
300 Ok(())
301 }
302 Ok(_) => fatal(
303 "persist-transaction",
304 "transaction store returned an unexpected transaction ID",
305 ),
306 Err(StoreError::IdCollision(_)) => Err(SubmitError::Other(
307 "transaction ID collision with stored bytes".to_owned(),
308 )),
309 Err(error) => fatal("persist-transaction", error),
310 }
311 }
312 fn canonical_bytes(&self, id: TxId) -> Vec<u8> {
313 match self.store.get(id) {
314 Ok(Some(bytes)) => bytes,
315 Ok(None) => fatal(
316 "load-canonical-transaction",
317 "canonical transaction bytes are missing",
318 ),
319 Err(error) => fatal("load-canonical-transaction", error),
320 }
321 }
322}
323fn is_reserved_id(id: TxId) -> bool {
324 id == GENESIS_PARENT || id == REGISTER_AT_TIP
325}
326fn prepare_root(root: &Path) -> Result<(), String> {
327 match fs::symlink_metadata(root) {
328 Ok(metadata) if metadata.file_type().is_dir() => Ok(()),
329 Ok(_) => Err("canonical chain root is not a directory".to_owned()),
330 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
331 fs::create_dir_all(root).unwrap_or_else(|error| fatal("create-root", error));
332 Ok(())
333 }
334 Err(error) => Err(format!("cannot inspect canonical chain root: {error}")),
335 }
336}
337fn path_type(path: &Path) -> Result<Option<fs::FileType>, String> {
338 match fs::symlink_metadata(path) {
339 Ok(metadata) => Ok(Some(metadata.file_type())),
340 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
341 Err(error) => Err(format!("cannot inspect canonical chain component: {error}")),
342 }
343}
344fn fork_decision(
345 incoming_creator: &[u8; 32],
346 incoming_timestamp: u64,
347 incoming_bytes: &[u8],
348 incumbent_creator: &[u8; 32],
349 incumbent_timestamp: u64,
350 incumbent_bytes: &[u8],
351) -> ForkDecision {
352 match incoming_creator.cmp(incumbent_creator) {
353 Ordering::Less => return ForkDecision::Incoming,
354 Ordering::Greater => return ForkDecision::Incumbent,
355 Ordering::Equal => {}
356 }
357 match incoming_timestamp.cmp(&incumbent_timestamp) {
358 Ordering::Less => return ForkDecision::Incoming,
359 Ordering::Greater => return ForkDecision::Incumbent,
360 Ordering::Equal => {}
361 }
362 let incoming_digest: [u8; 32] = Sha256::digest(incoming_bytes).into();
363 let incumbent_digest: [u8; 32] = Sha256::digest(incumbent_bytes).into();
364 match incoming_digest.cmp(&incumbent_digest) {
365 Ordering::Less => ForkDecision::Incoming,
366 Ordering::Greater => ForkDecision::Incumbent,
367 Ordering::Equal if incoming_bytes == incumbent_bytes => ForkDecision::Duplicate,
368 Ordering::Equal => ForkDecision::Collision,
369 }
370}
371fn fatal(operation: &str, error: impl fmt::Display) -> ! {
372 eprintln!("kcode-k1-canonical-chain fatal {operation}: {error}");
373 std::process::abort()
374}