1//! Commits whose WAL write or sync fails, through a shim over the platform's VFS. Its own
2//! binary, as the shim becomes the process's default VFS.
3#![cfg(unix)]
4
5use notebook::{
6 Replica,
7 session::{Event, Section},
8};
9use onestore::op::{Edit, Op, PageOp};
10use std::{path::Path, sync::atomic::Ordering, sync::mpsc, time::Duration};
11
12#[path = "support/vfs.rs"]
13mod vfs;
14use vfs::{
15 ARMED, KEEP_WAL, OPEN, SYNC_AFTER, SYNC_BEFORE, WAL_WRITES_AFTER_FAULT, WRITE_COMMIT, install,
16};
17
18fn colour(space: onestore::ExGuid, colour: u32) -> Edit {
19 Edit {
20 at: 133_000_000_000_000_000 + u64::from(colour),
21 ops: vec![Op::Page {
22 space,
23 op: PageOp::Color(Some(colour)),
24 }],
25 }
26}
27
28/// What the section answered for the edit committed under `fault` and the one after it,
29/// whether a failed write was reported, and the queue as the section saw it.
30struct Outcome {
31 replies: [Result<u64, String>; 2],
32 failed: bool,
33 pending: Option<Vec<u64>>,
34}
35
36/// Seeds the WAL with one edit, so the armed commit's sync is the only one it makes, then
37/// commits under `fault` and once more after it; with `crash`, leaves the cache open.
38fn scenario(root: &Path, fault: u8, crash: bool) -> Outcome {
39 install();
40 let replica = Replica::open(root.join("replica.sqlite")).unwrap();
41 let section = Section::resume(root.join("Fault.one"), replica, true, || {}).unwrap();
42 let space = section.pages().unwrap()[0].0;
43 let apply = |value| {
44 let (sender, receiver) = mpsc::channel();
45 section
46 .apply_with("Fixture", colour(space, value), move |result| {
47 sender.send(result).unwrap();
48 })
49 .unwrap();
50 receiver
51 .recv_timeout(Duration::from_secs(10))
52 .unwrap()
53 .map_err(|error| error.to_string())
54 };
55 apply(1).unwrap();
56 WAL_WRITES_AFTER_FAULT.store(0, Ordering::SeqCst);
57 ARMED.store(fault, Ordering::SeqCst);
58 let first = apply(2);
59 assert_eq!(ARMED.load(Ordering::SeqCst), 0, "fault {fault} never fired");
60 assert_eq!(
61 WAL_WRITES_AFTER_FAULT.load(Ordering::SeqCst),
62 0,
63 "the warm connection wrote frames after its commit failed"
64 );
65 let second = apply(3);
66 let failed = section
67 .events()
68 .iter()
69 .any(|event| matches!(event, Event::Failed(_)));
70 let pending = section
71 .pending()
72 .ok()
73 .map(|pending| pending.iter().map(|edit| edit.id).collect());
74 if crash {
75 std::mem::forget(section);
76 } else {
77 section.close().unwrap();
78 }
79 Outcome {
80 replies: [first, second],
81 failed,
82 pending,
83 }
84}
85
86#[test]
87fn commit_fault_child() {
88 let (Some(root), Some(fault)) = (
89 std::env::var_os("COMMIT_FAULT_CHILD"),
90 std::env::var("COMMIT_FAULT").ok(),
91 ) else {
92 return;
93 };
94 let root = Path::new(&root);
95 let outcome = scenario(root, fault.parse().unwrap(), true);
96 let ids = |result: &Result<u64, String>| result.as_ref().map_or(0, |id| *id);
97 std::fs::write(
98 root.join("outcome"),
99 format!(
100 "{} {} {}",
101 ids(&outcome.replies[0]),
102 ids(&outcome.replies[1]),
103 outcome.failed
104 ),
105 )
106 .unwrap();
107 // Ends as a crash would: no SQLite destructor checkpoints or removes the WAL.
108 std::process::exit(0);
109}
110
111#[test]
112fn a_failed_commit_is_decided_by_a_cold_read_and_never_replayed_twice() {
113 install();
114 // (fault, whether the edit reached the cache, whether the cache stays closed)
115 for (fault, landed, closed) in [
116 (SYNC_BEFORE, false, false),
117 (SYNC_AFTER, false, false),
118 (WRITE_COMMIT, false, false),
119 (SYNC_AFTER | KEEP_WAL, true, false),
120 (SYNC_BEFORE | OPEN, false, true),
121 ] {
122 for crash in [false, true] {
123 let directory = tempfile::tempdir().unwrap();
124 let root = directory.path();
125 let source = onestore::create_section("Fault.one", "Original", "Fixture").unwrap();
126 std::fs::write(root.join("Fault.one"), &source).unwrap();
127 drop(Replica::create(root.join("replica.sqlite"), &source).unwrap());
128 let (first, second, failed) = if crash {
129 let output = std::process::Command::new(std::env::current_exe().unwrap())
130 .args(["--exact", "commit_fault_child", "--nocapture"])
131 .env("COMMIT_FAULT_CHILD", root)
132 .env("COMMIT_FAULT", fault.to_string())
133 .output()
134 .unwrap();
135 assert!(output.status.success(), "{output:?}");
136 let outcome = std::fs::read_to_string(root.join("outcome")).unwrap();
137 let mut fields = outcome.split(' ');
138 let mut id = || Some(fields.next()?.parse::<u64>().unwrap()).filter(|id| *id > 0);
139 let (first, second) = (id(), id());
140 (first, second, fields.next() == Some("true"))
141 } else {
142 let outcome = scenario(root, fault, false);
143 let [first, second] = outcome.replies;
144 if !closed {
145 let ids = [first.clone().unwrap(), second.clone().unwrap()];
146 assert_eq!(&outcome.pending.unwrap()[1..], ids);
147 } else {
148 assert!(outcome.pending.is_none());
149 }
150 (first.ok(), second.ok(), outcome.failed)
151 };
152 assert_eq!(failed, !landed, "fault {fault}");
153 let cold = Replica::open(root.join("replica.sqlite")).unwrap();
154 let colours: Vec<_> = cold
155 .pending()
156 .unwrap()
157 .iter()
158 .map(|edit| match &edit.edit.ops[..] {
159 [
160 Op::Page {
161 op: PageOp::Color(Some(colour)),
162 ..
163 },
164 ] => (edit.id, *colour),
165 ops => panic!("{ops:?}"),
166 })
167 .collect();
168 if closed {
169 assert_eq!((first, second), (None, None));
170 assert_eq!(colours.len(), 1, "fault {fault}");
171 } else {
172 let (first, second) = (first.unwrap(), second.unwrap());
173 assert_eq!(&colours[1..], [(first, 2), (second, 3)], "fault {fault}");
174 }
175 }
176 }
177}