1use std::collections::{HashMap, HashSet};
2use std::str::FromStr;
3use std::sync::{Arc, Condvar, Mutex, MutexGuard};
4
5use base64::Engine;
6use base64::engine::general_purpose::URL_SAFE_NO_PAD;
7use hmac::{Hmac, Mac};
8use kcode_k1_peering::K1Peering;
9use kcode_k1_transaction::SubsystemId;
10use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem, TxId};
11use sha2::Sha256;
12use zeroize::{Zeroize, Zeroizing};
13
14const SUBSYSTEM_BYTES: [u8; 20] = *b"k1-invites-subsystem";
15const CODE_BYTES: usize = 6;
16const COMMITMENT_BYTES: usize = 32;
17const ISSUE_BYTES: usize = 34;
18const CONSUME_BYTES: usize = 46;
19const VERSION: u8 = 1;
20const ISSUE: u8 = 1;
21const CONSUME: u8 = 2;
22const DOMAIN: &[u8] = b"k1-invite-v1";
23
24type HmacSha256 = Hmac<Sha256>;
25
26pub struct InviteCode {
27 bytes: [u8; CODE_BYTES],
28}
29
30impl InviteCode {
31 fn from_bytes(bytes: [u8; CODE_BYTES]) -> Self {
32 Self { bytes }
33 }
34
35 #[must_use]
36 pub fn expose(&self) -> String {
37 URL_SAFE_NO_PAD.encode(self.bytes)
38 }
39}
40
41impl FromStr for InviteCode {
42 type Err = String;
43
44 fn from_str(text: &str) -> Result<Self, Self::Err> {
45 if text.len() != 8 || !text.is_ascii() {
46 return Err("invite code must be exactly eight URL-safe characters".to_owned());
47 }
48 let mut decoded = URL_SAFE_NO_PAD
49 .decode(text.as_bytes())
50 .map_err(|_| "invite code is not canonical URL-safe base64".to_owned())?;
51 if decoded.len() != CODE_BYTES {
52 decoded.zeroize();
53 return Err("invite code must decode to exactly six bytes".to_owned());
54 }
55 let mut bytes = [0; CODE_BYTES];
56 bytes.copy_from_slice(&decoded);
57 decoded.zeroize();
58 if URL_SAFE_NO_PAD.encode(bytes) != text {
59 bytes.zeroize();
60 return Err("invite code is not canonical URL-safe base64".to_owned());
61 }
62 Ok(Self::from_bytes(bytes))
63 }
64}
65
66impl Drop for InviteCode {
67 fn drop(&mut self) {
68 self.bytes.zeroize();
69 }
70}
71
72pub struct InviteVerifierKey {
73 bytes: Zeroizing<[u8; 32]>,
74}
75
76impl InviteVerifierKey {
77 #[must_use]
78 pub fn from_bytes(bytes: [u8; 32]) -> Self {
79 Self {
80 bytes: Zeroizing::new(bytes),
81 }
82 }
83}
84
85#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
86pub struct UserId(TxId);
87
88impl UserId {
89 #[must_use]
90 pub fn as_tx_id(self) -> TxId {
91 self.0
92 }
93}
94
95#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
96struct Commitment([u8; COMMITMENT_BYTES]);
97
98#[derive(Clone, Copy, Debug, Eq, PartialEq)]
99enum Record {
100 Issue(Commitment),
101 Consume {
102 issue_id: TxId,
103 commitment: Commitment,
104 },
105}
106
107#[derive(Clone, Copy, Eq, PartialEq)]
108enum Phase {
109 Replaying,
110 Ready,
111 Unavailable,
112}
113
114struct Projection {
115 issue_id: TxId,
116 user_id: Option<UserId>,
117}
118
119struct State {
120 phase: Phase,
121 issued: HashMap<Commitment, Projection>,
122 pending_issues: HashSet<Commitment>,
123 pending_consumes: HashSet<Commitment>,
124 consume_failures: HashMap<Commitment, String>,
125}
126
127impl State {
128 fn new() -> Self {
129 Self {
130 phase: Phase::Replaying,
131 issued: HashMap::new(),
132 pending_issues: HashSet::new(),
133 pending_consumes: HashSet::new(),
134 consume_failures: HashMap::new(),
135 }
136 }
137
138 fn invalidate(&mut self) {
139 self.phase = Phase::Unavailable;
140 self.issued.clear();
141 self.pending_issues.clear();
142 self.pending_consumes.clear();
143 self.consume_failures.clear();
144 }
145}
146
147trait CodeGenerator: Send + Sync {
148 fn generate(&self) -> Result<[u8; CODE_BYTES], String>;
149}
150
151struct OsGenerator;
152
153impl CodeGenerator for OsGenerator {
154 fn generate(&self) -> Result<[u8; CODE_BYTES], String> {
155 let mut bytes = [0; CODE_BYTES];
156 getrandom::fill(&mut bytes)
157 .map_err(|error| format!("invite randomness unavailable: {error}"))?;
158 Ok(bytes)
159 }
160}
161
162trait Submitter: Send + Sync {
163 fn submit(&self, subsystem: SubsystemId, payload: &[u8]) -> Result<TxId, String>;
164}
165
166impl Submitter for K1Peering {
167 fn submit(&self, subsystem: SubsystemId, payload: &[u8]) -> Result<TxId, String> {
168 self.submit_txn(subsystem, payload)
169 }
170}
171
172struct InviteInner {
173 key: InviteVerifierKey,
174 submitter: Arc<dyn Submitter>,
175 generator: Arc<dyn CodeGenerator>,
176 state: Mutex<State>,
177 changed: Condvar,
178}
179
180impl InviteInner {
181 fn new(
182 key: InviteVerifierKey,
183 submitter: Arc<dyn Submitter>,
184 generator: Arc<dyn CodeGenerator>,
185 ) -> Self {
186 Self {
187 key,
188 submitter,
189 generator,
190 state: Mutex::new(State::new()),
191 changed: Condvar::new(),
192 }
193 }
194
195 fn lock(&self) -> Result<MutexGuard<'_, State>, String> {
196 self.state
197 .lock()
198 .map_err(|_| "invite subsystem synchronization unavailable".to_owned())
199 }
200
201 fn require_ready(state: &State) -> Result<(), String> {
202 (state.phase == Phase::Ready)
203 .then_some(())
204 .ok_or_else(|| "invite subsystem unavailable".to_owned())
205 }
206
207 fn finish_replay(&self) -> Result<(), String> {
208 let mut state = self.lock()?;
209 if state.phase != Phase::Replaying {
210 return Err("invite subsystem unavailable after replay".to_owned());
211 }
212 state.phase = Phase::Ready;
213 self.changed.notify_all();
214 Ok(())
215 }
216
217 fn invalidate(&self) {
218 let mut state = self
219 .state
220 .lock()
221 .unwrap_or_else(|poisoned| poisoned.into_inner());
222 state.invalidate();
223 self.changed.notify_all();
224 }
225
226 fn commitment(&self, code: &InviteCode) -> Commitment {
227 let mut mac = HmacSha256::new_from_slice(&self.key.bytes[..])
228 .expect("HMAC-SHA256 accepts a 32-byte key");
229 mac.update(DOMAIN);
230 mac.update(&code.bytes);
231 let output = mac.finalize().into_bytes();
232 let mut commitment = [0; COMMITMENT_BYTES];
233 commitment.copy_from_slice(&output);
234 Commitment(commitment)
235 }
236
237 fn create(&self) -> Result<(TxId, InviteCode), String> {
238 loop {
239 let code = InviteCode::from_bytes(self.generator.generate()?);
240 let commitment = self.commitment(&code);
241 {
242 let mut state = self.lock()?;
243 Self::require_ready(&state)?;
244 if state.issued.contains_key(&commitment)
245 || !state.pending_issues.insert(commitment)
246 {
247 continue;
248 }
249 }
250
251 let result = self
252 .submitter
253 .submit(subsystem(), &encode_issue(commitment));
254 let mut state = self.lock()?;
255 state.pending_issues.remove(&commitment);
256 self.changed.notify_all();
257 Self::require_ready(&state)?;
258 let projected = state.issued.get(&commitment).map(|entry| entry.issue_id);
259 match (result, projected) {
260 (_, Some(issue_id)) => return Ok((issue_id, code)),
261 (Err(error), None) => return Err(error),
262 (Ok(_), None) => {
263 return Err("invite issue acknowledgement was not projected".to_owned());
264 }
265 }
266 }
267 }
268
269 fn consume(&self, code: &InviteCode) -> Result<UserId, String> {
270 let commitment = self.commitment(code);
271 let issue_id = loop {
272 let mut state = self.lock()?;
273 Self::require_ready(&state)?;
274 let Some(entry) = state.issued.get(&commitment) else {
275 return Err("invite is unavailable".to_owned());
276 };
277 if let Some(user_id) = entry.user_id {
278 return Ok(user_id);
279 }
280 let issue_id = entry.issue_id;
281 if let Some(error) = state.consume_failures.get(&commitment) {
282 return Err(error.clone());
283 }
284 if state.pending_consumes.insert(commitment) {
285 break issue_id;
286 }
287 state = self
288 .changed
289 .wait(state)
290 .map_err(|_| "invite subsystem synchronization unavailable".to_owned())?;
291 drop(state);
292 };
293
294 let result = self
295 .submitter
296 .submit(subsystem(), &encode_consume(issue_id, commitment));
297 let mut state = self.lock()?;
298 state.pending_consumes.remove(&commitment);
299 self.changed.notify_all();
300 Self::require_ready(&state)?;
301 let projected = state
302 .issued
303 .get(&commitment)
304 .filter(|entry| entry.issue_id == issue_id)
305 .and_then(|entry| entry.user_id);
306 if let Some(user_id) = projected {
307 return Ok(user_id);
308 }
309 let error = match result {
310 Ok(_) => "invite consume acknowledgement was not projected".to_owned(),
311 Err(error) => error,
312 };
313 state.consume_failures.insert(commitment, error.clone());
314 Err(error)
315 }
316
317 fn apply(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
318 let record = match decode(payload) {
319 Ok(record) => record,
320 Err(error) => {
321 self.invalidate();
322 return Err(error);
323 }
324 };
325 let mut state = self.lock()?;
326 if state.phase == Phase::Unavailable {
327 return Err("invite subsystem unavailable".to_owned());
328 }
329 match record {
330 Record::Issue(commitment) => {
331 state.issued.entry(commitment).or_insert(Projection {
332 issue_id: id,
333 user_id: None,
334 });
335 }
336 Record::Consume {
337 issue_id,
338 commitment,
339 } => {
340 if let Some(entry) = state.issued.get_mut(&commitment)
341 && entry.issue_id == issue_id
342 && entry.user_id.is_none()
343 {
344 entry.user_id = Some(UserId(id));
345 state.consume_failures.remove(&commitment);
346 }
347 }
348 }
349 self.changed.notify_all();
350 Ok(())
351 }
352}
353
354impl Subsystem for InviteInner {
355 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
356 self.apply(id, payload)
357 }
358
359 fn reorg(&self) -> Result<(), String> {
360 self.invalidate();
361 Ok(())
362 }
363}
364
365pub struct K1Invites {
366 inner: Arc<InviteInner>,
367}
368
369impl K1Invites {
370 pub fn open(
371 ordering: Arc<K1TxnOrdering>,
372 peering: Arc<K1Peering>,
373 verifier_key: InviteVerifierKey,
374 ) -> Result<Self, String> {
375 let submitter: Arc<dyn Submitter> = peering;
376 let inner = Arc::new(InviteInner::new(
377 verifier_key,
378 submitter,
379 Arc::new(OsGenerator),
380 ));
381 let handler: Arc<dyn Subsystem> = inner.clone();
382 if let Err(error) = ordering.register_subsystem(subsystem(), None, handler) {
383 inner.invalidate();
384 return Err(error);
385 }
386 inner.finish_replay()?;
387 Ok(Self { inner })
388 }
389
390 pub fn create(&self) -> Result<(TxId, InviteCode), String> {
391 self.inner.create()
392 }
393
394 pub fn consume(&self, code: &InviteCode) -> Result<UserId, String> {
395 self.inner.consume(code)
396 }
397}
398
399fn subsystem() -> SubsystemId {
400 SubsystemId::from_bytes(SUBSYSTEM_BYTES)
401 .expect("invite subsystem ID is valid fixed-width UTF-8")
402}
403
404fn encode_issue(commitment: Commitment) -> [u8; ISSUE_BYTES] {
405 let mut payload = [0; ISSUE_BYTES];
406 payload[0] = VERSION;
407 payload[1] = ISSUE;
408 payload[2..].copy_from_slice(&commitment.0);
409 payload
410}
411
412fn encode_consume(issue_id: TxId, commitment: Commitment) -> [u8; CONSUME_BYTES] {
413 let mut payload = [0; CONSUME_BYTES];
414 payload[0] = VERSION;
415 payload[1] = CONSUME;
416 payload[2..14].copy_from_slice(issue_id.as_bytes());
417 payload[14..].copy_from_slice(&commitment.0);
418 payload
419}
420
421fn decode(payload: &[u8]) -> Result<Record, String> {
422 if payload.len() < 2 {
423 return Err("malformed invite transaction header".to_owned());
424 }
425 if payload[0] != VERSION {
426 return Err("unsupported invite transaction version".to_owned());
427 }
428 match payload[1] {
429 ISSUE if payload.len() == ISSUE_BYTES => {
430 let mut commitment = [0; COMMITMENT_BYTES];
431 commitment.copy_from_slice(&payload[2..]);
432 Ok(Record::Issue(Commitment(commitment)))
433 }
434 ISSUE => Err("malformed invite issue transaction".to_owned()),
435 CONSUME if payload.len() == CONSUME_BYTES => {
436 let mut issue_id = [0; 12];
437 issue_id.copy_from_slice(&payload[2..14]);
438 let mut commitment = [0; COMMITMENT_BYTES];
439 commitment.copy_from_slice(&payload[14..]);
440 Ok(Record::Consume {
441 issue_id: TxId::from_bytes(issue_id),
442 commitment: Commitment(commitment),
443 })
444 }
445 CONSUME => Err("malformed invite consume transaction".to_owned()),
446 _ => Err("unknown invite transaction kind".to_owned()),
447 }
448}
449
450#[cfg(test)]
451mod tests {
452 use super::*;
453 use kcode_k1_transaction::Transaction;
454 use std::collections::VecDeque;
455 use std::fs;
456 use std::path::PathBuf;
457 use std::sync::Barrier;
458 use std::sync::Weak;
459 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
460 use std::thread;
461
462 const KEY: u8 = 0x42;
463 static NEXT_ROOT: AtomicU64 = AtomicU64::new(0);
464
465 struct SequenceGenerator(Mutex<VecDeque<[u8; CODE_BYTES]>>);
466
467 impl SequenceGenerator {
468 fn new(values: Vec<[u8; CODE_BYTES]>) -> Self {
469 Self(Mutex::new(values.into()))
470 }
471 }
472
473 impl CodeGenerator for SequenceGenerator {
474 fn generate(&self) -> Result<[u8; CODE_BYTES], String> {
475 self.0
476 .lock()
477 .map_err(|_| "test generator unavailable".to_owned())?
478 .pop_front()
479 .ok_or_else(|| "test generator exhausted".to_owned())
480 }
481 }
482
483 #[derive(Clone)]
484 struct Committed {
485 id: TxId,
486 payload: Vec<u8>,
487 }
488
489 struct FakePeer {
490 next: AtomicU64,
491 attempts: AtomicUsize,
492 committed: Mutex<Vec<Committed>>,
493 handler: Mutex<Weak<InviteInner>>,
494 fail_before: AtomicBool,
495 fail_after: AtomicBool,
496 }
497
498 impl FakePeer {
499 fn new() -> Self {
500 Self {
501 next: AtomicU64::new(1),
502 attempts: AtomicUsize::new(0),
503 committed: Mutex::new(Vec::new()),
504 handler: Mutex::new(Weak::new()),
505 fail_before: AtomicBool::new(false),
506 fail_after: AtomicBool::new(false),
507 }
508 }
509
510 fn attach(&self, inner: &Arc<InviteInner>) {
511 *self.handler.lock().expect("handler lock") = Arc::downgrade(inner);
512 }
513
514 fn committed(&self) -> Vec<Committed> {
515 self.committed.lock().expect("commit lock").clone()
516 }
517 }
518
519 impl Submitter for FakePeer {
520 fn submit(&self, id: SubsystemId, payload: &[u8]) -> Result<TxId, String> {
521 assert_eq!(id, subsystem());
522 self.attempts.fetch_add(1, Ordering::SeqCst);
523 if self.fail_before.swap(false, Ordering::SeqCst) {
524 return Err("injected peering error".to_owned());
525 }
526 let transaction_id = tx_id(self.next.fetch_add(1, Ordering::SeqCst));
527 self.committed.lock().expect("commit lock").push(Committed {
528 id: transaction_id,
529 payload: payload.to_vec(),
530 });
531 self.handler
532 .lock()
533 .expect("handler lock")
534 .upgrade()
535 .ok_or_else(|| "test handler unavailable".to_owned())?
536 .apply(transaction_id, payload)?;
537 if self.fail_after.swap(false, Ordering::SeqCst) {
538 return Err("injected post-commit error".to_owned());
539 }
540 Ok(transaction_id)
541 }
542 }
543
544 fn tx_id(value: u64) -> TxId {
545 let mut bytes = [0; 12];
546 bytes[4..].copy_from_slice(&value.to_be_bytes());
547 TxId::from_bytes(bytes)
548 }
549
550 fn test_invites(peer: Arc<FakePeer>, key: u8, codes: Vec<[u8; CODE_BYTES]>) -> K1Invites {
551 let submitter: Arc<dyn Submitter> = peer.clone();
552 let inner = Arc::new(InviteInner::new(
553 InviteVerifierKey::from_bytes([key; 32]),
554 submitter,
555 Arc::new(SequenceGenerator::new(codes)),
556 ));
557 peer.attach(&inner);
558 inner.finish_replay().unwrap();
559 K1Invites { inner }
560 }
561
562 fn replay(peer: Arc<FakePeer>, key: u8) -> K1Invites {
563 let invites = test_invites(peer.clone(), key, Vec::new());
564 {
565 let mut state = invites.inner.lock().unwrap();
566 state.phase = Phase::Replaying;
567 }
568 for record in peer.committed() {
569 invites.inner.apply(record.id, &record.payload).unwrap();
570 }
571 invites.inner.finish_replay().unwrap();
572 invites
573 }
574
575 #[test]
576 fn strict_code_and_wire_codecs() {
577 let code = InviteCode::from_bytes([0, 1, 2, 3, 4, 5]);
578 let text = code.expose();
579 assert_eq!(text, "AAECAwQF");
580 assert_eq!(text.parse::<InviteCode>().unwrap().bytes, code.bytes);
581 for bad in [
582 "",
583 "AAAAAAA",
584 "AAAAAAAAA",
585 "AAAAAAA=",
586 "AAAAAAA!",
587 "////////",
588 ] {
589 assert!(bad.parse::<InviteCode>().is_err());
590 }
591
592 let commitment = Commitment([7; COMMITMENT_BYTES]);
593 let issue_id = tx_id(9);
594 assert_eq!(
595 decode(&encode_issue(commitment)),
596 Ok(Record::Issue(commitment))
597 );
598 assert_eq!(
599 decode(&encode_consume(issue_id, commitment)),
600 Ok(Record::Consume {
601 issue_id,
602 commitment,
603 })
604 );
605 let mut malformed = vec![vec![], vec![VERSION], vec![2, ISSUE], vec![VERSION, 9]];
606 malformed.push(encode_issue(commitment)[..ISSUE_BYTES - 1].to_vec());
607 malformed.push([encode_issue(commitment).as_slice(), &[0]].concat());
608 malformed.push(encode_consume(issue_id, commitment)[..CONSUME_BYTES - 1].to_vec());
609 malformed.push([encode_consume(issue_id, commitment).as_slice(), &[0]].concat());
610 assert!(malformed.iter().all(|payload| decode(payload).is_err()));
611 }
612
613 #[test]
614 fn creation_collision_consumption_and_wrong_key_are_deterministic() {
615 let peer = Arc::new(FakePeer::new());
616 let first = [1, 2, 3, 4, 5, 6];
617 let replacement = [7, 8, 9, 10, 11, 12];
618 let invites = test_invites(peer.clone(), KEY, vec![first, first, replacement]);
619 let (first_id, first_code) = invites.create().unwrap();
620 let (second_id, second_code) = invites.create().unwrap();
621 assert_ne!(first_id, second_id);
622 assert_eq!(first_code.bytes, first);
623 assert_eq!(second_code.bytes, replacement);
624 assert_eq!(peer.committed().len(), 2);
625
626 let first_payload = &peer.committed()[0].payload;
627 assert_eq!(
628 first_payload,
629 &encode_issue(invites.inner.commitment(&first_code))
630 );
631 assert!(!first_payload.windows(CODE_BYTES).any(|part| part == first));
632
633 let wrong = InviteCode::from_bytes([99; CODE_BYTES]);
634 assert!(invites.consume(&wrong).is_err());
635 assert_eq!(peer.committed().len(), 2);
636
637 let user = invites.consume(&first_code).unwrap();
638 assert_eq!(user.as_tx_id(), peer.committed()[2].id);
639 let attempts = peer.attempts.load(Ordering::SeqCst);
640 assert_eq!(invites.consume(&first_code).unwrap(), user);
641 assert_eq!(peer.attempts.load(Ordering::SeqCst), attempts);
642
643 let wrong_key = replay(peer.clone(), KEY + 1);
644 assert!(wrong_key.consume(&second_code).is_err());
645 }
646
647 #[test]
648 fn concurrent_consumers_share_one_commit_and_errors_are_not_retried() {
649 let peer = Arc::new(FakePeer::new());
650 let invites = Arc::new(test_invites(peer.clone(), KEY, vec![[3; CODE_BYTES]]));
651 peer.fail_after.store(true, Ordering::SeqCst);
652 let (_, code) = invites.create().unwrap();
653 let text = code.expose();
654 let barrier = Arc::new(Barrier::new(12));
655 let handles: Vec<_> = (0..12)
656 .map(|_| {
657 let invites = invites.clone();
658 let barrier = barrier.clone();
659 let text = text.clone();
660 thread::spawn(move || {
661 let code: InviteCode = text.parse().unwrap();
662 barrier.wait();
663 invites.consume(&code)
664 })
665 })
666 .collect();
667 let users: Vec<_> = handles
668 .into_iter()
669 .map(|handle| handle.join().unwrap().unwrap())
670 .collect();
671 assert!(users.iter().all(|user| *user == users[0]));
672 assert_eq!(peer.committed().len(), 2);
673
674 let failed_peer = Arc::new(FakePeer::new());
675 let failed = test_invites(failed_peer.clone(), KEY, vec![[4; CODE_BYTES]]);
676 let (_, failed_code) = failed.create().unwrap();
677 failed_peer.fail_before.store(true, Ordering::SeqCst);
678 let before = failed_peer.attempts.load(Ordering::SeqCst);
679 let error = failed.consume(&failed_code).unwrap_err();
680 assert_eq!(failed.consume(&failed_code).unwrap_err(), error);
681 assert_eq!(failed_peer.attempts.load(Ordering::SeqCst), before + 1);
682 }
683
684 #[test]
685 fn replay_and_reorg_are_fail_closed() {
686 let peer = Arc::new(FakePeer::new());
687 let invites = test_invites(peer.clone(), KEY, vec![[5; CODE_BYTES]]);
688 let (_, code) = invites.create().unwrap();
689 let user = invites.consume(&code).unwrap();
690 let restarted = replay(peer.clone(), KEY);
691 assert_eq!(restarted.consume(&code).unwrap(), user);
692 let before = peer.committed().len();
693 restarted.inner.reorg().unwrap();
694 assert!(restarted.consume(&code).is_err());
695 assert!(restarted.create().is_err());
696 assert_eq!(peer.committed().len(), before);
697 let reopened = replay(peer, KEY);
698 assert_eq!(reopened.consume(&code).unwrap(), user);
699 }
700
701 struct TempRoots(PathBuf);
702
703 impl TempRoots {
704 fn new(label: &str) -> Self {
705 let number = NEXT_ROOT.fetch_add(1, Ordering::Relaxed);
706 let root = std::env::temp_dir().join(format!(
707 "kcode-k1-invites-{}-{number}-{label}",
708 std::process::id()
709 ));
710 let _ = fs::remove_dir_all(&root);
711 Self(root)
712 }
713
714 fn ordering(&self) -> PathBuf {
715 self.0.join("ordering")
716 }
717
718 fn peering(&self) -> PathBuf {
719 self.0.join("peering")
720 }
721 }
722
723 impl Drop for TempRoots {
724 fn drop(&mut self) {
725 let _ = fs::remove_dir_all(&self.0);
726 }
727 }
728
729 fn open_real(roots: &TempRoots) -> (Arc<K1TxnOrdering>, Arc<K1Peering>, K1Invites) {
730 let ordering = Arc::new(K1TxnOrdering::open(&roots.ordering()).unwrap());
731 let peering = Arc::new(K1Peering::open(&roots.peering(), ordering.clone()).unwrap());
732 let invites = K1Invites::open(
733 ordering.clone(),
734 peering.clone(),
735 InviteVerifierKey::from_bytes([KEY; 32]),
736 )
737 .unwrap();
738 (ordering, peering, invites)
739 }
740
741 #[test]
742 fn real_stack_receipts_restart_and_persistence_boundary() {
743 let roots = TempRoots::new("restart");
744 let (ordering, peering, invites) = open_real(&roots);
745 let (issue_id, code) = invites.create().unwrap();
746 assert_eq!(ordering.tip(), Some(issue_id));
747 let bytes = ordering.get_txn(issue_id).unwrap().unwrap();
748 let transaction = Transaction::parse(&bytes).unwrap();
749 assert_eq!(transaction.subsystem(), subsystem());
750 assert_eq!(
751 decode(transaction.payload()),
752 Ok(Record::Issue(invites.inner.commitment(&code)))
753 );
754 let user = invites.consume(&code).unwrap();
755 assert_eq!(ordering.tip(), Some(user.as_tx_id()));
756 let tip = ordering.tip();
757 assert_eq!(invites.consume(&code).unwrap(), user);
758 assert_eq!(ordering.tip(), tip);
759 let text = code.expose();
760 drop(invites);
761 drop(peering);
762 drop(ordering);
763
764 let (ordering, peering, invites) = open_real(&roots);
765 let code: InviteCode = text.parse().unwrap();
766 assert_eq!(invites.consume(&code).unwrap(), user);
767 assert_eq!(ordering.tip(), tip);
768 let mut entries: Vec<_> = fs::read_dir(&roots.0)
769 .unwrap()
770 .map(|entry| entry.unwrap().file_name())
771 .collect();
772 entries.sort();
773 assert_eq!(entries, vec!["ordering", "peering"]);
774 drop(invites);
775 drop(peering);
776 drop(ordering);
777 }
778
779 struct Noop;
780
781 impl Subsystem for Noop {
782 fn submit_txn(&self, _id: TxId, _payload: &[u8]) -> Result<(), String> {
783 Ok(())
784 }
785
786 fn reorg(&self) -> Result<(), String> {
787 Ok(())
788 }
789 }
790
791 #[test]
792 fn malformed_canonical_payload_faults_only_invites() {
793 let roots = TempRoots::new("isolation");
794 let (ordering, peering, invites) = open_real(&roots);
795 let other = SubsystemId::from_bytes([b'o'; 20]).unwrap();
796 ordering
797 .register_subsystem(other, None, Arc::new(Noop))
798 .unwrap();
799 let result = ordering.submit_local_txn(
800 1,
801 [9; 32],
802 subsystem(),
803 &[VERSION],
804 |_| Ok([9; 64]),
805 |_| Ok(()),
806 );
807 assert!(result.is_err());
808 assert!(invites.create().is_err());
809 let other_id = peering.submit_txn(other, b"still available").unwrap();
810 assert_eq!(ordering.tip(), Some(other_id));
811 }
812}