From 0ca523ec41c7dcf7e26cabf4f933d0901b23e09f Mon Sep 17 00:00:00 2001 From: Lewis Date: Mon, 03 Aug 2026 15:36:48 +0000 Subject: [PATCH] knot2/resource,fixtures: pace repeation lookups, turn off git auto-maintenance Lewis: May this revision serve well! --- knot2/crates/knot-fixtures/src/lib.rs | 5 +++++ knot2/crates/knot-lfs/src/admission.rs | 57 +++++++++++++++++++++++++++++++++++++-------------------- knot2/crates/knot-pack/tests/common/mod.rs | 14 ++++++++++++-- knot2/crates/knot-resource/src/admission.rs | 229 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----- knot2/crates/knot-resource/src/disk.rs | 71 ++++++++++++++++++++++++++++++++++++----------------------------------- knot2/crates/knot-resource/src/lib.rs | 6 +++--- 6 file(s) changed, 317 insertion(s)(+), 65 deletion(s)(-) diff --git a/knot2/crates/knot-fixtures/src/lib.rs b/knot2/crates/knot-fixtures/src/lib.rs --- a/knot2/crates/knot-fixtures/src/lib.rs +++ b/knot2/crates/knot-fixtures/src/lib.rs @@ -12,6 +12,11 @@ command .current_dir(cwd) .env("GIT_CONFIG_GLOBAL", "/dev/null") .env("GIT_CONFIG_SYSTEM", "/dev/null") + .env("GIT_CONFIG_COUNT", "2") + .env("GIT_CONFIG_KEY_0", "maintenance.auto") + .env("GIT_CONFIG_VALUE_0", "false") + .env("GIT_CONFIG_KEY_1", "gc.autoDetach") + .env("GIT_CONFIG_VALUE_1", "false") .env("GIT_TERMINAL_PROMPT", "0") .env("GIT_ASKPASS", "true") .env("GIT_AUTHOR_NAME", AUTHOR_NAME) diff --git a/knot2/crates/knot-lfs/src/admission.rs b/knot2/crates/knot-lfs/src/admission.rs --- a/knot2/crates/knot-lfs/src/admission.rs +++ b/knot2/crates/knot-lfs/src/admission.rs @@ -1,4 +1,4 @@ -use knot_resource::{DiskGovernor, DiskReservation, ReserveError}; +use knot_resource::{BelowFloor, DiskGovernor, DiskReservation, FreeBytes}; use crate::{ClaimedSize, FreeSpaceFloor, LfsError, LfsSize, LfsStorePath}; @@ -53,14 +53,34 @@ } if self.floor.get() == 0 { return Ok(UploadPermit::unreserved()); } - match self.governor.reserve( - self.root.as_path(), - knot_resource::ReserveBytes::new(declared.get()), - ) { + let free = + knot_resource::disk_free_bytes(self.root.as_path()).map_err(|source| LfsError::Io { + op: "probe free space under", + path: self.root.as_path().to_path_buf(), + source, + })?; + self.admit_against(free, declared) + } + + fn max_object(&self) -> LfsSize { + self.max_object + } +} + +impl StoreAdmission { + fn admit_against( + &self, + free: FreeBytes, + declared: ClaimedSize, + ) -> Result { + match self + .governor + .reserve_against(free, knot_resource::ReserveBytes::new(declared.get())) + { Ok(reservation) => Ok(UploadPermit { _reservation: Some(reservation), }), - Err(ReserveError::BelowFloor { free, .. }) => { + Err(BelowFloor { free, .. }) => { tracing::warn!( declared = declared.get(), free = free.get(), @@ -72,16 +92,7 @@ free: LfsSize::new(free.get()), floor: self.floor, }) } - Err(ReserveError::Probe(source)) => Err(LfsError::Io { - op: "probe free space under", - path: self.root.as_path().to_path_buf(), - source, - }), } - } - - fn max_object(&self) -> LfsSize { - self.max_object } } @@ -140,21 +151,27 @@ #[test] fn a_held_permit_reserves_against_the_next_admission() { let dir = tempfile::tempdir().unwrap(); - let free = knot_resource::disk_free_bytes(dir.path()).unwrap(); + let free = FreeBytes::new(10_240); let gate = StoreAdmission::new( LfsStorePath::new(dir.path()), LfsSize::new(u64::MAX), - FreeSpaceFloor::new(free.get().saturating_sub(6_144)), + FreeSpaceFloor::new(4_096), ); - let held = gate.admit(ClaimedSize::new(4_096)).unwrap(); + let held = gate + .admit_against(free, ClaimedSize::new(4_096)) + .expect("a free-space reading the floor leaves room in admits a lone upload"); assert!( matches!( - gate.admit(ClaimedSize::new(4_096)), + gate.admit_against(free, ClaimedSize::new(4_096)), Err(LfsError::FreeSpaceDenied { .. }) ), "a second upload cannot pass the floor while the first is in flight" ); drop(held); - assert!(gate.admit(ClaimedSize::new(4_096)).is_ok()); + assert!( + gate.admit_against(free, ClaimedSize::new(4_096)).is_ok(), + "the same reading must admit again once the first upload finishes, or the refusal \ + above was the floor rather than the reservation still being held" + ); } } diff --git a/knot2/crates/knot-pack/tests/common/mod.rs b/knot2/crates/knot-pack/tests/common/mod.rs --- a/knot2/crates/knot-pack/tests/common/mod.rs +++ b/knot2/crates/knot-pack/tests/common/mod.rs @@ -35,7 +35,12 @@ .unwrap() .write_all(oids.join("\n").as_bytes()) .unwrap(); let out = child.wait_with_output().unwrap(); - assert!(out.status.success(), "pack-objects failed"); + assert!( + out.status.success(), + "pack-objects failed ({}): {}", + out.status, + String::from_utf8_lossy(&out.stderr) + ); out.stdout } @@ -74,7 +79,12 @@ .unwrap() .write_all(oids.join("\n").as_bytes()) .unwrap(); let out = child.wait_with_output().unwrap(); - assert!(out.status.success(), "pack-objects failed"); + assert!( + out.status.success(), + "pack-objects failed ({}): {}", + out.status, + String::from_utf8_lossy(&out.stderr) + ); out.stdout } diff --git a/knot2/crates/knot-resource/src/admission.rs b/knot2/crates/knot-resource/src/admission.rs --- a/knot2/crates/knot-resource/src/admission.rs +++ b/knot2/crates/knot-resource/src/admission.rs @@ -1,10 +1,14 @@ use std::collections::HashMap; +use std::hash::Hash; use std::net::IpAddr; use std::sync::{Arc, Mutex}; +use std::time::Duration; use knot_types::UnixMicros; const MAX_TRACKED_PEERS: usize = 100_000; + +const MAX_PACED_KEYS: usize = 4_096; const SWEEP_INTERVAL_MICROS: u64 = 1_000_000; @@ -22,10 +26,10 @@ pub refill: RefillMicros, } impl RateLimit { - const fn interval(self) -> u64 { + pub const fn interval(self) -> RefillMicros { match self.refill.get() { - 0 => 1, - micros => micros, + 0 => RefillMicros::new(1), + micros => RefillMicros::new(micros), } } } @@ -83,12 +87,12 @@ } fn tokens_at(&self, rate: RateLimit, now: UnixMicros) -> u32 { let elapsed = now.get().saturating_sub(self.last_refill.get()); - let gained = (elapsed / rate.interval()).min(u64::from(rate.burst.get())) as u32; + let gained = (elapsed / rate.interval().get()).min(u64::from(rate.burst.get())) as u32; self.tokens.saturating_add(gained).min(rate.burst.get()) } fn replenish(&mut self, rate: RateLimit, now: UnixMicros) -> bool { - if now.get().saturating_sub(self.last_refill.get()) >= rate.interval() { + if now.get().saturating_sub(self.last_refill.get()) >= rate.interval().get() { self.tokens = self.tokens_at(rate, now); self.last_refill = now; } @@ -314,6 +318,124 @@ self.limiter.leave(self.peer); } } +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct HostKey(String); + +impl HostKey { + pub fn new(host: &str) -> Self { + Self(host.to_ascii_lowercase()) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct SubjectKey(String); + +impl SubjectKey { + pub fn new(subject: &str) -> Self { + Self(subject.to_ascii_lowercase()) + } +} + +struct Booked { + turns: HashMap, + last_sweep: UnixMicros, +} + +impl Booked { + fn has_room_for(&mut self, key: &K, now: UnixMicros) -> bool { + if self.turns.contains_key(key) || self.turns.len() < MAX_PACED_KEYS { + return true; + } + if now.get().saturating_sub(self.last_sweep.get()) >= SWEEP_INTERVAL_MICROS { + self.last_sweep = now; + self.turns.retain(|_, until| until.get() > now.get()); + } + self.turns.len() < MAX_PACED_KEYS + } +} + +pub struct Pacer { + rate: RateLimit, + inner: Mutex>, +} + +pub type HostPacer = Pacer; +pub type SubjectPacer = Pacer; +pub type PeerPacer = Pacer; + +impl Pacer { + pub fn new(rate: RateLimit) -> Self { + Self { + rate, + inner: Mutex::new(Booked { + turns: HashMap::new(), + last_sweep: UnixMicros::new(0), + }), + } + } + + pub fn reserve(&self, key: &K, now: UnixMicros) -> Duration { + let mut booked = self.lock(); + match booked.has_room_for(key, now) { + false => Duration::from_micros(self.rate.interval().get()), + true => { + let wait = self.wait_for(&booked, key, now); + self.claim_turn(&mut booked, key, now); + wait + } + } + } + + pub fn reserve_now(&self, key: &K, now: UnixMicros) -> bool { + let mut booked = self.lock(); + match booked.has_room_for(key, now) { + false => false, + true => match self.wait_for(&booked, key, now).is_zero() { + false => false, + true => { + self.claim_turn(&mut booked, key, now); + true + } + }, + } + } + + fn wait_for(&self, booked: &Booked, key: &K, now: UnixMicros) -> Duration { + let tolerance = self + .rate + .interval() + .get() + .saturating_mul(u64::from(self.rate.burst.get().saturating_sub(1))); + Duration::from_micros( + self.turn(booked, key, now) + .get() + .saturating_sub(tolerance) + .saturating_sub(now.get()), + ) + } + + fn claim_turn(&self, booked: &mut Booked, key: &K, now: UnixMicros) { + let until = self + .turn(booked, key, now) + .get() + .saturating_add(self.rate.interval().get()); + booked.turns.insert(key.clone(), UnixMicros::new(until)); + } + + fn turn(&self, booked: &Booked, key: &K, now: UnixMicros) -> UnixMicros { + booked + .turns + .get(key) + .map_or(now, |until| UnixMicros::new(until.get().max(now.get()))) + } + + fn lock(&self) -> std::sync::MutexGuard<'_, Booked> { + self.inner + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -556,6 +678,103 @@ let tracked = tracked(&limiter); assert!( tracked <= MAX_TRACKED_PEERS, "a flood of distinct source addresses mustn't grow the peer map past its limit, saw {tracked}" + ); + } + + #[test] + fn a_pacer_spends_a_hosts_burst_at_once_then_spaces_it_and_leaves_every_other_host_alone() { + let pacer = HostPacer::new(RateLimit { + burst: Burst::new(3), + refill: RefillMicros::new(100), + }); + let plc = HostKey::new("plc.directory"); + let pds = HostKey::new("PDS.Nel.Pet"); + let waits: Vec = (0..5) + .map(|_| pacer.reserve(&plc, at(0)).as_micros()) + .collect(); + assert_eq!( + waits, + vec![0, 0, 0, 100, 200], + "a cold host takes its whole burst without waiting. Every visit after that is one \ + refill interval further out" + ); + assert_eq!( + pacer.reserve(&pds, at(0)), + Duration::ZERO, + "a knot whose accounts spread over many PDSes must fill at the sum of their rates, \ + so one busy host mustn't delay another" + ); + assert_eq!( + HostKey::new("PDS.Nel.Pet"), + HostKey::new("pds.nel.pet"), + "a PDS endpoint written in mixed case is the same host and must share its schedule" + ); + assert_eq!( + pacer.reserve(&plc, at(10_000)), + Duration::ZERO, + "a host the caller hasn't visited since its last turn is due immediately" + ); + } + + #[test] + fn a_turn_that_isnt_due_is_refused_without_pushing_the_schedule_further_out() { + let pacer = SubjectPacer::new(RateLimit { + burst: Burst::new(1), + refill: RefillMicros::new(1_000), + }); + let nel = SubjectKey::new("did:plc:nel"); + assert!( + pacer.reserve_now(&nel, at(0)), + "a subject the caller hasn't read takes its turn straight away" + ); + assert!( + !pacer.reserve_now(&nel, at(500)), + "a caller that mustn't wait is refused inside the interval" + ); + assert!( + !pacer.reserve_now(&nel, at(999)), + "refusals must leave the booking alone, or whoever keeps trying pushes the turn \ + further out every time and the knot never reads the subject again" + ); + assert!(pacer.reserve_now(&nel, at(1_000))); + assert_eq!( + SubjectKey::new("DID:PLC:NEL"), + SubjectKey::new("did:plc:nel"), + "a DID a client typed in mixed case is the same account and shares its schedule" + ); + } + + #[test] + fn a_flood_of_distinct_hosts_doesnt_grow_the_schedule_past_its_limit_or_delay_a_newcomer() { + let pacer = HostPacer::new(RateLimit { + burst: Burst::new(1), + refill: RefillMicros::new(1_000_000), + }); + (0..MAX_PACED_KEYS as u64 + 5_000).for_each(|index| { + let _ = pacer.reserve(&HostKey::new(&format!("{index}.nel.pet")), at(0)); + }); + let tracked = pacer.lock().turns.len(); + assert!( + tracked <= MAX_PACED_KEYS, + "a grant set spread over more did:web hosts than the schedule can track mustn't \ + grow it past its limit, saw {tracked}" + ); + assert_eq!( + pacer.reserve(&HostKey::new("plc.directory"), at(0)), + Duration::from_micros(1_000_000), + "a host the full schedule can't track waits one refill interval, so a caller that \ + outgrows the schedule slows itself down" + ); + let settled = at(SWEEP_INTERVAL_MICROS + 2_000_000); + assert_eq!( + pacer.reserve(&HostKey::new("plc.directory"), settled), + Duration::ZERO + ); + assert_eq!( + pacer.lock().turns.len(), + 1, + "the sweep reclaims the schedule and tracks the newcomer once every booked turn \ + has passed" ); } diff --git a/knot2/crates/knot-resource/src/disk.rs b/knot2/crates/knot-resource/src/disk.rs --- a/knot2/crates/knot-resource/src/disk.rs +++ b/knot2/crates/knot-resource/src/disk.rs @@ -19,12 +19,9 @@ .map_err(io::Error::from) } #[derive(Debug)] -pub enum ReserveError { - BelowFloor { - free: FreeBytes, - floor: DiskFloorBytes, - }, - Probe(io::Error), +pub struct BelowFloor { + pub free: FreeBytes, + pub floor: DiskFloorBytes, } struct Ledger { @@ -47,23 +44,16 @@ pub fn reserved_bytes(&self) -> u64 { self.0.reserved.load(Ordering::SeqCst) } - pub fn reserve( + pub fn reserve_against( &self, - path: &Path, + free: FreeBytes, bytes: ReserveBytes, - ) -> Result { + ) -> Result { let amount = bytes.get(); let projected = self.0.reserved.fetch_add(amount, Ordering::SeqCst) + amount; - let free = match free_bytes(path) { - Ok(free) => free, - Err(source) => { - self.0.reserved.fetch_sub(amount, Ordering::SeqCst); - return Err(ReserveError::Probe(source)); - } - }; if free.get() < self.0.floor.get().saturating_add(projected) { self.0.reserved.fetch_sub(amount, Ordering::SeqCst); - return Err(ReserveError::BelowFloor { + return Err(BelowFloor { free, floor: self.0.floor, }); @@ -103,13 +93,17 @@ } #[test] fn a_reservation_holds_bytes_until_it_drops() { - let dir = std::env::temp_dir(); + let free = FreeBytes::new(1 << 20); let governor = DiskGovernor::new(DiskFloorBytes::new(0)); assert_eq!(governor.reserved_bytes(), 0); { - let _held = governor.reserve(&dir, ReserveBytes::new(4_096)).unwrap(); + let _held = governor + .reserve_against(free, ReserveBytes::new(4_096)) + .unwrap(); assert_eq!(governor.reserved_bytes(), 4_096); - let _also = governor.reserve(&dir, ReserveBytes::new(1_024)).unwrap(); + let _also = governor + .reserve_against(free, ReserveBytes::new(1_024)) + .unwrap(); assert_eq!(governor.reserved_bytes(), 5_120); } assert_eq!(governor.reserved_bytes(), 0); @@ -117,30 +111,37 @@ } #[test] fn concurrent_reservations_cannot_jointly_punch_through_the_floor() { - let dir = std::env::temp_dir(); - let free = free_bytes(&dir).unwrap(); - let floor = DiskFloorBytes::new(free.get().saturating_sub(6_144)); - let governor = DiskGovernor::new(floor); - let first = governor.reserve(&dir, ReserveBytes::new(4_096)).unwrap(); - let denied = governor.reserve(&dir, ReserveBytes::new(4_096)); + let free = FreeBytes::new(10_240); + let governor = DiskGovernor::new(DiskFloorBytes::new(4_096)); + let first = governor + .reserve_against(free, ReserveBytes::new(4_096)) + .unwrap(); + let denied = governor.reserve_against(free, ReserveBytes::new(4_096)); assert!( - matches!(denied, Err(ReserveError::BelowFloor { .. })), + denied.is_err(), "the second reservation must see the first still held" ); assert_eq!(governor.reserved_bytes(), 4_096); drop(first); assert_eq!(governor.reserved_bytes(), 0); - assert!(governor.reserve(&dir, ReserveBytes::new(4_096)).is_ok()); + assert!( + governor + .reserve_against(free, ReserveBytes::new(4_096)) + .is_ok(), + "a free-space reading the floor leaves room in must admit a lone reservation, or \ + the refusal above was the floor rather than the reservation still being held" + ); } #[test] - fn a_probe_fault_leaves_the_ledger_untouched() { - let governor = DiskGovernor::new(DiskFloorBytes::new(0)); - let fault = governor.reserve( - Path::new("/definitely/not/a/mounted/path"), - ReserveBytes::new(4_096), + fn a_refused_reservation_leaves_the_ledger_untouched() { + let governor = DiskGovernor::new(DiskFloorBytes::new(4_096)); + let refused = governor.reserve_against(FreeBytes::new(4_096), ReserveBytes::new(1)); + assert!(refused.is_err()); + assert_eq!( + governor.reserved_bytes(), + 0, + "a refusal that left its bytes on the ledger would deny every later upload too" ); - assert!(matches!(fault, Err(ReserveError::Probe(_)))); - assert_eq!(governor.reserved_bytes(), 0); } } diff --git a/knot2/crates/knot-resource/src/lib.rs b/knot2/crates/knot-resource/src/lib.rs --- a/knot2/crates/knot-resource/src/lib.rs +++ b/knot2/crates/knot-resource/src/lib.rs @@ -6,12 +6,12 @@ mod mem; mod slots; pub use admission::{ - AdmitGuard, Burst, GlobalInflight, LimitConfig, PerPeerInflight, PreAuthLimiter, RateLimit, - RefillMicros, Refusal, + AdmitGuard, Burst, GlobalInflight, HostKey, HostPacer, LimitConfig, PeerPacer, PerPeerInflight, + PreAuthLimiter, RateLimit, RefillMicros, Refusal, SubjectKey, SubjectPacer, }; pub use cpu::{Saturate, ThreadCount, gix_thread_limit, map_chunks, map_spans, saturate, threads}; pub use disk::{ - DiskFloorBytes, DiskGovernor, DiskReservation, FreeBytes, ReserveBytes, ReserveError, + BelowFloor, DiskFloorBytes, DiskGovernor, DiskReservation, FreeBytes, ReserveBytes, free_bytes as disk_free_bytes, }; pub use fsio::{ -- tangled.sh