use std::sync::Mutex;
use std::time::{Duration, Instant};
use crate::client::DeployApi;
use crate::client::scan::{LockObservation, LockVerdict, ScanKind, ScanLockInfo, lock_verdict};
use crate::error::CoreError;
pub const ACQUIRE_WAIT_S: u64 = 1;
pub const HOLD_FLOOR_S: u64 = 10;
pub const HOLD_CEILING_S: u64 = 300;
pub fn hold_for(p95_commit_s: f64, p95_scan_s: f64) -> u64 {
(2.0 * (p95_commit_s + p95_scan_s) + 1.0).ceil() as u64
}
pub fn hold_clamped(raw: u64) -> u64 {
raw.clamp(HOLD_FLOOR_S, HOLD_CEILING_S)
}
pub fn fits(hold_s: u64, est_commit_s: f64, p95_scan_s: f64) -> bool {
hold_for(est_commit_s, p95_scan_s) <= hold_s
}
pub const FILES_SCALE1: usize = 152;
pub const FILES_SCALE10: usize = 1520;
pub const P95_COMMIT_S_SCALE1: f64 = 10.123;
pub const P95_COMMIT_S_SCALE10: f64 = 85.496;
pub const P95_SCAN_S: f64 = 1.579;
pub const HOLD_S: u64 = 176;
const FIT_PER_FILE_S: f64 =
(P95_COMMIT_S_SCALE10 - P95_COMMIT_S_SCALE1) / (FILES_SCALE10 as f64 - FILES_SCALE1 as f64);
const FIT_FIXED_S: f64 = {
let fixed = P95_COMMIT_S_SCALE1 - FILES_SCALE1 as f64 * FIT_PER_FILE_S;
if fixed > 0.0 { fixed } else { 0.0 }
};
pub const LOCK_MARGIN_S: i64 = 2;
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct HoldModel {
pub hold_s: u64,
pub commit_fixed_s: f64,
pub commit_per_file_s: f64,
pub p95_scan_s: f64,
}
impl HoldModel {
pub const MEASURED: HoldModel = HoldModel {
hold_s: HOLD_S,
commit_fixed_s: FIT_FIXED_S,
commit_per_file_s: FIT_PER_FILE_S,
p95_scan_s: P95_SCAN_S,
};
pub fn estimate_commit_s(&self, files: usize) -> f64 {
self.commit_fixed_s + files as f64 * self.commit_per_file_s
}
pub fn fits(&self, files: usize) -> bool {
fits(self.hold_s, self.estimate_commit_s(files), self.p95_scan_s)
}
}
pub trait Clock: Send + Sync {
fn now(&self) -> Instant;
}
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemClock;
impl Clock for SystemClock {
fn now(&self) -> Instant {
Instant::now()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct LockIdentity {
timestamp: Option<i64>,
actor: Option<String>,
}
impl LockIdentity {
fn of(info: &ScanLockInfo) -> Self {
LockIdentity {
timestamp: info.timestamp,
actor: info.actor.clone(),
}
}
}
#[derive(Debug, Clone)]
pub struct HeldLock {
kind: ScanKind,
hold_s: u64,
acquired: Instant,
identity: Option<LockIdentity>,
}
impl HeldLock {
pub async fn acquire(
deploy: &dyn DeployApi,
kind: ScanKind,
hold_s: u64,
clock: &dyn Clock,
) -> Result<HeldLock, CoreError> {
let acquired = clock.now();
deploy
.scan_lock_acquire(kind, ACQUIRE_WAIT_S, hold_s)
.await?;
let identity = match deploy.scan_lock_info(kind).await? {
LockObservation::Held(info) => Some(LockIdentity::of(&info)),
LockObservation::Free => None,
};
Ok(HeldLock {
kind,
hold_s,
acquired,
identity,
})
}
pub fn kind(&self) -> ScanKind {
self.kind
}
pub fn hold_s(&self) -> u64 {
self.hold_s
}
pub fn local_remaining_s(&self, clock: &dyn Clock) -> i64 {
let elapsed = clock.now().saturating_duration_since(self.acquired);
let elapsed_s = elapsed.as_secs() + u64::from(elapsed.subsec_nanos() > 0);
i64::try_from(self.hold_s)
.unwrap_or(i64::MAX)
.saturating_sub(i64::try_from(elapsed_s).unwrap_or(i64::MAX))
}
fn is_ours(&self, info: &ScanLockInfo) -> bool {
self.identity.as_ref() == Some(&LockIdentity::of(info))
}
pub async fn check_gateway(
&self,
deploy: &dyn DeployApi,
margin_s: i64,
) -> Result<LockVerdict, CoreError> {
let observed = deploy.scan_lock_info(self.kind).await?;
Ok(match &observed {
LockObservation::Held(info) if !self.is_ours(info) => LockVerdict::NotHeld,
_ => lock_verdict(&observed, margin_s),
})
}
pub async fn check(
&self,
deploy: &dyn DeployApi,
margin_s: i64,
clock: &dyn Clock,
) -> Result<LockVerdict, CoreError> {
let local_s = self.local_remaining_s(clock);
Ok(match self.check_gateway(deploy, margin_s).await? {
LockVerdict::Held { remaining_s } if local_s < margin_s => LockVerdict::Expiring {
remaining_s: remaining_s.min(local_s),
},
verdict => verdict,
})
}
pub async fn wait_for_lapse(
&self,
deploy: &dyn DeployApi,
interval: Duration,
) -> Result<(), CoreError> {
let limit = Duration::from_secs(self.hold_s.saturating_add(LOCK_MARGIN_S as u64 + 3));
let started = tokio::time::Instant::now();
loop {
if self.check_gateway(deploy, 0).await? == LockVerdict::NotHeld {
return Ok(());
}
if started.elapsed() >= limit {
return Err(CoreError::Internal(
"the scan-lock did not lapse within its hold".to_string(),
));
}
tokio::time::sleep(interval).await;
}
}
}
#[doc(hidden)]
#[derive(Debug)]
pub struct ManualClock {
base: Instant,
offset: Mutex<Duration>,
}
impl Default for ManualClock {
fn default() -> Self {
Self::new()
}
}
impl ManualClock {
pub fn new() -> Self {
ManualClock {
base: Instant::now(),
offset: Mutex::new(Duration::ZERO),
}
}
pub fn advance(&self, by: Duration) {
*self.offset.lock().expect("manual clock") += by;
}
}
impl Clock for ManualClock {
fn now(&self) -> Instant {
self.base + *self.offset.lock().expect("manual clock")
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use super::*;
use crate::client::config_resources::{
MutationOutcome, ResourceCreate, ResourceName, ResourceRecord, ResourceUpdate,
};
use crate::client::mode::{ModeCreate, ModeRecord};
use crate::client::query::{ListEnvelope, ListQuery};
use crate::client::scan::{LockObservation, ScanLockInfo, ScanStatus};
#[test]
fn hold_formula_regression() {
assert_eq!(hold_for(251.063, 0.888), 505);
assert_eq!(hold_for(348.225, 0.642), 699);
assert!(hold_for(251.063, 0.888) > HOLD_CEILING_S);
assert_eq!(hold_clamped(505), HOLD_CEILING_S);
assert_eq!(hold_clamped(699), HOLD_CEILING_S);
assert_eq!(hold_clamped(HOLD_CEILING_S), HOLD_CEILING_S);
assert_eq!(hold_clamped(120), 120);
}
#[test]
fn hold_floor() {
assert_eq!(hold_for(1.0, 0.5), 4);
assert_eq!(hold_clamped(4), HOLD_FLOOR_S);
assert_eq!(hold_clamped(0), HOLD_FLOOR_S);
assert_eq!(hold_clamped(HOLD_FLOOR_S), HOLD_FLOOR_S);
}
#[test]
fn d02_numbers_are_the_users() {
assert_eq!((HOLD_FLOOR_S, HOLD_CEILING_S), (10, 300));
}
const RECORDED_COMMIT_S_152: f64 = 10.123;
const RECORDED_COMMIT_S_1520: f64 = 85.496;
const RECORDED_SCAN_S: f64 = 1.579;
const RECORDED_HOLD_S: u64 = 176;
#[test]
fn hold_s_is_derived_from_recorded_measurements() {
assert_eq!(P95_COMMIT_S_SCALE1, RECORDED_COMMIT_S_152);
assert_eq!(P95_COMMIT_S_SCALE10, RECORDED_COMMIT_S_1520);
assert_eq!(P95_SCAN_S, RECORDED_SCAN_S);
assert_eq!((FILES_SCALE1, FILES_SCALE10), (152, 1520));
let unclamped = hold_for(RECORDED_COMMIT_S_1520, RECORDED_SCAN_S);
assert_eq!(
unclamped, 176,
"ceil(2 x (85.496 + 1.579) + 1) = ceil(175.15)"
);
assert!(
unclamped <= HOLD_CEILING_S,
"D-03a is not triggered: the unclamped hold is inside the ceiling"
);
assert_eq!(HOLD_S, hold_clamped(unclamped));
assert_eq!(HOLD_S, RECORDED_HOLD_S);
let per_file = (RECORDED_COMMIT_S_1520 - RECORDED_COMMIT_S_152) / (1520.0 - 152.0);
let fixed = (RECORDED_COMMIT_S_152 - 152.0 * per_file).max(0.0);
assert!(fixed > 0.0, "the floor at 0 did not apply");
assert_eq!(HoldModel::MEASURED.commit_per_file_s, per_file);
assert_eq!(HoldModel::MEASURED.commit_fixed_s, fixed);
assert_eq!(HoldModel::MEASURED.p95_scan_s, RECORDED_SCAN_S);
assert_eq!(HoldModel::MEASURED.hold_s, HOLD_S);
assert!((HoldModel::MEASURED.estimate_commit_s(152) - RECORDED_COMMIT_S_152).abs() < 1e-9);
assert!(
(HoldModel::MEASURED.estimate_commit_s(1520) - RECORDED_COMMIT_S_1520).abs() < 1e-9
);
}
#[test]
fn hold_model_fits_the_measured_largest_tree() {
let model = HoldModel::MEASURED;
assert!(model.fits(1520), "the largest measured tree fits");
assert!(model.fits(0));
let first_over = (1520..4000)
.find(|files| !model.fits(*files))
.expect("some size does not fit");
assert_eq!(
first_over,
1528,
"1,527 files is the largest change that fits {} s; 1,528 needs {} s",
model.hold_s,
hold_for(model.estimate_commit_s(1528), model.p95_scan_s)
);
assert!(model.fits(first_over - 1));
assert!(!model.fits(first_over));
assert_eq!(
hold_for(model.estimate_commit_s(1520), model.p95_scan_s),
HOLD_S
);
}
#[test]
fn hold_fit_boundary() {
let needed = hold_for(6.0, 0.5);
assert_eq!(needed, 14);
assert!(fits(needed, 6.0, 0.5), "equal fits");
assert!(!fits(needed - 1, 6.0, 0.5), "one second over does not");
let model = HoldModel {
hold_s: 14,
commit_fixed_s: 1.0,
commit_per_file_s: 0.5,
p95_scan_s: 0.5,
};
assert_eq!(model.estimate_commit_s(10), 6.0);
assert!(model.fits(10), "needs exactly 14 s");
assert!(!model.fits(11), "needs 16 s");
let tight = HoldModel {
hold_s: 13,
..model
};
assert!(!tight.fits(10), "one second short");
}
struct LockGateway {
info: Mutex<LockObservation>,
acquired: Mutex<Vec<(ScanKind, u64, u64)>>,
}
impl LockGateway {
fn holding(remaining: i64, timestamp: Option<i64>) -> Self {
LockGateway {
info: Mutex::new(LockObservation::Held(ScanLockInfo {
timestamp,
remaining,
actor: Some("api-token:t:127.0.0.1".to_string()),
extra: BTreeMap::new(),
})),
acquired: Mutex::new(Vec::new()),
}
}
fn free() -> Self {
LockGateway {
info: Mutex::new(LockObservation::Free),
acquired: Mutex::new(Vec::new()),
}
}
}
fn unused<T>() -> Result<T, CoreError> {
Err(CoreError::Internal(
"not used by the hold tests".to_string(),
))
}
#[async_trait::async_trait]
impl DeployApi for LockGateway {
async fn resource_names(
&self,
_module: &str,
_ty: &str,
) -> Result<ListEnvelope<ResourceName>, CoreError> {
unused()
}
async fn resource_list(
&self,
_module: &str,
_ty: &str,
_q: &ListQuery,
) -> Result<ListEnvelope<ResourceRecord>, CoreError> {
unused()
}
async fn resource_find(
&self,
_module: &str,
_ty: &str,
_name: &str,
_collection: Option<&str>,
) -> Result<ResourceRecord, CoreError> {
unused()
}
async fn resource_singleton(
&self,
_module: &str,
_ty: &str,
_collection: Option<&str>,
) -> Result<ResourceRecord, CoreError> {
unused()
}
async fn resource_create(
&self,
_module: &str,
_ty: &str,
_items: &[ResourceCreate],
) -> Result<MutationOutcome, CoreError> {
unused()
}
async fn resource_update(
&self,
_module: &str,
_ty: &str,
_items: &[ResourceUpdate],
) -> Result<MutationOutcome, CoreError> {
unused()
}
async fn scan_lock_acquire(
&self,
kind: ScanKind,
acquire_s: u64,
hold_s: u64,
) -> Result<(), CoreError> {
self.acquired
.lock()
.expect("acquired")
.push((kind, acquire_s, hold_s));
Ok(())
}
async fn scan_lock_info(&self, _kind: ScanKind) -> Result<LockObservation, CoreError> {
Ok(self.info.lock().expect("info").clone())
}
async fn scan_status(&self, _kind: ScanKind) -> Result<ScanStatus, CoreError> {
unused()
}
async fn scan_request(&self, _kind: ScanKind) -> Result<(), CoreError> {
unused()
}
async fn modes(&self) -> Result<ListEnvelope<ModeRecord>, CoreError> {
unused()
}
async fn mode_create(&self, _body: &ModeCreate) -> Result<ModeRecord, CoreError> {
unused()
}
async fn deployment_mode(&self) -> Result<Option<String>, CoreError> {
unused()
}
async fn restart_pending(&self) -> Result<Vec<String>, CoreError> {
unused()
}
async fn redundancy_resync(&self) -> Result<(), CoreError> {
unused()
}
async fn secret_names(
&self,
_provider: &str,
) -> Result<crate::client::deploy_api::SecretNameListing, CoreError> {
unused()
}
}
const HOLD: u64 = 60;
async fn lock_on(gateway: &LockGateway, clock: &ManualClock) -> HeldLock {
HeldLock::acquire(gateway, ScanKind::Config, HOLD, clock)
.await
.expect("acquire")
}
#[tokio::test]
async fn held_lock_acquires_the_config_lock_with_the_hold_and_the_smallest_wait() {
let gateway = LockGateway::holding(HOLD as i64, None);
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
assert_eq!(
*gateway.acquired.lock().expect("acquired"),
[(ScanKind::Config, ACQUIRE_WAIT_S, HOLD)]
);
assert_eq!((lock.kind(), lock.hold_s()), (ScanKind::Config, HOLD));
}
#[tokio::test]
async fn held_lock_is_held_while_both_clocks_have_the_margin() {
let gateway = LockGateway::holding(30, None);
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
clock.advance(Duration::from_secs(5));
let verdict = lock
.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check");
assert_eq!(verdict, LockVerdict::Held { remaining_s: 30 });
}
#[tokio::test]
async fn held_lock_local_clock_expiry() {
let gateway = LockGateway::holding(30, None);
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
clock.advance(Duration::from_secs(HOLD - LOCK_MARGIN_S as u64));
assert_eq!(lock.local_remaining_s(&clock), LOCK_MARGIN_S);
assert!(matches!(
lock.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check"),
LockVerdict::Held { .. }
));
clock.advance(Duration::from_secs(1));
assert_eq!(lock.local_remaining_s(&clock), LOCK_MARGIN_S - 1);
let verdict = lock
.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check");
assert_eq!(
verdict,
LockVerdict::Expiring {
remaining_s: LOCK_MARGIN_S - 1
}
);
clock.advance(Duration::from_secs(HOLD));
assert!(lock.local_remaining_s(&clock) < 0);
}
#[tokio::test]
async fn held_lock_gateway_expiring() {
let gateway = LockGateway::holding(1, None);
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
let verdict = lock
.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check");
assert_eq!(verdict, LockVerdict::Expiring { remaining_s: 1 });
}
#[tokio::test]
async fn held_lock_free_on_the_gateway_is_not_held() {
let gateway = LockGateway::free();
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
let verdict = lock
.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check");
assert_eq!(verdict, LockVerdict::NotHeld);
}
#[tokio::test]
async fn held_lock_never_reads_the_gateway_timestamp() {
let clock = ManualClock::new();
let mut verdicts = Vec::new();
for timestamp in [None, Some(0), Some(i64::MAX), Some(i64::MIN)] {
let gateway = LockGateway::holding(30, timestamp);
let lock = lock_on(&gateway, &clock).await;
verdicts.push(
lock.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check"),
);
}
assert!(
verdicts
.iter()
.all(|v| *v == LockVerdict::Held { remaining_s: 30 })
);
}
fn held_by(timestamp: i64, actor: &str, remaining: i64) -> LockObservation {
LockObservation::Held(ScanLockInfo {
timestamp: Some(timestamp),
remaining,
actor: Some(actor.to_string()),
extra: BTreeMap::new(),
})
}
#[tokio::test]
async fn held_lock_another_actors_hold_is_not_ours() {
let gateway = LockGateway::holding(30, Some(100));
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
*gateway.info.lock().expect("info") = held_by(200, "api-token:t:127.0.0.1", 59);
assert_eq!(
lock.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check"),
LockVerdict::NotHeld
);
*gateway.info.lock().expect("info") = held_by(100, "api-token:other:127.0.0.1", 59);
assert_eq!(
lock.check_gateway(&gateway, 0).await.expect("check"),
LockVerdict::NotHeld
);
*gateway.info.lock().expect("info") = held_by(100, "api-token:t:127.0.0.1", 29);
assert_eq!(
lock.check_gateway(&gateway, 0).await.expect("check"),
LockVerdict::Held { remaining_s: 29 }
);
}
#[tokio::test]
async fn held_lock_unseen_after_the_grant_owns_no_later_lock() {
let gateway = LockGateway::free();
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
*gateway.info.lock().expect("info") = held_by(5, "api-token:t:127.0.0.1", 59);
assert_eq!(
lock.check(&gateway, LOCK_MARGIN_S, &clock)
.await
.expect("check"),
LockVerdict::NotHeld
);
}
#[tokio::test]
async fn held_lock_check_gateway_ignores_the_local_clock() {
let gateway = LockGateway::holding(30, Some(100));
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
clock.advance(Duration::from_secs(HOLD * 2));
assert!(lock.local_remaining_s(&clock) < 0);
assert_eq!(
lock.check_gateway(&gateway, LOCK_MARGIN_S)
.await
.expect("check"),
LockVerdict::Held { remaining_s: 30 }
);
}
#[tokio::test]
async fn held_lock_wait_for_lapse_returns_when_the_hold_is_gone() {
let gateway = LockGateway::holding(30, Some(100));
let clock = ManualClock::new();
let lock = lock_on(&gateway, &clock).await;
*gateway.info.lock().expect("info") = held_by(7, "api-token:other:127.0.0.1", 59);
lock.wait_for_lapse(&gateway, Duration::from_millis(1))
.await
.expect("not ours");
*gateway.info.lock().expect("info") = LockObservation::Free;
lock.wait_for_lapse(&gateway, Duration::from_millis(1))
.await
.expect("free");
}
#[test]
fn manual_clock_moves_only_when_told() {
let clock = ManualClock::new();
let start = clock.now();
assert_eq!(clock.now(), start);
clock.advance(Duration::from_secs(3));
assert_eq!(clock.now() - start, Duration::from_secs(3));
}
}