planarian_txn/
fake.rs

1//! In-memory fake backends for the txn-kernel harness (catalog `harness = "fake"`).
2//!
3//! One `FakeWorld` is shared by every `FakeHost`, the same way the daemon and the
4//! watchdog share one machine. Fencing reads the epochs in the shared `MemStore`,
5//! as real backends read the journal's. Every mutating call is logged, so tests
6//! can assert that calls never overlap and that stale owners were refused.
7
8use crate::kernel::{CheckPhase, Host, Verdict};
9use crate::store::{MemStore, Store};
10use planarian_schema::{
11    Built, DeviceInfo, DriverRef, EvaluatedGeneration, GenerationBackend, GenerationId, Hash, InventoryBackend, JobId,
12    Millis, Observed, OpId, OverrideSet, PinnedSource, Resources, StagedSource, StorePath, Timestamp, Transaction,
13    Transition, UnitId,
14};
15use std::collections::{BTreeMap, BTreeSet};
16use std::sync::{Arc, Mutex, MutexGuard};
17
18#[derive(Clone, Debug)]
19pub struct FakeJob {
20    pub op: OpId,
21    pub done_at: Timestamp,
22    pub cancellable: bool,
23    /// The generation that becomes current when the job completes.
24    pub activates: Option<GenerationId>,
25}
26
27#[derive(Clone, Debug, PartialEq, Eq)]
28pub struct CallLog {
29    pub at: Timestamp,
30    pub op: OpId,
31    pub call: String,
32    pub fenced: bool,
33    /// Supervisor jobs pending when the call was admitted.
34    pub pending: usize,
35}
36
37#[derive(Debug, Default)]
38pub struct World {
39    pub now: Timestamp,
40    pub current: GenerationId,
41    pub boot_default: GenerationId,
42    pub boot_entries: BTreeSet<GenerationId>,
43    pub generations: BTreeMap<GenerationId, StorePath>,
44    /// Generations whose probes fail.
45    pub bad: BTreeSet<GenerationId>,
46    /// Every probe fails (an environmental fault coinciding with a change).
47    pub env_fault: bool,
48    pub override_hash: Hash,
49    pub units: BTreeMap<UnitId, Observed>,
50    pub devices: BTreeMap<UnitId, Option<DriverRef>>,
51    /// Probes fail while any device is bound to one of these drivers.
52    pub bad_drivers: BTreeSet<String>,
53    pub jobs: BTreeMap<JobId, FakeJob>,
54    pub next_job: u64,
55    /// `activate_live` accepts the job but never replies; the job runs for this long.
56    pub slow_activation: Option<(Millis, bool)>,
57    /// `activate_live` of these generations fails outright.
58    pub fail_activate: BTreeSet<GenerationId>,
59    pub fail_publish: bool,
60    pub jobs_tagged_unknown: bool,
61    pub calls: Vec<CallLog>,
62    pub alive: BTreeSet<u32>,
63    pub unkillable: BTreeSet<u32>,
64    pub gc_roots: BTreeSet<String>,
65}
66
67impl World {
68    fn tick(&mut self) {
69        let now = self.now;
70        let done: Vec<JobId> = self.jobs.iter().filter(|(_, j)| j.done_at <= now).map(|(id, _)| id.clone()).collect();
71        for id in done {
72            let job = self.jobs.remove(&id).expect("listed");
73            if let Some(g) = job.activates {
74                self.current = g;
75            }
76        }
77    }
78}
79
80#[derive(Clone, Debug)]
81pub struct FakeHost {
82    pub world: Arc<Mutex<World>>,
83    pub store: MemStore,
84}
85
86impl FakeHost {
87    pub fn new(world: Arc<Mutex<World>>, store: MemStore) -> FakeHost {
88        FakeHost { world, store }
89    }
90
91    pub fn w(&self) -> MutexGuard<'_, World> {
92        let mut w = self.world.lock().unwrap_or_else(|e| e.into_inner());
93        w.tick();
94        w
95    }
96
97    /// Fence and log a mutating call. Returns false if the call must be refused.
98    fn admit(&self, w: &mut World, op: &OpId, call: &str) -> bool {
99        let current = self.store.fence(op.txn).ok().flatten();
100        let fenced = !crate::kernel::fence_admits(current, op.epoch);
101        let pending = w.jobs.len();
102        w.calls.push(CallLog { at: w.now, op: *op, call: call.into(), fenced, pending });
103        !fenced
104    }
105}
106
107use planarian_schema::Outcome;
108
109impl GenerationBackend for FakeHost {
110    fn current(&self) -> Result<GenerationId, Outcome> {
111        Ok(self.w().current)
112    }
113
114    fn boot_entries(&self) -> Result<Vec<GenerationId>, Outcome> {
115        Ok(self.w().boot_entries.iter().copied().collect())
116    }
117
118    fn boot_default(&self) -> Result<GenerationId, Outcome> {
119        Ok(self.w().boot_default)
120    }
121
122    fn last_booted(&self) -> Result<GenerationId, Outcome> {
123        Ok(self.w().boot_default)
124    }
125
126    fn stage(&self, _owner: &PinnedSource, _overrides: &OverrideSet) -> Result<StagedSource, Outcome> {
127        Err(Outcome::Failed("fake: staging is exec's concern".into()))
128    }
129
130    fn source_hash(&self, _owner: &PinnedSource) -> Result<Hash, Outcome> {
131        Err(Outcome::Failed("fake: staging is exec's concern".into()))
132    }
133
134    fn evaluate(&self, _src: &StagedSource) -> Result<EvaluatedGeneration, Outcome> {
135        Err(Outcome::Failed("fake: evaluation is exec's concern".into()))
136    }
137
138    fn build(&self, _eval: &EvaluatedGeneration) -> Result<Built, Outcome> {
139        Err(Outcome::Failed("fake: building is exec's concern".into()))
140    }
141
142    fn verify_candidate(&self, path: &StorePath, _nar: Option<&Hash>) -> Result<bool, Outcome> {
143        Ok(self.w().generations.values().any(|p| p == path))
144    }
145
146    fn pending_jobs(&self) -> Result<Vec<JobId>, Outcome> {
147        Ok(self.w().jobs.keys().cloned().collect())
148    }
149
150    fn jobs_tagged(&self, op: &OpId) -> Result<Option<Vec<JobId>>, Outcome> {
151        let w = self.w();
152        if w.jobs_tagged_unknown {
153            return Ok(None);
154        }
155        Ok(Some(w.jobs.iter().filter(|(_, j)| j.op == *op).map(|(id, _)| id.clone()).collect()))
156    }
157
158    fn override_hash(&self) -> Result<Hash, Outcome> {
159        Ok(self.w().override_hash)
160    }
161
162    fn activate_live(&self, g: GenerationId, op: &OpId) -> Outcome {
163        let mut w = self.w();
164        if !self.admit(&mut w, op, &format!("activate_live({})", g.0)) {
165            return Outcome::fenced();
166        }
167        if w.fail_activate.contains(&g) {
168            return Outcome::Failed(format!("activation of {} failed (injected)", g.0));
169        }
170        if let Some((delay, cancellable)) = w.slow_activation {
171            w.next_job += 1;
172            let id = JobId(format!("job-{}", w.next_job));
173            let done_at = w.now.plus(delay);
174            w.jobs.insert(id, FakeJob { op: *op, done_at, cancellable, activates: Some(g) });
175            return Outcome::Unknown("job accepted; reply withheld".into());
176        }
177        w.current = g;
178        Outcome::Succeeded
179    }
180
181    fn publish_boot(&self, g: GenerationId, op: &OpId) -> Outcome {
182        let mut w = self.w();
183        if !self.admit(&mut w, op, &format!("publish_boot({})", g.0)) {
184            return Outcome::fenced();
185        }
186        if w.fail_publish {
187            return Outcome::Failed("ENOSPC on /boot (injected)".into());
188        }
189        w.boot_entries.insert(g);
190        w.boot_default = g;
191        Outcome::Succeeded
192    }
193
194    fn protect(&self, g: GenerationId, op: &OpId) -> Outcome {
195        let mut w = self.w();
196        if !self.admit(&mut w, op, &format!("protect({})", g.0)) {
197            return Outcome::fenced();
198        }
199        Outcome::Succeeded
200    }
201
202    fn cancel(&self, job: &JobId, op: &OpId) -> Outcome {
203        let mut w = self.w();
204        if !self.admit(&mut w, op, &format!("cancel({})", job.0)) {
205            return Outcome::fenced();
206        }
207        match w.jobs.get(job) {
208            Some(j) if j.cancellable => {
209                w.jobs.remove(job);
210                Outcome::Succeeded
211            }
212            Some(_) => Outcome::Failed("job cannot be cancelled".into()),
213            None => Outcome::Succeeded,
214        }
215    }
216
217    fn swap_override(&self, from: &Hash, to: &Hash, op: &OpId) -> Outcome {
218        let mut w = self.w();
219        if !self.admit(&mut w, op, "swap_override") {
220            return Outcome::fenced();
221        }
222        if w.override_hash == *to {
223            return Outcome::Succeeded;
224        }
225        if w.override_hash != *from {
226            return Outcome::Failed("override changed by a third party".into());
227        }
228        w.override_hash = *to;
229        Outcome::Succeeded
230    }
231
232    fn path_of(&self, g: GenerationId) -> Result<StorePath, Outcome> {
233        self.w().generations.get(&g).cloned().ok_or_else(|| Outcome::Failed(format!("no generation {}", g.0)))
234    }
235
236    fn evaluate_built(&self, _path: &StorePath) -> Result<EvaluatedGeneration, Outcome> {
237        Err(Outcome::Failed("fake: evaluation is exec's concern".into()))
238    }
239
240    fn read_overrides(&self) -> Result<(OverrideSet, Hash), Outcome> {
241        Ok((OverrideSet::default(), self.w().override_hash))
242    }
243
244    fn remember_override(&self, set: &OverrideSet) -> Result<Hash, Outcome> {
245        Ok(Hash::of(&serde_json::to_vec(set).expect("serializes")))
246    }
247
248    fn override_for(&self, _g: GenerationId) -> Result<Option<Hash>, Outcome> {
249        Ok(None)
250    }
251
252    fn record_override(&self, _g: GenerationId, _hash: &Hash) -> Result<(), Outcome> {
253        Ok(())
254    }
255
256    fn release_root(&self, root: &str) -> bool {
257        self.w().gc_roots.remove(root)
258    }
259
260    fn collect_garbage(&self, _protect: &[GenerationId], _emergency: bool, op: &OpId) -> Outcome {
261        let mut w = self.w();
262        if !self.admit(&mut w, op, "collect_garbage") {
263            return Outcome::fenced();
264        }
265        Outcome::Succeeded
266    }
267
268    fn gc_plan(&self, _protect: &[GenerationId], _emergency: bool) -> Result<planarian_schema::gc::GcPlan, Outcome> {
269        Ok(planarian_schema::gc::plan(&[], &[], 0, 0, |_| 0))
270    }
271
272    fn space_report(&self) -> Option<planarian_schema::gc::SpaceReport> {
273        None
274    }
275}
276
277impl InventoryBackend for FakeHost {
278    fn devices(&self) -> Result<Vec<DeviceInfo>, Outcome> {
279        Ok(self.w().devices.iter().map(|(id, d)| DeviceInfo { id: id.clone(), driver: d.clone() }).collect())
280    }
281
282    fn resources(&self) -> Result<Resources, Outcome> {
283        Ok(Resources::default())
284    }
285
286    fn bound_driver(&self, device: &UnitId) -> Result<Option<DriverRef>, Outcome> {
287        Ok(self.w().devices.get(device).cloned().flatten())
288    }
289
290    fn bind(&self, device: &UnitId, driver: &DriverRef, op: &OpId) -> Outcome {
291        let mut w = self.w();
292        if !self.admit(&mut w, op, &format!("bind({device}, {})", driver.name)) {
293            return Outcome::fenced();
294        }
295        w.devices.insert(device.clone(), Some(driver.clone()));
296        Outcome::Succeeded
297    }
298
299    fn unbind(&self, device: &UnitId, op: &OpId) -> Outcome {
300        let mut w = self.w();
301        if !self.admit(&mut w, op, &format!("unbind({device})")) {
302            return Outcome::fenced();
303        }
304        w.devices.insert(device.clone(), None);
305        Outcome::Succeeded
306    }
307}
308
309impl Host for FakeHost {
310    type Gen = FakeHost;
311    type Inv = FakeHost;
312
313    fn generations(&self) -> &FakeHost {
314        self
315    }
316
317    fn inventory(&self) -> &FakeHost {
318        self
319    }
320
321    fn execute_runtime(&self, t: &Transition, op: &OpId) -> Outcome {
322        let mut w = self.w();
323        let name = match t {
324            Transition::RestartUnit(u) => format!("restart({u})"),
325            Transition::StopUnit(u) => format!("stop({u})"),
326            other => format!("{other:?}"),
327        };
328        if !self.admit(&mut w, op, &name) {
329            return Outcome::fenced();
330        }
331        match t {
332            Transition::RestartUnit(u) => {
333                w.units.insert(u.clone(), Observed::Active);
334            }
335            Transition::StopUnit(u) => {
336                w.units.insert(u.clone(), Observed::Bound);
337            }
338            _ => {}
339        }
340        Outcome::Succeeded
341    }
342
343    fn reobserve(&self, _txn: &Transaction) -> Outcome {
344        Outcome::Succeeded
345    }
346
347    fn check(&self, _txn: &Transaction, _phase: CheckPhase) -> Verdict {
348        let w = self.w();
349        if w.env_fault {
350            return Verdict::Fail("canonical probe failed (environment)".into());
351        }
352        if w.bad.contains(&w.current) {
353            return Verdict::Fail(format!("probes fail on generation {}", w.current.0));
354        }
355        if let Some((dev, d)) =
356            w.devices.iter().find(|(_, d)| d.as_ref().is_some_and(|d| w.bad_drivers.contains(&d.name)))
357        {
358            return Verdict::Fail(format!("{dev} bound to {}", d.as_ref().expect("checked").name));
359        }
360        Verdict::Pass
361    }
362
363    fn release_gc_root(&self, path: &str) -> bool {
364        self.w().gc_roots.remove(path)
365    }
366
367    fn now(&self) -> Timestamp {
368        self.w().now
369    }
370
371    fn boot_now(&self) -> Timestamp {
372        self.w().now
373    }
374
375    fn sleep(&self, d: Millis) {
376        let mut w = self.w();
377        w.now = w.now.plus(d);
378        w.tick();
379    }
380
381    fn kill_worker(&self, pid: u32) -> bool {
382        let mut w = self.w();
383        if w.unkillable.contains(&pid) {
384            return false;
385        }
386        w.alive.remove(&pid);
387        true
388    }
389
390    fn pid_alive(&self, pid: u32) -> bool {
391        self.w().alive.contains(&pid)
392    }
393}
planarian_txn/fake.rs

planarian_txn/
fake.rs
