kcode_k1_txn_ordering/
lib.rs

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

kcode_k1_txn_ordering/
lib.rs
