kcode_k1_txn_ordering/
lib.rs

1use kcode_k1_canonical_chain::{CanonicalChain, CommitOutcome};
2pub use kcode_k1_canonical_chain::{SubmitError, TxId};
3use kcode_k1_transaction::Transaction;
4pub use kcode_k1_transaction::{GENESIS_PARENT, SubsystemId};
5use std::collections::HashMap;
6use std::panic::{AssertUnwindSafe, catch_unwind};
7use std::path::Path;
8use std::sync::atomic::{AtomicBool, Ordering};
9use std::sync::{Arc, Condvar, Mutex, MutexGuard};
10use std::thread;
11use std::time::{Duration, Instant};
12
13pub trait Subsystem: Send + Sync + 'static {
14    fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String>;
15    fn reorg(&self) -> Result<(), String>;
16}
17
18pub struct K1TxnOrdering {
19    writer: Mutex<WriterState>,
20    chain: Mutex<CanonicalChain>,
21}
22
23struct WriterState {
24    registrations: HashMap<SubsystemId, Registration>,
25}
26
27struct Registration {
28    lane: Arc<Lane>,
29    latest: Option<TxId>,
30    next_ticket: u64,
31    mode: RegistrationMode,
32}
33
34#[derive(Clone, Copy, Eq, PartialEq)]
35enum RegistrationMode {
36    Replaying,
37    Active,
38    OutOfCommission,
39}
40
41struct Lane {
42    handler: Arc<dyn Subsystem>,
43    progress: Mutex<LaneProgress>,
44    changed: Condvar,
45    running: AtomicBool,
46    faulted: AtomicBool,
47}
48
49struct LaneProgress {
50    serving: u64,
51    fault: Option<String>,
52}
53
54struct Reservation {
55    lane: Arc<Lane>,
56    ticket: u64,
57}
58
59enum CommittedWork {
60    Duplicate,
61    Extension {
62        id: TxId,
63        delivery: Option<Reservation>,
64    },
65    Reorganization {
66        id: TxId,
67        reorgs: Vec<Reservation>,
68        replacement: Option<Reservation>,
69    },
70}
71
72impl K1TxnOrdering {
73    pub fn open(root: &Path) -> Result<Self, String> {
74        let started = Instant::now();
75        let result = CanonicalChain::open(root);
76        let elapsed = started.elapsed();
77        if elapsed > Duration::from_millis(100) {
78            eprintln!(
79                "{{\"module\":\"kcode-k1-txn-ordering\",\"operation\":\"open\",\"elapsed_microseconds\":{},\"outcome\":\"{}\"}}",
80                elapsed.as_micros(),
81                if result.is_ok() { "ready" } else { "error" }
82            );
83        }
84        result.map(|chain| Self {
85            writer: Mutex::new(WriterState {
86                registrations: HashMap::new(),
87            }),
88            chain: Mutex::new(chain),
89        })
90    }
91
92    pub fn register_subsystem(
93        &self,
94        subsystem: SubsystemId,
95        after: Option<TxId>,
96        handler: Arc<dyn Subsystem>,
97    ) -> Result<(), String> {
98        let (mut cursor, lane) = {
99            let mut writer = self.lock_writer();
100            if let Some(registration) = writer.registrations.get(&subsystem) {
101                ensure_replaceable(registration)?;
102            }
103            let chain = self.lock_chain();
104            let cursor = chain.replay_cursor(subsystem, after)?;
105            let lane = Arc::new(Lane::new(handler));
106            writer.registrations.insert(
107                subsystem,
108                Registration {
109                    lane: lane.clone(),
110                    latest: after,
111                    next_ticket: 0,
112                    mode: RegistrationMode::Replaying,
113                },
114            );
115            (cursor, lane)
116        };
117        let mut latest = after;
118        loop {
119            let mut next = match self.lock_chain().replay_next(&mut cursor) {
120                Ok(next) => next,
121                Err(message) => return Err(fault_cursor(&lane, message)),
122            };
123            if next.is_none() {
124                let final_result = {
125                    let mut writer = self.lock_writer();
126                    let chain = self.lock_chain();
127                    match chain.replay_next(&mut cursor) {
128                        Ok(None) => {
129                            let registration = writer
130                                .registrations
131                                .get_mut(&subsystem)
132                                .expect("replaying registration exists");
133                            if !Arc::ptr_eq(&registration.lane, &lane) {
134                                return Err("registration changed during replay".to_owned());
135                            }
136                            registration.latest = latest;
137                            registration.mode = RegistrationMode::Active;
138                            return Ok(());
139                        }
140                        result => result,
141                    }
142                };
143                next = match final_result {
144                    Ok(next) => next,
145                    Err(message) => return Err(fault_cursor(&lane, message)),
146                };
147            }
148            let replayed = next.expect("replay result contains a transaction");
149            let transaction = match Transaction::parse(&replayed.bytes) {
150                Ok(transaction) => transaction,
151                Err(message) => {
152                    let failure = committed_error(
153                        replayed.id,
154                        vec![format!(
155                            "canonical replay transaction was invalid: {message}"
156                        )],
157                    );
158                    lane.record_fault(failure.clone());
159                    return Err(failure);
160                }
161            };
162            if let Err(message) = lane.run_replay(replayed.id, transaction.payload()) {
163                return Err(committed_error(replayed.id, vec![message]));
164            }
165            latest = Some(replayed.id);
166        }
167    }
168
169    pub fn submit_txn(&self, transaction: &[u8]) -> Result<(), SubmitError> {
170        let parsed = Transaction::parse(transaction)
171            .map_err(|message| SubmitError::Other(format!("invalid transaction: {message}")))?;
172        let payload = parsed.payload();
173        let work = {
174            let mut writer = self.lock_writer();
175            let mut chain = self.lock_chain();
176            match chain.submit_validated(transaction)? {
177                CommitOutcome::Duplicate => CommittedWork::Duplicate,
178                CommitOutcome::Extension { id, subsystem } => CommittedWork::Extension {
179                    id,
180                    delivery: reserve_active_delivery(&mut writer, subsystem, id),
181                },
182                CommitOutcome::Reorganization { id, subsystem } => {
183                    let affected: Vec<SubsystemId> = writer
184                        .registrations
185                        .iter()
186                        .filter_map(|(&registered, registration)| {
187                            (registration.is_active()
188                                && registration
189                                    .latest
190                                    .is_some_and(|latest| !chain.contains(latest)))
191                            .then_some(registered)
192                        })
193                        .collect();
194                    let mut reorgs = Vec::with_capacity(affected.len());
195                    for affected_subsystem in affected {
196                        let registration = writer
197                            .registrations
198                            .get_mut(&affected_subsystem)
199                            .expect("affected registration exists");
200                        reorgs.push(registration.reserve_reorg());
201                        registration.mode = RegistrationMode::OutOfCommission;
202                    }
203                    let replacement = reserve_active_delivery(&mut writer, subsystem, id);
204                    CommittedWork::Reorganization {
205                        id,
206                        reorgs,
207                        replacement,
208                    }
209                }
210            }
211        };
212        match work {
213            CommittedWork::Duplicate => Ok(()),
214            CommittedWork::Extension { id, delivery } => {
215                let Some(delivery) = delivery else {
216                    return Ok(());
217                };
218                delivery
219                    .run_delivery(id, payload)
220                    .map_err(|message| SubmitError::Other(committed_error(id, vec![message])))
221            }
222            CommittedWork::Reorganization {
223                id,
224                reorgs,
225                replacement,
226            } => {
227                let failures = run_reorganization(id, payload, reorgs, replacement);
228                if failures.is_empty() {
229                    Ok(())
230                } else {
231                    Err(SubmitError::Other(committed_error(id, failures)))
232                }
233            }
234        }
235    }
236
237    pub fn submit_local_txn<F, Q>(
238        &self,
239        timestamp: u64,
240        creator: [u8; 32],
241        subsystem: SubsystemId,
242        payload: &[u8],
243        signer: F,
244        queue_propagation: Q,
245    ) -> Result<Vec<u8>, String>
246    where
247        F: FnOnce(&[u8]) -> Result<[u8; 64], String>,
248        Q: FnOnce(&[u8]) -> Result<(), String>,
249    {
250        let (bytes, id, delivery) = {
251            let mut writer = self.lock_writer();
252            let lane = writer
253                .registrations
254                .get(&subsystem)
255                .filter(|registration| registration.is_active())
256                .map(|registration| registration.lane.clone())
257                .ok_or_else(|| "target subsystem is not registered and active".to_owned())?;
258            let mut chain = self.lock_chain();
259            let bytes =
260                chain.submit_local(timestamp, creator, subsystem, payload, move |prefix| {
261                    match catch_unwind(AssertUnwindSafe(|| signer(prefix))) {
262                        Ok(result) => result,
263                        Err(_) => Err("signer panicked".to_owned()),
264                    }
265                })?;
266            let id = TxId::for_transaction(&bytes);
267            let registration = writer
268                .registrations
269                .get_mut(&subsystem)
270                .expect("prechecked registration exists");
271            if !Arc::ptr_eq(&registration.lane, &lane) {
272                return Err(committed_error(
273                    id,
274                    vec!["target registration changed during local commit".to_owned()],
275                ));
276            }
277            let delivery = registration.reserve_delivery(id);
278            (bytes, id, delivery)
279        };
280        let mut failures = Vec::new();
281        match catch_unwind(AssertUnwindSafe(|| queue_propagation(&bytes))) {
282            Ok(Ok(())) => {}
283            Ok(Err(message)) => failures.push(format!("queue propagation failed: {message}")),
284            Err(_) => failures.push("queue propagation panicked".to_owned()),
285        }
286        if let Err(message) = delivery.run_delivery(id, payload) {
287            failures.push(message);
288        }
289        if failures.is_empty() {
290            Ok(bytes)
291        } else {
292            Err(committed_error(id, failures))
293        }
294    }
295
296    pub fn contains(&self, id: TxId) -> bool {
297        self.lock_chain().contains(id)
298    }
299
300    pub fn tip(&self) -> Option<TxId> {
301        self.lock_chain().tip()
302    }
303
304    pub fn between_txids(&self, older: TxId, newer: TxId) -> Result<Vec<TxId>, String> {
305        self.lock_chain().between_txids(older, newer)
306    }
307
308    pub fn get_txn(&self, id: TxId) -> Result<Option<Vec<u8>>, String> {
309        self.lock_chain().get_txn(id)
310    }
311
312    fn lock_writer(&self) -> MutexGuard<'_, WriterState> {
313        self.writer.lock().expect("KTO writer mutex poisoned")
314    }
315
316    fn lock_chain(&self) -> MutexGuard<'_, CanonicalChain> {
317        self.chain.lock().expect("KTO chain mutex poisoned")
318    }
319}
320
321impl Registration {
322    fn is_active(&self) -> bool {
323        self.mode == RegistrationMode::Active && !self.lane.faulted.load(Ordering::Acquire)
324    }
325
326    fn reserve_delivery(&mut self, id: TxId) -> Reservation {
327        let reservation = Reservation {
328            lane: self.lane.clone(),
329            ticket: self.next_ticket,
330        };
331        self.next_ticket += 1;
332        self.latest = Some(id);
333        reservation
334    }
335
336    fn reserve_reorg(&mut self) -> Reservation {
337        let reservation = Reservation {
338            lane: self.lane.clone(),
339            ticket: self.next_ticket,
340        };
341        self.next_ticket += 1;
342        reservation
343    }
344}
345
346impl Lane {
347    fn new(handler: Arc<dyn Subsystem>) -> Self {
348        Self {
349            handler,
350            progress: Mutex::new(LaneProgress {
351                serving: 0,
352                fault: None,
353            }),
354            changed: Condvar::new(),
355            running: AtomicBool::new(false),
356            faulted: AtomicBool::new(false),
357        }
358    }
359
360    fn run_replay(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
361        self.running.store(true, Ordering::Release);
362        let result = invoke_callback("replay submit", || self.handler.submit_txn(id, payload));
363        if let Err(message) = &result {
364            self.record_fault(message.clone());
365        }
366        self.running.store(false, Ordering::Release);
367        result
368    }
369
370    fn run_ticket<F>(&self, ticket: u64, operation: &str, callback: F) -> Result<(), String>
371    where
372        F: FnOnce() -> Result<(), String>,
373    {
374        let mut progress = self.progress.lock().expect("subsystem lane mutex poisoned");
375        while progress.serving < ticket {
376            progress = self
377                .changed
378                .wait(progress)
379                .expect("subsystem lane mutex poisoned");
380        }
381        if progress.serving > ticket {
382            return Err(format!("{operation} ticket was already completed"));
383        }
384        if let Some(reason) = progress.fault.clone() {
385            progress.serving += 1;
386            self.changed.notify_all();
387            return Err(format!(
388                "{operation} callback was not run because the registration faulted: {reason}"
389            ));
390        }
391        self.running.store(true, Ordering::Release);
392        drop(progress);
393        let result = invoke_callback(operation, callback);
394        let mut progress = self.progress.lock().expect("subsystem lane mutex poisoned");
395        if let Err(message) = &result
396            && progress.fault.is_none()
397        {
398            progress.fault = Some(message.clone());
399            self.faulted.store(true, Ordering::Release);
400        }
401        progress.serving += 1;
402        self.running.store(false, Ordering::Release);
403        self.changed.notify_all();
404        result
405    }
406
407    fn record_fault(&self, message: String) {
408        let mut progress = self.progress.lock().expect("subsystem lane mutex poisoned");
409        if progress.fault.is_none() {
410            progress.fault = Some(message);
411            self.faulted.store(true, Ordering::Release);
412        }
413    }
414
415    fn is_quiescent(&self, next_ticket: u64) -> bool {
416        let progress = self.progress.lock().expect("subsystem lane mutex poisoned");
417        progress.serving == next_ticket && !self.running.load(Ordering::Acquire)
418    }
419}
420
421impl Reservation {
422    fn run_delivery(self, id: TxId, payload: &[u8]) -> Result<(), String> {
423        self.lane.run_ticket(self.ticket, "submit", || {
424            self.lane.handler.submit_txn(id, payload)
425        })
426    }
427
428    fn run_reorg(self) -> Result<(), String> {
429        self.lane
430            .run_ticket(self.ticket, "reorg", || self.lane.handler.reorg())
431    }
432}
433
434fn invoke_callback<F>(operation: &str, callback: F) -> Result<(), String>
435where
436    F: FnOnce() -> Result<(), String>,
437{
438    match catch_unwind(AssertUnwindSafe(callback)) {
439        Ok(Ok(())) => Ok(()),
440        Ok(Err(message)) => Err(format!("{operation} callback failed: {message}")),
441        Err(_) => Err(format!("{operation} callback panicked")),
442    }
443}
444
445fn ensure_replaceable(registration: &Registration) -> Result<(), String> {
446    if registration.mode == RegistrationMode::Replaying
447        && !registration.lane.faulted.load(Ordering::Acquire)
448    {
449        return Err("subsystem registration is replaying".to_owned());
450    }
451    if registration.mode == RegistrationMode::Active
452        && !registration.lane.faulted.load(Ordering::Acquire)
453    {
454        return Err("subsystem is already registered and active".to_owned());
455    }
456    if !registration.lane.is_quiescent(registration.next_ticket) {
457        return Err("previous subsystem lane work has not quiesced".to_owned());
458    }
459    Ok(())
460}
461
462fn reserve_active_delivery(
463    writer: &mut WriterState,
464    subsystem: SubsystemId,
465    id: TxId,
466) -> Option<Reservation> {
467    writer
468        .registrations
469        .get_mut(&subsystem)
470        .filter(|registration| registration.is_active())
471        .map(|registration| registration.reserve_delivery(id))
472}
473
474fn run_reorganization(
475    id: TxId,
476    payload: &[u8],
477    reorgs: Vec<Reservation>,
478    replacement: Option<Reservation>,
479) -> Vec<String> {
480    thread::scope(|scope| {
481        let mut jobs = Vec::with_capacity(reorgs.len() + usize::from(replacement.is_some()));
482        for reservation in reorgs {
483            jobs.push(scope.spawn(move || reservation.run_reorg()));
484        }
485        if let Some(reservation) = replacement {
486            jobs.push(scope.spawn(move || reservation.run_delivery(id, payload)));
487        }
488        let mut failures = Vec::new();
489        for job in jobs {
490            match job.join() {
491                Ok(Ok(())) => {}
492                Ok(Err(message)) => failures.push(message),
493                Err(_) => failures.push("callback lane panicked".to_owned()),
494            }
495        }
496        failures
497    })
498}
499
500fn fault_cursor(lane: &Lane, message: String) -> String {
501    let failure = format!("subsystem replay cursor failed: {message}");
502    lane.record_fault(failure.clone());
503    failure
504}
505
506fn committed_error(id: TxId, failures: Vec<String>) -> String {
507    format!("TxId {id:?} was committed; {}", failures.join("; "))
508}
509
510#[cfg(test)]
511mod tests {
512    use super::*;
513    use kcode_k1_txn_ordering_testkit::{
514        OrderingCandidate, OrderingHarness, TestQueuePropagation, TestSigner, TestSubsystem,
515        verify_callback_failure_isolation, verify_independent_subsystem_lanes,
516        verify_queue_before_callback, verify_reorganization_isolation, verify_replay_live_handoff,
517        verify_restart_duplicates_and_queries, verify_signing_commit_exclusion,
518        verify_unrelated_callback_reentry,
519    };
520    use std::path::Path;
521
522    struct Harness;
523
524    struct TestSubsystemAdapter {
525        inner: Arc<dyn TestSubsystem>,
526    }
527
528    impl Subsystem for TestSubsystemAdapter {
529        fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
530            self.inner.submit_txn(id, payload)
531        }
532
533        fn reorg(&self) -> Result<(), String> {
534            self.inner.reorg()
535        }
536    }
537
538    impl OrderingCandidate for K1TxnOrdering {
539        fn register_subsystem(
540            &self,
541            subsystem: SubsystemId,
542            after: Option<TxId>,
543            handler: Arc<dyn TestSubsystem>,
544        ) -> Result<(), String> {
545            K1TxnOrdering::register_subsystem(
546                self,
547                subsystem,
548                after,
549                Arc::new(TestSubsystemAdapter { inner: handler }),
550            )
551        }
552
553        fn submit_txn(&self, transaction: &[u8]) -> Result<(), SubmitError> {
554            K1TxnOrdering::submit_txn(self, transaction)
555        }
556
557        fn submit_local_txn(
558            &self,
559            timestamp: u64,
560            creator: [u8; 32],
561            subsystem: SubsystemId,
562            payload: &[u8],
563            signer: TestSigner,
564            queue_propagation: TestQueuePropagation,
565        ) -> Result<Vec<u8>, String> {
566            K1TxnOrdering::submit_local_txn(
567                self,
568                timestamp,
569                creator,
570                subsystem,
571                payload,
572                signer,
573                queue_propagation,
574            )
575        }
576
577        fn contains(&self, id: TxId) -> bool {
578            K1TxnOrdering::contains(self, id)
579        }
580
581        fn tip(&self) -> Option<TxId> {
582            K1TxnOrdering::tip(self)
583        }
584
585        fn between_txids(&self, older: TxId, newer: TxId) -> Result<Vec<TxId>, String> {
586            K1TxnOrdering::between_txids(self, older, newer)
587        }
588
589        fn get_txn(&self, id: TxId) -> Result<Option<Vec<u8>>, String> {
590            K1TxnOrdering::get_txn(self, id)
591        }
592    }
593
594    impl OrderingHarness for Harness {
595        fn open(&self, root: &Path) -> Result<Arc<dyn OrderingCandidate>, String> {
596            Ok(Arc::new(K1TxnOrdering::open(root)?))
597        }
598    }
599
600    #[test]
601    fn queue_before_callback() {
602        verify_queue_before_callback(&Harness).unwrap();
603    }
604
605    #[test]
606    fn independent_subsystem_lanes() {
607        verify_independent_subsystem_lanes(&Harness).unwrap();
608    }
609
610    #[test]
611    fn signing_commit_exclusion() {
612        verify_signing_commit_exclusion(&Harness).unwrap();
613    }
614
615    #[test]
616    fn replay_live_handoff() {
617        verify_replay_live_handoff(&Harness).unwrap();
618    }
619
620    #[test]
621    fn callback_failure_isolation() {
622        verify_callback_failure_isolation(&Harness).unwrap();
623    }
624
625    #[test]
626    fn reorganization_isolation() {
627        verify_reorganization_isolation(&Harness).unwrap();
628    }
629
630    #[test]
631    fn unrelated_callback_reentry() {
632        verify_unrelated_callback_reentry(&Harness).unwrap();
633    }
634
635    #[test]
636    fn restart_duplicates_and_queries() {
637        verify_restart_duplicates_and_queries(&Harness).unwrap();
638    }
639}
kcode_k1_txn_ordering/lib.rs

kcode_k1_txn_ordering/
lib.rs
