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

kcode_k1_txn_ordering/
lib.rs
