Skip to main content

ignition_core/deploy/apply/
hold.rs

1//! The hold layer of the apply engine (Phase 23, D-02): the scan-lock held
2//! for the length of one commit and its scan, judged by two clocks.
3//!
4//! # D-02's rule
5//!
6//! The hold is the scan-lock's `holdTimeout`. The user fixed its shape in
7//! D-02, from the 22-14 measurement (22-LIVE-CAPTURES `## Decisions: OQ5
8//! timings, hold budget and timeout defaults (22-14)`, OQ5-2):
9//!
10//! ```text
11//! hold = ceil(2 x (p95 commit + p95 scan) + 1), floor 10 s, ceiling 300 s
12//! ```
13//!
14//! [`HOLD_FLOOR_S`] and [`HOLD_CEILING_S`] are those two numbers. They are
15//! user decisions and not measurements, and the ceiling changes only by the
16//! user's D-03a choice (23-10 Task 3 option-a). [`hold_for`] returns the
17//! formula UNCLAMPED so that D-03a can see the raw value; [`hold_clamped`]
18//! is the separate step that applies floor and ceiling, so nothing clamps
19//! silently.
20//!
21//! # The measured value
22//!
23//! 505 s and 699 s, which the formula gives for the 22-14 timings, came from
24//! the old per-file sftp commit that D-01 replaces. They appear in this
25//! file's tests as formula regression values and are never a hold. The hold
26//! the engine ships with is [`HOLD_S`], computed from the batched commit
27//! measured on the Windows capture host (23-10, `23-LIVE-CAPTURES.md`
28//! `## OQ5b` and `## Decisions: OQ5b`): the p95 literals below, the formula
29//! on the largest measured scale, and D-02's clamp. [`HoldModel::MEASURED`]
30//! is the model the apply verb (23-13) names. The literals are written here
31//! and nowhere else; `hold_s_is_derived_from_recorded_measurements`
32//! recomputes the hold and the fit from the recorded numbers and fails if
33//! either drifts.
34//!
35//! # Two clocks
36//!
37//! [`HeldLock`] never compares the gateway's `timestamp` or
38//! `lastScanTimestamp` with the runner's clock (Pitfall 15). It judges the
39//! lock by the gateway's `remaining` and by its own monotonic [`Instant`],
40//! taken before the acquire request; either one saying the lock is
41//! expiring makes it expired.
42//!
43//! # Whose lock
44//!
45//! The gateway frees a lapsed lock and grants it to the next actor, so a
46//! held lock after our hold lapsed is not proof that it is still ours
47//! (23-17, DAPL-07). [`HeldLock::acquire`] reads the lock once right after
48//! the grant and remembers its `timestamp` and `actor`; C1c and D9 show the
49//! `timestamp` is unchanged across reads of one hold and different for each
50//! new hold. A later reading with another `timestamp` or `actor` is another
51//! actor's lock and reads as [`LockVerdict::NotHeld`]. The two readings come
52//! from the same gateway and are only compared with each other, never with a
53//! clock.
54
55use std::sync::Mutex;
56use std::time::{Duration, Instant};
57
58use crate::client::DeployApi;
59use crate::client::scan::{LockObservation, LockVerdict, ScanKind, ScanLockInfo, lock_verdict};
60use crate::error::CoreError;
61
62/// Seconds the engine asks the gateway to wait for a lock held by someone
63/// else (`acquireTimeout`). The gateway refuses 0 (and any non-positive
64/// timeout) with a 400 "Timeouts must be positive" (23-16 live gate), and it
65/// never waits out a positive one: a held lock is an immediate 409 (D1,
66/// `20-LIVE-CAPTURES.md`), so the smallest positive value is "no wait".
67pub const ACQUIRE_WAIT_S: u64 = 1;
68
69/// D-02's floor: the shortest hold the engine ever requests, in seconds.
70/// A user decision, not a measurement.
71pub const HOLD_FLOOR_S: u64 = 10;
72
73/// D-02's ceiling: the longest hold the engine ever requests, in seconds.
74/// A user decision, not a measurement. It changes only by the user's D-03a
75/// choice (23-10 Task 3 option-a), which updates the clamp assertions in
76/// the same commit.
77pub const HOLD_CEILING_S: u64 = 300;
78
79/// D-02's formula, unclamped: `ceil(2 x (p95_commit_s + p95_scan_s) + 1)`.
80pub fn hold_for(p95_commit_s: f64, p95_scan_s: f64) -> u64 {
81    // `as` saturates: a negative or NaN input reads as 0.
82    (2.0 * (p95_commit_s + p95_scan_s) + 1.0).ceil() as u64
83}
84
85/// `raw` held to D-02's floor and ceiling.
86pub fn hold_clamped(raw: u64) -> u64 {
87    raw.clamp(HOLD_FLOOR_S, HOLD_CEILING_S)
88}
89
90/// Whether a run whose commit is estimated at `est_commit_s` and whose scan
91/// takes `p95_scan_s` fits in `hold_s`: the hold the formula asks for is at
92/// most `hold_s`.
93pub fn fits(hold_s: u64, est_commit_s: f64, p95_scan_s: f64) -> bool {
94    hold_for(est_commit_s, p95_scan_s) <= hold_s
95}
96
97/// Files in the scale 1 corpus the commit was timed with (22-14, 23-10).
98pub const FILES_SCALE1: usize = 152;
99
100/// Files in the scale 10 corpus, the largest tree measured.
101pub const FILES_SCALE10: usize = 1520;
102
103/// p95 seconds of the slowest commit variant at 152 files: the replace
104/// commit, every file replaced and every pre-image copied (`## Decisions:
105/// OQ5b`, OQ5b-1 and OQ5b-2; n = 5, so the nearest-rank p95 is the maximum).
106pub const P95_COMMIT_S_SCALE1: f64 = 10.123;
107
108/// p95 seconds of the slowest commit variant at 1,520 files, the largest
109/// measured scale (FA-23-07): the replace commit (`## Decisions: OQ5b`,
110/// OQ5b-1 and OQ5b-2; FA-23-08). The new-file commit measured 62.040 s.
111pub const P95_COMMIT_S_SCALE10: f64 = 85.496;
112
113/// p95 seconds of the slowest config scan at 1,520 files, wall clock through
114/// `ign api call` (`## Decisions: OQ5b`, OQ5b-1): the scan after the replace
115/// commit. The gateway's own figure is under 1.2 s.
116pub const P95_SCAN_S: f64 = 1.579;
117
118/// The hold the apply engine requests, in seconds: D-02's formula on the
119/// largest measured scale, clamped. `ceil(2 x (85.496 + 1.579) + 1)` is 176,
120/// inside the ceiling, so D-03a did not trigger. A literal so that it reads in
121/// `grep`; `hold_s_is_derived_from_recorded_measurements` recomputes it from
122/// the three literals above and fails if any of the four drifts. One constant:
123/// no per-run scaling and no per-target override (D-02).
124pub const HOLD_S: u64 = 176;
125
126/// The two-scale fit of the p95 commit time against the file count: seconds
127/// per file, `(p95 at 1,520 - p95 at 152) / (1520 - 152)`.
128const FIT_PER_FILE_S: f64 =
129    (P95_COMMIT_S_SCALE10 - P95_COMMIT_S_SCALE1) / (FILES_SCALE10 as f64 - FILES_SCALE1 as f64);
130
131/// The fixed part of the fit: `p95 at 152 - 152 x per file`, floored at 0.
132const FIT_FIXED_S: f64 = {
133    let fixed = P95_COMMIT_S_SCALE1 - FILES_SCALE1 as f64 * FIT_PER_FILE_S;
134    if fixed > 0.0 { fixed } else { 0.0 }
135};
136
137/// Seconds of margin a lock needs left at a check: one REST round trip from
138/// the check to the scan request, which releases the lock (D12).
139pub const LOCK_MARGIN_S: i64 = 2;
140
141/// What the engine knows about the length of a run.
142#[derive(Debug, Clone, Copy, PartialEq)]
143pub struct HoldModel {
144    /// Seconds the scan-lock is requested for.
145    pub hold_s: u64,
146    /// Fixed seconds a commit costs, whatever its size.
147    pub commit_fixed_s: f64,
148    /// Seconds a commit costs per touched file.
149    pub commit_per_file_s: f64,
150    /// 95th percentile seconds a config scan takes.
151    pub p95_scan_s: f64,
152}
153
154impl HoldModel {
155    /// The model measured on the Windows capture host (23-10): the shipped
156    /// hold, the p95 scan and the two-scale fit of the commit time. The D-03
157    /// refusal (23-07) and the apply verb (23-13) use this and no other.
158    pub const MEASURED: HoldModel = HoldModel {
159        hold_s: HOLD_S,
160        commit_fixed_s: FIT_FIXED_S,
161        commit_per_file_s: FIT_PER_FILE_S,
162        p95_scan_s: P95_SCAN_S,
163    };
164
165    /// Seconds a commit of `files` files is estimated to take.
166    pub fn estimate_commit_s(&self, files: usize) -> f64 {
167        self.commit_fixed_s + files as f64 * self.commit_per_file_s
168    }
169
170    /// Whether a run touching `files` files fits in this model's hold (D-03).
171    pub fn fits(&self, files: usize) -> bool {
172        fits(self.hold_s, self.estimate_commit_s(files), self.p95_scan_s)
173    }
174}
175
176/// A monotonic clock, so a test can move time.
177pub trait Clock: Send + Sync {
178    /// The current instant.
179    fn now(&self) -> Instant;
180}
181
182/// The real monotonic clock.
183#[derive(Debug, Clone, Copy, Default)]
184pub struct SystemClock;
185
186impl Clock for SystemClock {
187    fn now(&self) -> Instant {
188        Instant::now()
189    }
190}
191
192/// What tells one hold from another on the gateway (C1c, D9).
193#[derive(Debug, Clone, PartialEq, Eq)]
194struct LockIdentity {
195    timestamp: Option<i64>,
196    actor: Option<String>,
197}
198
199impl LockIdentity {
200    fn of(info: &ScanLockInfo) -> Self {
201        LockIdentity {
202            timestamp: info.timestamp,
203            actor: info.actor.clone(),
204        }
205    }
206}
207
208/// A scan-lock this run took.
209#[derive(Debug, Clone)]
210pub struct HeldLock {
211    kind: ScanKind,
212    hold_s: u64,
213    acquired: Instant,
214    /// The hold as the gateway showed it right after the grant; `None` when
215    /// the lock read free then, in which case no later lock is ours.
216    identity: Option<LockIdentity>,
217}
218
219impl HeldLock {
220    /// Take the lock for `hold_s` seconds. The local instant is taken
221    /// before the request, so the local clock never reads later than the
222    /// gateway's own count of the hold (research Pattern 6). One read right
223    /// after the grant remembers which hold is ours (see the module doc).
224    pub async fn acquire(
225        deploy: &dyn DeployApi,
226        kind: ScanKind,
227        hold_s: u64,
228        clock: &dyn Clock,
229    ) -> Result<HeldLock, CoreError> {
230        let acquired = clock.now();
231        deploy
232            .scan_lock_acquire(kind, ACQUIRE_WAIT_S, hold_s)
233            .await?;
234        let identity = match deploy.scan_lock_info(kind).await? {
235            LockObservation::Held(info) => Some(LockIdentity::of(&info)),
236            LockObservation::Free => None,
237        };
238        Ok(HeldLock {
239            kind,
240            hold_s,
241            acquired,
242            identity,
243        })
244    }
245
246    /// The scan kind this lock covers.
247    pub fn kind(&self) -> ScanKind {
248        self.kind
249    }
250
251    /// The hold that was requested, in seconds.
252    pub fn hold_s(&self) -> u64 {
253        self.hold_s
254    }
255
256    /// Seconds left by the local monotonic clock: the hold minus the whole
257    /// seconds since the request, rounded up so the answer errs short.
258    pub fn local_remaining_s(&self, clock: &dyn Clock) -> i64 {
259        let elapsed = clock.now().saturating_duration_since(self.acquired);
260        let elapsed_s = elapsed.as_secs() + u64::from(elapsed.subsec_nanos() > 0);
261        i64::try_from(self.hold_s)
262            .unwrap_or(i64::MAX)
263            .saturating_sub(i64::try_from(elapsed_s).unwrap_or(i64::MAX))
264    }
265
266    /// Whether `info` is the hold this value took.
267    fn is_ours(&self, info: &ScanLockInfo) -> bool {
268        self.identity.as_ref() == Some(&LockIdentity::of(info))
269    }
270
271    /// The gateway's side of the judgement alone: [`lock_verdict`] on its
272    /// `remaining`, where a lock that is not ours reads as
273    /// [`LockVerdict::NotHeld`]. This is the only call that asks the gateway.
274    pub async fn check_gateway(
275        &self,
276        deploy: &dyn DeployApi,
277        margin_s: i64,
278    ) -> Result<LockVerdict, CoreError> {
279        let observed = deploy.scan_lock_info(self.kind).await?;
280        Ok(match &observed {
281            LockObservation::Held(info) if !self.is_ours(info) => LockVerdict::NotHeld,
282            _ => lock_verdict(&observed, margin_s),
283        })
284    }
285
286    /// Judge the lock now, by both clocks: the gateway's `remaining`
287    /// through [`HeldLock::check_gateway`], and the local count of the hold.
288    /// Either one saying the lock has less than `margin_s` left makes the
289    /// answer [`LockVerdict::Expiring`]; a free lock, or one that is another
290    /// actor's, is [`LockVerdict::NotHeld`]. `timestamp` and
291    /// `lastScanTimestamp` are never compared with the runner's clock
292    /// (Pitfall 15).
293    pub async fn check(
294        &self,
295        deploy: &dyn DeployApi,
296        margin_s: i64,
297        clock: &dyn Clock,
298    ) -> Result<LockVerdict, CoreError> {
299        let local_s = self.local_remaining_s(clock);
300        Ok(match self.check_gateway(deploy, margin_s).await? {
301            LockVerdict::Held { remaining_s } if local_s < margin_s => LockVerdict::Expiring {
302                remaining_s: remaining_s.min(local_s),
303            },
304            verdict => verdict,
305        })
306    }
307
308    /// Wait until the gateway no longer shows this hold: it lapsed, a scan
309    /// released it, or another actor holds the lock now. A scan is never
310    /// the way to end it, because a scan loads whatever is on disk. Polls
311    /// every `interval` and gives up after the hold plus a margin of real
312    /// time.
313    pub async fn wait_for_lapse(
314        &self,
315        deploy: &dyn DeployApi,
316        interval: Duration,
317    ) -> Result<(), CoreError> {
318        let limit = Duration::from_secs(self.hold_s.saturating_add(LOCK_MARGIN_S as u64 + 3));
319        let started = tokio::time::Instant::now();
320        loop {
321            if self.check_gateway(deploy, 0).await? == LockVerdict::NotHeld {
322                return Ok(());
323            }
324            if started.elapsed() >= limit {
325                return Err(CoreError::Internal(
326                    "the scan-lock did not lapse within its hold".to_string(),
327                ));
328            }
329            tokio::time::sleep(interval).await;
330        }
331    }
332}
333
334/// A clock a test moves by hand.
335#[doc(hidden)]
336#[derive(Debug)]
337pub struct ManualClock {
338    base: Instant,
339    offset: Mutex<Duration>,
340}
341
342impl Default for ManualClock {
343    fn default() -> Self {
344        Self::new()
345    }
346}
347
348impl ManualClock {
349    /// A clock at an arbitrary starting instant.
350    pub fn new() -> Self {
351        ManualClock {
352            base: Instant::now(),
353            offset: Mutex::new(Duration::ZERO),
354        }
355    }
356
357    /// Move the clock forward.
358    pub fn advance(&self, by: Duration) {
359        *self.offset.lock().expect("manual clock") += by;
360    }
361}
362
363impl Clock for ManualClock {
364    fn now(&self) -> Instant {
365        self.base + *self.offset.lock().expect("manual clock")
366    }
367}
368
369#[cfg(test)]
370mod tests {
371    use std::collections::BTreeMap;
372
373    use super::*;
374    use crate::client::config_resources::{
375        MutationOutcome, ResourceCreate, ResourceName, ResourceRecord, ResourceUpdate,
376    };
377    use crate::client::mode::{ModeCreate, ModeRecord};
378    use crate::client::query::{ListEnvelope, ListQuery};
379    use crate::client::scan::{LockObservation, ScanLockInfo, ScanStatus};
380
381    // ---- the formula --------------------------------------------------------
382
383    #[test]
384    fn hold_formula_regression() {
385        // The 22-14 per-file-path numbers, used only to pin the formula.
386        assert_eq!(hold_for(251.063, 0.888), 505);
387        assert_eq!(hold_for(348.225, 0.642), 699);
388        // `hold_for` does not clamp; `hold_clamped` is a separate step.
389        assert!(hold_for(251.063, 0.888) > HOLD_CEILING_S);
390        assert_eq!(hold_clamped(505), HOLD_CEILING_S);
391        assert_eq!(hold_clamped(699), HOLD_CEILING_S);
392        assert_eq!(hold_clamped(HOLD_CEILING_S), HOLD_CEILING_S);
393        assert_eq!(hold_clamped(120), 120);
394    }
395
396    #[test]
397    fn hold_floor() {
398        assert_eq!(hold_for(1.0, 0.5), 4);
399        assert_eq!(hold_clamped(4), HOLD_FLOOR_S);
400        assert_eq!(hold_clamped(0), HOLD_FLOOR_S);
401        assert_eq!(hold_clamped(HOLD_FLOOR_S), HOLD_FLOOR_S);
402    }
403
404    #[test]
405    fn d02_numbers_are_the_users() {
406        assert_eq!((HOLD_FLOOR_S, HOLD_CEILING_S), (10, 300));
407    }
408
409    // ---- the measured hold (23-10, `## Decisions: OQ5b`) ---------------------
410
411    /// The numbers recorded in `23-LIVE-CAPTURES.md` `## Decisions: OQ5b`,
412    /// copied here so a change to a constant in the source fails this test
413    /// until the record and the test are changed with it.
414    const RECORDED_COMMIT_S_152: f64 = 10.123;
415    const RECORDED_COMMIT_S_1520: f64 = 85.496;
416    const RECORDED_SCAN_S: f64 = 1.579;
417    const RECORDED_HOLD_S: u64 = 176;
418
419    #[test]
420    fn hold_s_is_derived_from_recorded_measurements() {
421        assert_eq!(P95_COMMIT_S_SCALE1, RECORDED_COMMIT_S_152);
422        assert_eq!(P95_COMMIT_S_SCALE10, RECORDED_COMMIT_S_1520);
423        assert_eq!(P95_SCAN_S, RECORDED_SCAN_S);
424        assert_eq!((FILES_SCALE1, FILES_SCALE10), (152, 1520));
425
426        // D-02's formula on the largest measured scale, then its clamp.
427        let unclamped = hold_for(RECORDED_COMMIT_S_1520, RECORDED_SCAN_S);
428        assert_eq!(
429            unclamped, 176,
430            "ceil(2 x (85.496 + 1.579) + 1) = ceil(175.15)"
431        );
432        assert!(
433            unclamped <= HOLD_CEILING_S,
434            "D-03a is not triggered: the unclamped hold is inside the ceiling"
435        );
436        assert_eq!(HOLD_S, hold_clamped(unclamped));
437        assert_eq!(HOLD_S, RECORDED_HOLD_S);
438
439        // The fit through the two scales, recomputed from the literals.
440        let per_file = (RECORDED_COMMIT_S_1520 - RECORDED_COMMIT_S_152) / (1520.0 - 152.0);
441        let fixed = (RECORDED_COMMIT_S_152 - 152.0 * per_file).max(0.0);
442        assert!(fixed > 0.0, "the floor at 0 did not apply");
443        assert_eq!(HoldModel::MEASURED.commit_per_file_s, per_file);
444        assert_eq!(HoldModel::MEASURED.commit_fixed_s, fixed);
445        assert_eq!(HoldModel::MEASURED.p95_scan_s, RECORDED_SCAN_S);
446        assert_eq!(HoldModel::MEASURED.hold_s, HOLD_S);
447        // The fit goes through both measured points.
448        assert!((HoldModel::MEASURED.estimate_commit_s(152) - RECORDED_COMMIT_S_152).abs() < 1e-9);
449        assert!(
450            (HoldModel::MEASURED.estimate_commit_s(1520) - RECORDED_COMMIT_S_1520).abs() < 1e-9
451        );
452    }
453
454    #[test]
455    fn hold_model_fits_the_measured_largest_tree() {
456        let model = HoldModel::MEASURED;
457        assert!(model.fits(1520), "the largest measured tree fits");
458        assert!(model.fits(0));
459        // The smallest change that does not fit; recorded in OQ5b-3.
460        let first_over = (1520..4000)
461            .find(|files| !model.fits(*files))
462            .expect("some size does not fit");
463        assert_eq!(
464            first_over,
465            1528,
466            "1,527 files is the largest change that fits {} s; 1,528 needs {} s",
467            model.hold_s,
468            hold_for(model.estimate_commit_s(1528), model.p95_scan_s)
469        );
470        assert!(model.fits(first_over - 1));
471        assert!(!model.fits(first_over));
472        // The refusal reads the same numbers the hold was built from.
473        assert_eq!(
474            hold_for(model.estimate_commit_s(1520), model.p95_scan_s),
475            HOLD_S
476        );
477    }
478
479    #[test]
480    fn hold_fit_boundary() {
481        let needed = hold_for(6.0, 0.5);
482        assert_eq!(needed, 14);
483        assert!(fits(needed, 6.0, 0.5), "equal fits");
484        assert!(!fits(needed - 1, 6.0, 0.5), "one second over does not");
485
486        let model = HoldModel {
487            hold_s: 14,
488            commit_fixed_s: 1.0,
489            commit_per_file_s: 0.5,
490            p95_scan_s: 0.5,
491        };
492        assert_eq!(model.estimate_commit_s(10), 6.0);
493        assert!(model.fits(10), "needs exactly 14 s");
494        assert!(!model.fits(11), "needs 16 s");
495        let tight = HoldModel {
496            hold_s: 13,
497            ..model
498        };
499        assert!(!tight.fits(10), "one second short");
500    }
501
502    // ---- the held lock ------------------------------------------------------
503
504    /// A gateway that only knows the scan-lock; the rest is not used.
505    struct LockGateway {
506        info: Mutex<LockObservation>,
507        acquired: Mutex<Vec<(ScanKind, u64, u64)>>,
508    }
509
510    impl LockGateway {
511        fn holding(remaining: i64, timestamp: Option<i64>) -> Self {
512            LockGateway {
513                info: Mutex::new(LockObservation::Held(ScanLockInfo {
514                    timestamp,
515                    remaining,
516                    actor: Some("api-token:t:127.0.0.1".to_string()),
517                    extra: BTreeMap::new(),
518                })),
519                acquired: Mutex::new(Vec::new()),
520            }
521        }
522
523        fn free() -> Self {
524            LockGateway {
525                info: Mutex::new(LockObservation::Free),
526                acquired: Mutex::new(Vec::new()),
527            }
528        }
529    }
530
531    fn unused<T>() -> Result<T, CoreError> {
532        Err(CoreError::Internal(
533            "not used by the hold tests".to_string(),
534        ))
535    }
536
537    #[async_trait::async_trait]
538    impl DeployApi for LockGateway {
539        async fn resource_names(
540            &self,
541            _module: &str,
542            _ty: &str,
543        ) -> Result<ListEnvelope<ResourceName>, CoreError> {
544            unused()
545        }
546        async fn resource_list(
547            &self,
548            _module: &str,
549            _ty: &str,
550            _q: &ListQuery,
551        ) -> Result<ListEnvelope<ResourceRecord>, CoreError> {
552            unused()
553        }
554        async fn resource_find(
555            &self,
556            _module: &str,
557            _ty: &str,
558            _name: &str,
559            _collection: Option<&str>,
560        ) -> Result<ResourceRecord, CoreError> {
561            unused()
562        }
563        async fn resource_singleton(
564            &self,
565            _module: &str,
566            _ty: &str,
567            _collection: Option<&str>,
568        ) -> Result<ResourceRecord, CoreError> {
569            unused()
570        }
571        async fn resource_create(
572            &self,
573            _module: &str,
574            _ty: &str,
575            _items: &[ResourceCreate],
576        ) -> Result<MutationOutcome, CoreError> {
577            unused()
578        }
579        async fn resource_update(
580            &self,
581            _module: &str,
582            _ty: &str,
583            _items: &[ResourceUpdate],
584        ) -> Result<MutationOutcome, CoreError> {
585            unused()
586        }
587        async fn scan_lock_acquire(
588            &self,
589            kind: ScanKind,
590            acquire_s: u64,
591            hold_s: u64,
592        ) -> Result<(), CoreError> {
593            self.acquired
594                .lock()
595                .expect("acquired")
596                .push((kind, acquire_s, hold_s));
597            Ok(())
598        }
599        async fn scan_lock_info(&self, _kind: ScanKind) -> Result<LockObservation, CoreError> {
600            Ok(self.info.lock().expect("info").clone())
601        }
602        async fn scan_status(&self, _kind: ScanKind) -> Result<ScanStatus, CoreError> {
603            unused()
604        }
605        async fn scan_request(&self, _kind: ScanKind) -> Result<(), CoreError> {
606            unused()
607        }
608        async fn modes(&self) -> Result<ListEnvelope<ModeRecord>, CoreError> {
609            unused()
610        }
611        async fn mode_create(&self, _body: &ModeCreate) -> Result<ModeRecord, CoreError> {
612            unused()
613        }
614        async fn deployment_mode(&self) -> Result<Option<String>, CoreError> {
615            unused()
616        }
617        async fn restart_pending(&self) -> Result<Vec<String>, CoreError> {
618            unused()
619        }
620        async fn redundancy_resync(&self) -> Result<(), CoreError> {
621            unused()
622        }
623        async fn secret_names(
624            &self,
625            _provider: &str,
626        ) -> Result<crate::client::deploy_api::SecretNameListing, CoreError> {
627            unused()
628        }
629    }
630
631    const HOLD: u64 = 60;
632
633    async fn lock_on(gateway: &LockGateway, clock: &ManualClock) -> HeldLock {
634        HeldLock::acquire(gateway, ScanKind::Config, HOLD, clock)
635            .await
636            .expect("acquire")
637    }
638
639    #[tokio::test]
640    async fn held_lock_acquires_the_config_lock_with_the_hold_and_the_smallest_wait() {
641        let gateway = LockGateway::holding(HOLD as i64, None);
642        let clock = ManualClock::new();
643        let lock = lock_on(&gateway, &clock).await;
644        assert_eq!(
645            *gateway.acquired.lock().expect("acquired"),
646            [(ScanKind::Config, ACQUIRE_WAIT_S, HOLD)]
647        );
648        assert_eq!((lock.kind(), lock.hold_s()), (ScanKind::Config, HOLD));
649    }
650
651    #[tokio::test]
652    async fn held_lock_is_held_while_both_clocks_have_the_margin() {
653        let gateway = LockGateway::holding(30, None);
654        let clock = ManualClock::new();
655        let lock = lock_on(&gateway, &clock).await;
656        clock.advance(Duration::from_secs(5));
657        let verdict = lock
658            .check(&gateway, LOCK_MARGIN_S, &clock)
659            .await
660            .expect("check");
661        assert_eq!(verdict, LockVerdict::Held { remaining_s: 30 });
662    }
663
664    #[tokio::test]
665    async fn held_lock_local_clock_expiry() {
666        // The gateway still says 30 s are left; the local clock says the hold
667        // is nearly gone, and the local clock wins.
668        let gateway = LockGateway::holding(30, None);
669        let clock = ManualClock::new();
670        let lock = lock_on(&gateway, &clock).await;
671
672        // Exactly the margin left locally is still held (the gateway's rule).
673        clock.advance(Duration::from_secs(HOLD - LOCK_MARGIN_S as u64));
674        assert_eq!(lock.local_remaining_s(&clock), LOCK_MARGIN_S);
675        assert!(matches!(
676            lock.check(&gateway, LOCK_MARGIN_S, &clock)
677                .await
678                .expect("check"),
679            LockVerdict::Held { .. }
680        ));
681
682        // One more second and it is expiring.
683        clock.advance(Duration::from_secs(1));
684        assert_eq!(lock.local_remaining_s(&clock), LOCK_MARGIN_S - 1);
685        let verdict = lock
686            .check(&gateway, LOCK_MARGIN_S, &clock)
687            .await
688            .expect("check");
689        assert_eq!(
690            verdict,
691            LockVerdict::Expiring {
692                remaining_s: LOCK_MARGIN_S - 1
693            }
694        );
695
696        // Past the hold the local count is negative, never wrapped.
697        clock.advance(Duration::from_secs(HOLD));
698        assert!(lock.local_remaining_s(&clock) < 0);
699    }
700
701    #[tokio::test]
702    async fn held_lock_gateway_expiring() {
703        // A fresh local clock, but the gateway reports one second left.
704        let gateway = LockGateway::holding(1, None);
705        let clock = ManualClock::new();
706        let lock = lock_on(&gateway, &clock).await;
707        let verdict = lock
708            .check(&gateway, LOCK_MARGIN_S, &clock)
709            .await
710            .expect("check");
711        assert_eq!(verdict, LockVerdict::Expiring { remaining_s: 1 });
712    }
713
714    #[tokio::test]
715    async fn held_lock_free_on_the_gateway_is_not_held() {
716        let gateway = LockGateway::free();
717        let clock = ManualClock::new();
718        let lock = lock_on(&gateway, &clock).await;
719        let verdict = lock
720            .check(&gateway, LOCK_MARGIN_S, &clock)
721            .await
722            .expect("check");
723        assert_eq!(verdict, LockVerdict::NotHeld);
724    }
725
726    #[tokio::test]
727    async fn held_lock_never_reads_the_gateway_timestamp() {
728        let clock = ManualClock::new();
729        let mut verdicts = Vec::new();
730        for timestamp in [None, Some(0), Some(i64::MAX), Some(i64::MIN)] {
731            let gateway = LockGateway::holding(30, timestamp);
732            let lock = lock_on(&gateway, &clock).await;
733            verdicts.push(
734                lock.check(&gateway, LOCK_MARGIN_S, &clock)
735                    .await
736                    .expect("check"),
737            );
738        }
739        assert!(
740            verdicts
741                .iter()
742                .all(|v| *v == LockVerdict::Held { remaining_s: 30 })
743        );
744    }
745
746    fn held_by(timestamp: i64, actor: &str, remaining: i64) -> LockObservation {
747        LockObservation::Held(ScanLockInfo {
748            timestamp: Some(timestamp),
749            remaining,
750            actor: Some(actor.to_string()),
751            extra: BTreeMap::new(),
752        })
753    }
754
755    #[tokio::test]
756    async fn held_lock_another_actors_hold_is_not_ours() {
757        let gateway = LockGateway::holding(30, Some(100));
758        let clock = ManualClock::new();
759        let lock = lock_on(&gateway, &clock).await;
760        // The same actor name, a new hold: the timestamp tells them apart.
761        *gateway.info.lock().expect("info") = held_by(200, "api-token:t:127.0.0.1", 59);
762        assert_eq!(
763            lock.check(&gateway, LOCK_MARGIN_S, &clock)
764                .await
765                .expect("check"),
766            LockVerdict::NotHeld
767        );
768        // Another actor with the very same timestamp is not ours either.
769        *gateway.info.lock().expect("info") = held_by(100, "api-token:other:127.0.0.1", 59);
770        assert_eq!(
771            lock.check_gateway(&gateway, 0).await.expect("check"),
772            LockVerdict::NotHeld
773        );
774        // And our own hold reads as held again.
775        *gateway.info.lock().expect("info") = held_by(100, "api-token:t:127.0.0.1", 29);
776        assert_eq!(
777            lock.check_gateway(&gateway, 0).await.expect("check"),
778            LockVerdict::Held { remaining_s: 29 }
779        );
780    }
781
782    #[tokio::test]
783    async fn held_lock_unseen_after_the_grant_owns_no_later_lock() {
784        // The lock read free right after the grant, so a lock seen later was
785        // taken by someone else.
786        let gateway = LockGateway::free();
787        let clock = ManualClock::new();
788        let lock = lock_on(&gateway, &clock).await;
789        *gateway.info.lock().expect("info") = held_by(5, "api-token:t:127.0.0.1", 59);
790        assert_eq!(
791            lock.check(&gateway, LOCK_MARGIN_S, &clock)
792                .await
793                .expect("check"),
794            LockVerdict::NotHeld
795        );
796    }
797
798    #[tokio::test]
799    async fn held_lock_check_gateway_ignores_the_local_clock() {
800        let gateway = LockGateway::holding(30, Some(100));
801        let clock = ManualClock::new();
802        let lock = lock_on(&gateway, &clock).await;
803        clock.advance(Duration::from_secs(HOLD * 2));
804        assert!(lock.local_remaining_s(&clock) < 0);
805        assert_eq!(
806            lock.check_gateway(&gateway, LOCK_MARGIN_S)
807                .await
808                .expect("check"),
809            LockVerdict::Held { remaining_s: 30 }
810        );
811    }
812
813    #[tokio::test]
814    async fn held_lock_wait_for_lapse_returns_when_the_hold_is_gone() {
815        let gateway = LockGateway::holding(30, Some(100));
816        let clock = ManualClock::new();
817        let lock = lock_on(&gateway, &clock).await;
818        // Another actor's lock ends the wait at once: it is not ours to wait for.
819        *gateway.info.lock().expect("info") = held_by(7, "api-token:other:127.0.0.1", 59);
820        lock.wait_for_lapse(&gateway, Duration::from_millis(1))
821            .await
822            .expect("not ours");
823        *gateway.info.lock().expect("info") = LockObservation::Free;
824        lock.wait_for_lapse(&gateway, Duration::from_millis(1))
825            .await
826            .expect("free");
827    }
828
829    #[test]
830    fn manual_clock_moves_only_when_told() {
831        let clock = ManualClock::new();
832        let start = clock.now();
833        assert_eq!(clock.now(), start);
834        clock.advance(Duration::from_secs(3));
835        assert_eq!(clock.now() - start, Duration::from_secs(3));
836    }
837}