planarian_daemon/
renew.rs

1//! Lease renewal (§12): while this daemon is busy in a long backend call, a
2//! thread renews the leases it holds, so the watchdog does not take over
3//! from a live daemon. It keeps one journal handle and catches up on what
4//! others appended (`sync_ledger`), rather than opening the journal again
5//! each time: opening reads the checkpoint and the whole log, and after a
6//! compaction the rest of the store, so renewing would cost the journal's
7//! size every 20 s and be slowest when the journal is fullest.
8
9use planarian_schema::{OwnerId, Timestamp, TxnId};
10use planarian_txn::journal::{Journal, JournalError};
11use planarian_txn::ledger::{Ledger, Record};
12use planarian_txn::store::Store;
13
14/// A lease to renew: the transaction, its epoch and its worker.
15pub type Due = (TxnId, u64, Option<u32>);
16
17/// The leases `me` holds on transactions mid-flight (past their last
18/// pre-mutation state), each with the epoch and worker its lease has.
19pub fn due(l: &Ledger, me: &OwnerId) -> Vec<Due> {
20    l.live
21        .values()
22        .filter(|t| t.txn.lease.owner == *me && !t.txn.state.is_pre_mutation())
23        .map(|t| (t.txn.id, t.txn.lease.epoch, t.txn.lease.worker_pid))
24        .collect()
25}
26
27/// A `Renew` for each lease in `due`. A record refused (a takeover won the
28/// race since `due` was read) costs only itself; the others are renewed.
29/// Returns how many were.
30pub fn renew_due<S: Store>(j: &mut Journal<S>, due: &[Due], now: Timestamp, boot_at: Timestamp) -> usize {
31    due.iter()
32        .filter(|(txn, epoch, worker_pid)| {
33            j.append(now, Record::Renew { txn: *txn, epoch: *epoch, worker_pid: *worker_pid, boot_at }).is_ok()
34        })
35        .count()
36}
37
38/// One renewal: catch up, then renew what `me` holds mid-flight. An error
39/// catching up leaves the handle to be dropped and opened again.
40pub fn renew<S: Store>(
41    j: &mut Journal<S>,
42    me: &OwnerId,
43    now: Timestamp,
44    boot_at: Timestamp,
45) -> Result<usize, JournalError> {
46    let d = due(j.sync_ledger()?, me);
47    Ok(renew_due(j, &d, now, boot_at))
48}
49
50#[cfg(test)]
51mod tests {
52    use super::*;
53    use planarian_schema::*;
54    use planarian_txn::store::MemStore;
55    use std::io;
56    use std::sync::Arc;
57    use std::sync::atomic::{AtomicU64, Ordering};
58
59    /// A MemStore that counts the log bytes read through it.
60    #[derive(Clone)]
61    struct Counting(MemStore, Arc<AtomicU64>);
62
63    impl Store for Counting {
64        type Guard = <MemStore as Store>::Guard;
65        fn lock(&self) -> io::Result<Self::Guard> {
66            self.0.lock()
67        }
68        fn capacity(&self) -> u64 {
69            self.0.capacity()
70        }
71        fn read_at(&self, off: u64, len: usize) -> io::Result<Vec<u8>> {
72            let b = self.0.read_at(off, len)?;
73            self.1.fetch_add(b.len() as u64, Ordering::Relaxed);
74            Ok(b)
75        }
76        fn write_at(&self, off: u64, bytes: &[u8]) -> io::Result<()> {
77            self.0.write_at(off, bytes)
78        }
79        fn sync(&self) -> io::Result<()> {
80            self.0.sync()
81        }
82        fn read_checkpoint(&self) -> io::Result<Option<Vec<u8>>> {
83            self.0.read_checkpoint()
84        }
85        fn write_checkpoint(&self, bytes: &[u8]) -> io::Result<()> {
86            self.0.write_checkpoint(bytes)
87        }
88        fn read_sentinel(&self) -> io::Result<Option<Vec<u8>>> {
89            self.0.read_sentinel()
90        }
91        fn write_sentinel(&self, bytes: &[u8]) -> io::Result<()> {
92            self.0.write_sentinel(bytes)
93        }
94        fn fence(&self, txn: TxnId) -> io::Result<Option<u64>> {
95            self.0.fence(txn)
96        }
97        fn set_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()> {
98            self.0.set_fence(txn, epoch)
99        }
100        fn create_fence(&self, txn: TxnId, epoch: u64) -> io::Result<()> {
101            self.0.create_fence(txn, epoch)
102        }
103        fn forget_fences<F: Fn(TxnId) -> bool>(&self, keep: F) -> io::Result<Vec<TxnId>> {
104            self.0.forget_fences(keep)
105        }
106    }
107
108    use proptest::prelude::*;
109
110    proptest! {
111        #![proptest_config(ProptestConfig { cases: 32, ..ProptestConfig::default() })]
112
113        /// A renewal with a kept handle costs what others appended since the
114        /// last one, not the journal's size, before and after compactions.
115        /// Only the first renewal after a compaction pays once for the store
116        /// (it reloads: the checkpoint, then one bounded search of the
117        /// leftovers, #54). Before the scan read the body only for a header
118        /// that could be the next record, every catch-up after a compaction
119        /// read the rest of the store: the leftovers' "length".
120        #[test]
121        fn a_renewal_costs_what_was_appended(rounds in proptest::collection::vec((1u64..120, proptest::bool::weighted(0.25)), 1..10)) {
122            let capacity = 256 << 10;
123            let mem = MemStore::new(capacity);
124            let mut daemon = Journal::open(mem.clone()).unwrap();
125            let read = Arc::new(AtomicU64::new(0));
126            let mut r = Journal::open(Counting(mem.clone(), read.clone())).unwrap();
127            let me = OwnerId::Daemon { pid: 1, boot_id: planarian_schema::Hash::ZERO };
128            let mut i = 0;
129            for (n, compact) in rounds {
130                let before = daemon.offset();
131                let mut compacted = false;
132                for _ in 0..n {
133                    if daemon.needs_compaction() {
134                        daemon.compact().unwrap();
135                        compacted = true;
136                    }
137                    daemon.append(Timestamp(i), Record::KnownGood(GenerationId(i + 1))).unwrap();
138                    i += 1;
139                }
140                if compact {
141                    daemon.compact().unwrap();
142                    compacted = true;
143                }
144                let appended = if compacted { daemon.offset() } else { daemon.offset() - before };
145                read.store(0, Ordering::Relaxed);
146                renew(&mut r, &me, Timestamp(0), Timestamp(0)).unwrap();
147                prop_assert_eq!(r.ledger().last_seq, daemon.ledger().last_seq);
148                let cost = read.load(Ordering::Relaxed);
149                // Each frame's header is read twice (to size it, then whole).
150                let bound = 2 * appended + 1024 + if compacted { capacity } else { 0 };
151                prop_assert!(cost <= bound, "renewal read {} bytes; {} appended since (compacted: {})", cost, appended, compacted);
152            }
153        }
154    }
155
156    /// A transaction leased to `owner` (exec's shape, nothing the renewer
157    /// reads but the lease and the state).
158    fn txn(id: TxnId, owner: OwnerId, worker_pid: Option<u32>) -> Transaction {
159        let transition = Transition::RestartUnit(UnitId::service("gym-web"));
160        Transaction {
161            id,
162            state: TxnState::Prepared,
163            lease: Lease { txn: id, owner, epoch: 1, worker_pid, renewed: Timestamp(0), jobs: vec![] },
164            base_source: None,
165            staged_source: None,
166            candidate_nar: None,
167            base_override: None,
168            candidate_override: None,
169            gc_root: None,
170            recovery: transition.recovery(),
171            transition,
172            facts: DerivedFacts {
173                affected_units: Vec::new(),
174                sensitivity: Sensitivities::default(),
175                disruption: Disruptions::of(Disruption::Brief),
176                recovery: Recovery::Idempotent,
177                touches_manifest: false,
178                touches_canonical_probe: false,
179                changes_desired: false,
180                below_security_floor: false,
181                restores_unattributed_stop: false,
182                restarts_failed_unit: false,
183                removes_sensitivity: Default::default(),
184                removes: Default::default(),
185                removes_each: Default::default(),
186                restarts_held_unit: false,
187                other_sessions_affected: false,
188                affected_sessions: Vec::new(),
189                touches_core: false,
190                caller: Actor::Daemon,
191                config_diff: None,
192            },
193            base: GenerationId(1),
194            candidate: None,
195            policy: PolicyVersion(Hash::ZERO),
196            approval: None,
197            snapshot: None,
198            expected: Expectation {
199                units: Default::default(),
200                generation: None,
201                may_change: Default::default(),
202                others_unchanged: true,
203                settle: Settle::Duration(Millis::secs(15)),
204            },
205            started: Timestamp(1_000_000),
206            deadline: Timestamp(1_000_000 + 15 * 60 * 1000),
207            revert_to: GenerationId(1),
208            prior_driver: None,
209            check: CheckContext::default(),
210            prepared_again_from: None,
211        }
212    }
213
214    /// One round: new transactions (owner 0 me, 1 another daemon, 2 the
215    /// watchdog; mid-flight or not; a worker), some earlier ones advanced
216    /// past their last pre-mutation state, maybe a compaction, then a
217    /// renewal; and which of this daemon's due leases the watchdog takes
218    /// over between reading what is due and renewing.
219    type Round = (Vec<(u8, bool, Option<u32>)>, Vec<prop::sample::Index>, bool, Vec<prop::sample::Index>);
220
221    fn round() -> impl Strategy<Value = Round> {
222        (
223            proptest::collection::vec((0u8..3, any::<bool>(), proptest::option::of(1u32..9999)), 0..5),
224            proptest::collection::vec(any::<prop::sample::Index>(), 0..3),
225            proptest::bool::weighted(0.2),
226            proptest::collection::vec(any::<prop::sample::Index>(), 0..2),
227        )
228    }
229
230    proptest! {
231        #![proptest_config(ProptestConfig { cases: 64, ..ProptestConfig::default() })]
232
233        /// The coordinator's (#58, after prop-auditor's): after each
234        /// renewal, exactly the leases this daemon holds on transactions
235        /// mid-flight are fresh, each with its own epoch and worker; not
236        /// another daemon's, not the watchdog's, not one before its first
237        /// mutation; whatever other handles appended and compacted in
238        /// between. A lease taken over between reading what is due and
239        /// renewing it refuses only its own record. (Renewing every live
240        /// lease, or none, fails it.)
241        #[test]
242        fn exactly_my_mid_flight_leases_are_renewed(rounds in proptest::collection::vec(round(), 1..6)) {
243            let me = OwnerId::Daemon { pid: 100, boot_id: Hash::ZERO };
244            let owners = [me.clone(), OwnerId::Daemon { pid: 200, boot_id: Hash::ZERO }, OwnerId::Watchdog { pid: 300, boot_id: Hash::ZERO }];
245            let mem = MemStore::new(1 << 20);
246            let mut w = Journal::open(mem.clone()).unwrap();
247            let mut r = Journal::open(mem.clone()).unwrap();
248            let mut t = 10;
249            for (fresh, advance, compact, races) in rounds {
250                t += 10;
251                for (owner, mid, worker) in fresh {
252                    let id = TxnId(w.ledger().next_txn.max(1));
253                    w.append(Timestamp(t), Record::Prepared(Box::new(txn(id, owners[owner as usize].clone(), worker)))).unwrap();
254                    if mid {
255                        w.append(Timestamp(t), Record::Advance { txn: id, epoch: 1, to: TxnState::Activating, note: String::new(), withdrawn_for: None }).unwrap();
256                    }
257                }
258                // Advance some transactions still before their first mutation.
259                let waiting: Vec<(TxnId, u64)> = w.ledger().live.values().filter(|l| l.txn.state.is_pre_mutation()).map(|l| (l.txn.id, l.txn.lease.epoch)).collect();
260                for ix in advance {
261                    if waiting.is_empty() { break; }
262                    let (id, epoch) = waiting[ix.index(waiting.len())];
263                    if w.ledger().live.get(&id).is_some_and(|l| l.txn.state.is_pre_mutation()) {
264                        w.append(Timestamp(t), Record::Advance { txn: id, epoch, to: TxnState::Activating, note: String::new(), withdrawn_for: None }).unwrap();
265                    }
266                }
267                if compact {
268                    w.compact().unwrap();
269                }
270                let boot_at = Timestamp(t + 5);
271                let d = due(r.sync_ledger().unwrap(), &me);
272                let want: Vec<Due> = r.ledger().live.values()
273                    .filter(|l| l.txn.lease.owner == me && !l.txn.state.is_pre_mutation())
274                    .map(|l| (l.txn.id, l.txn.lease.epoch, l.txn.lease.worker_pid))
275                    .collect();
276                prop_assert_eq!(&d, &want);
277                // The watchdog wins some races after `due` was read.
278                let mut raced = vec![];
279                for ix in races {
280                    if d.is_empty() { break; }
281                    let (id, epoch, _) = d[ix.index(d.len())];
282                    if raced.contains(&id) { continue; }
283                    let lease = Lease { txn: id, owner: owners[2].clone(), epoch: epoch + 1, worker_pid: None, renewed: Timestamp(t), jobs: vec![] };
284                    w.append(Timestamp(t), Record::Takeover { txn: id, from_epoch: epoch, lease }).unwrap();
285                    raced.push(id);
286                }
287                let renewed = renew_due(&mut r, &d, Timestamp(t + 5), boot_at);
288                prop_assert_eq!(renewed, d.len() - raced.len());
289                let l = w.sync_ledger().unwrap().clone();
290                for live in l.live.values() {
291                    let lease = &live.txn.lease;
292                    let should = d.iter().any(|(id, _, _)| *id == live.txn.id) && !raced.contains(&live.txn.id);
293                    prop_assert_eq!(
294                        lease.renewed == boot_at, should,
295                        "txn {:?} ({:?}, {:?}, epoch {}): renewed {:?}", live.txn.id, lease.owner, live.txn.state, lease.epoch, lease.renewed
296                    );
297                    if should {
298                        let (_, epoch, worker) = d.iter().find(|(id, _, _)| *id == live.txn.id).unwrap();
299                        prop_assert_eq!((lease.epoch, lease.worker_pid, &lease.owner), (*epoch, *worker, &me));
300                    }
301                }
302            }
303        }
304    }
305}
planarian_daemon/renew.rs

planarian_daemon/
renew.rs
