1use 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
62pub const ACQUIRE_WAIT_S: u64 = 1;
68
69pub const HOLD_FLOOR_S: u64 = 10;
72
73pub const HOLD_CEILING_S: u64 = 300;
78
79pub fn hold_for(p95_commit_s: f64, p95_scan_s: f64) -> u64 {
81 (2.0 * (p95_commit_s + p95_scan_s) + 1.0).ceil() as u64
83}
84
85pub fn hold_clamped(raw: u64) -> u64 {
87 raw.clamp(HOLD_FLOOR_S, HOLD_CEILING_S)
88}
89
90pub 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
97pub const FILES_SCALE1: usize = 152;
99
100pub const FILES_SCALE10: usize = 1520;
102
103pub const P95_COMMIT_S_SCALE1: f64 = 10.123;
107
108pub const P95_COMMIT_S_SCALE10: f64 = 85.496;
112
113pub const P95_SCAN_S: f64 = 1.579;
117
118pub const HOLD_S: u64 = 176;
125
126const FIT_PER_FILE_S: f64 =
129 (P95_COMMIT_S_SCALE10 - P95_COMMIT_S_SCALE1) / (FILES_SCALE10 as f64 - FILES_SCALE1 as f64);
130
131const 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
137pub const LOCK_MARGIN_S: i64 = 2;
140
141#[derive(Debug, Clone, Copy, PartialEq)]
143pub struct HoldModel {
144 pub hold_s: u64,
146 pub commit_fixed_s: f64,
148 pub commit_per_file_s: f64,
150 pub p95_scan_s: f64,
152}
153
154impl HoldModel {
155 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 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 pub fn fits(&self, files: usize) -> bool {
172 fits(self.hold_s, self.estimate_commit_s(files), self.p95_scan_s)
173 }
174}
175
176pub trait Clock: Send + Sync {
178 fn now(&self) -> Instant;
180}
181
182#[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#[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#[derive(Debug, Clone)]
210pub struct HeldLock {
211 kind: ScanKind,
212 hold_s: u64,
213 acquired: Instant,
214 identity: Option<LockIdentity>,
217}
218
219impl HeldLock {
220 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 pub fn kind(&self) -> ScanKind {
248 self.kind
249 }
250
251 pub fn hold_s(&self) -> u64 {
253 self.hold_s
254 }
255
256 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 fn is_ours(&self, info: &ScanLockInfo) -> bool {
268 self.identity.as_ref() == Some(&LockIdentity::of(info))
269 }
270
271 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 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 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#[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 pub fn new() -> Self {
351 ManualClock {
352 base: Instant::now(),
353 offset: Mutex::new(Duration::ZERO),
354 }
355 }
356
357 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 #[test]
384 fn hold_formula_regression() {
385 assert_eq!(hold_for(251.063, 0.888), 505);
387 assert_eq!(hold_for(348.225, 0.642), 699);
388 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 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 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 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 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 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 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 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 let gateway = LockGateway::holding(30, None);
669 let clock = ManualClock::new();
670 let lock = lock_on(&gateway, &clock).await;
671
672 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 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 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 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 *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 *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 *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 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 *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}