+5

knot2/crates/knot-fixtures/src/lib.rs

View file Reviewed

··· 12 12 .current_dir(cwd) 13 13 .env("GIT_CONFIG_GLOBAL", "/dev/null") 14 14 .env("GIT_CONFIG_SYSTEM", "/dev/null") 15 15 + .env("GIT_CONFIG_COUNT", "2") 16 16 + .env("GIT_CONFIG_KEY_0", "maintenance.auto") 17 17 + .env("GIT_CONFIG_VALUE_0", "false") 18 18 + .env("GIT_CONFIG_KEY_1", "gc.autoDetach") 19 19 + .env("GIT_CONFIG_VALUE_1", "false") 15 20 .env("GIT_TERMINAL_PROMPT", "0") 16 21 .env("GIT_ASKPASS", "true") 17 22 .env("GIT_AUTHOR_NAME", AUTHOR_NAME)

+37 -20

knot2/crates/knot-lfs/src/admission.rs

View file Reviewed

··· 1 1 - use knot_resource::{DiskGovernor, DiskReservation, ReserveError}; 1 1 + use knot_resource::{BelowFloor, DiskGovernor, DiskReservation, FreeBytes}; 2 2 3 3 use crate::{ClaimedSize, FreeSpaceFloor, LfsError, LfsSize, LfsStorePath}; 4 4 ··· 53 53 if self.floor.get() == 0 { 54 54 return Ok(UploadPermit::unreserved()); 55 55 } 56 56 - match self.governor.reserve( 57 57 - self.root.as_path(), 58 58 - knot_resource::ReserveBytes::new(declared.get()), 59 59 - ) { 56 56 + let free = 57 57 + knot_resource::disk_free_bytes(self.root.as_path()).map_err(|source| LfsError::Io { 58 58 + op: "probe free space under", 59 59 + path: self.root.as_path().to_path_buf(), 60 60 + source, 61 61 + })?; 62 62 + self.admit_against(free, declared) 63 63 + } 64 64 + 65 65 + fn max_object(&self) -> LfsSize { 66 66 + self.max_object 67 67 + } 68 68 + } 69 69 + 70 70 + impl StoreAdmission { 71 71 + fn admit_against( 72 72 + &self, 73 73 + free: FreeBytes, 74 74 + declared: ClaimedSize, 75 75 + ) -> Result<UploadPermit, LfsError> { 76 76 + match self 77 77 + .governor 78 78 + .reserve_against(free, knot_resource::ReserveBytes::new(declared.get())) 79 79 + { 60 80 Ok(reservation) => Ok(UploadPermit { 61 81 _reservation: Some(reservation), 62 82 }), 63 63 - Err(ReserveError::BelowFloor { free, .. }) => { 83 83 + Err(BelowFloor { free, .. }) => { 64 84 tracing::warn!( 65 85 declared = declared.get(), 66 86 free = free.get(), ··· 72 92 floor: self.floor, 73 93 }) 74 94 } 75 75 - Err(ReserveError::Probe(source)) => Err(LfsError::Io { 76 76 - op: "probe free space under", 77 77 - path: self.root.as_path().to_path_buf(), 78 78 - source, 79 79 - }), 80 95 } 81 81 - } 82 82 - 83 83 - fn max_object(&self) -> LfsSize { 84 84 - self.max_object 85 96 } 86 97 } 87 98 ··· 140 151 #[test] 141 152 fn a_held_permit_reserves_against_the_next_admission() { 142 153 let dir = tempfile::tempdir().unwrap(); 143 143 - let free = knot_resource::disk_free_bytes(dir.path()).unwrap(); 154 154 + let free = FreeBytes::new(10_240); 144 155 let gate = StoreAdmission::new( 145 156 LfsStorePath::new(dir.path()), 146 157 LfsSize::new(u64::MAX), 147 147 - FreeSpaceFloor::new(free.get().saturating_sub(6_144)), 158 158 + FreeSpaceFloor::new(4_096), 148 159 ); 149 149 - let held = gate.admit(ClaimedSize::new(4_096)).unwrap(); 160 160 + let held = gate 161 161 + .admit_against(free, ClaimedSize::new(4_096)) 162 162 + .expect("a free-space reading the floor leaves room in admits a lone upload"); 150 163 assert!( 151 164 matches!( 152 152 - gate.admit(ClaimedSize::new(4_096)), 165 165 + gate.admit_against(free, ClaimedSize::new(4_096)), 153 166 Err(LfsError::FreeSpaceDenied { .. }) 154 167 ), 155 168 "a second upload cannot pass the floor while the first is in flight" 156 169 ); 157 170 drop(held); 158 158 - assert!(gate.admit(ClaimedSize::new(4_096)).is_ok()); 171 171 + assert!( 172 172 + gate.admit_against(free, ClaimedSize::new(4_096)).is_ok(), 173 173 + "the same reading must admit again once the first upload finishes, or the refusal \ 174 174 + above was the floor rather than the reservation still being held" 175 175 + ); 159 176 } 160 177 }

+12 -2

knot2/crates/knot-pack/tests/common/mod.rs

View file Reviewed

··· 35 35 .write_all(oids.join("\n").as_bytes()) 36 36 .unwrap(); 37 37 let out = child.wait_with_output().unwrap(); 38 38 - assert!(out.status.success(), "pack-objects failed"); 38 38 + assert!( 39 39 + out.status.success(), 40 40 + "pack-objects failed ({}): {}", 41 41 + out.status, 42 42 + String::from_utf8_lossy(&out.stderr) 43 43 + ); 39 44 out.stdout 40 45 } 41 46 ··· 74 79 .write_all(oids.join("\n").as_bytes()) 75 80 .unwrap(); 76 81 let out = child.wait_with_output().unwrap(); 77 77 - assert!(out.status.success(), "pack-objects failed"); 82 82 + assert!( 83 83 + out.status.success(), 84 84 + "pack-objects failed ({}): {}", 85 85 + out.status, 86 86 + String::from_utf8_lossy(&out.stderr) 87 87 + ); 78 88 out.stdout 79 89 } 80 90

+224 -5

knot2/crates/knot-resource/src/admission.rs

View file Reviewed

··· 1 1 use std::collections::HashMap; 2 2 + use std::hash::Hash; 2 3 use std::net::IpAddr; 3 4 use std::sync::{Arc, Mutex}; 5 5 + use std::time::Duration; 4 6 5 7 use knot_types::UnixMicros; 6 8 7 9 const MAX_TRACKED_PEERS: usize = 100_000; 10 10 + 11 11 + const MAX_PACED_KEYS: usize = 4_096; 8 12 9 13 const SWEEP_INTERVAL_MICROS: u64 = 1_000_000; 10 14 ··· 22 26 } 23 27 24 28 impl RateLimit { 25 25 - const fn interval(self) -> u64 { 29 29 + pub const fn interval(self) -> RefillMicros { 26 30 match self.refill.get() { 27 27 - 0 => 1, 28 28 - micros => micros, 31 31 + 0 => RefillMicros::new(1), 32 32 + micros => RefillMicros::new(micros), 29 33 } 30 34 } 31 35 } ··· 83 87 84 88 fn tokens_at(&self, rate: RateLimit, now: UnixMicros) -> u32 { 85 89 let elapsed = now.get().saturating_sub(self.last_refill.get()); 86 86 - let gained = (elapsed / rate.interval()).min(u64::from(rate.burst.get())) as u32; 90 90 + let gained = (elapsed / rate.interval().get()).min(u64::from(rate.burst.get())) as u32; 87 91 self.tokens.saturating_add(gained).min(rate.burst.get()) 88 92 } 89 93 90 94 fn replenish(&mut self, rate: RateLimit, now: UnixMicros) -> bool { 91 91 - if now.get().saturating_sub(self.last_refill.get()) >= rate.interval() { 95 95 + if now.get().saturating_sub(self.last_refill.get()) >= rate.interval().get() { 92 96 self.tokens = self.tokens_at(rate, now); 93 97 self.last_refill = now; 94 98 } ··· 314 318 } 315 319 } 316 320 321 321 + #[derive(Debug, Clone, PartialEq, Eq, Hash)] 322 322 + pub struct HostKey(String); 323 323 + 324 324 + impl HostKey { 325 325 + pub fn new(host: &str) -> Self { 326 326 + Self(host.to_ascii_lowercase()) 327 327 + } 328 328 + } 329 329 + 330 330 + #[derive(Debug, Clone, PartialEq, Eq, Hash)] 331 331 + pub struct SubjectKey(String); 332 332 + 333 333 + impl SubjectKey { 334 334 + pub fn new(subject: &str) -> Self { 335 335 + Self(subject.to_ascii_lowercase()) 336 336 + } 337 337 + } 338 338 + 339 339 + struct Booked<K> { 340 340 + turns: HashMap<K, UnixMicros>, 341 341 + last_sweep: UnixMicros, 342 342 + } 343 343 + 344 344 + impl<K: Eq + Hash> Booked<K> { 345 345 + fn has_room_for(&mut self, key: &K, now: UnixMicros) -> bool { 346 346 + if self.turns.contains_key(key) || self.turns.len() < MAX_PACED_KEYS { 347 347 + return true; 348 348 + } 349 349 + if now.get().saturating_sub(self.last_sweep.get()) >= SWEEP_INTERVAL_MICROS { 350 350 + self.last_sweep = now; 351 351 + self.turns.retain(|_, until| until.get() > now.get()); 352 352 + } 353 353 + self.turns.len() < MAX_PACED_KEYS 354 354 + } 355 355 + } 356 356 + 357 357 + pub struct Pacer<K> { 358 358 + rate: RateLimit, 359 359 + inner: Mutex<Booked<K>>, 360 360 + } 361 361 + 362 362 + pub type HostPacer = Pacer<HostKey>; 363 363 + pub type SubjectPacer = Pacer<SubjectKey>; 364 364 + pub type PeerPacer = Pacer<IpAddr>; 365 365 + 366 366 + impl<K: Clone + Eq + Hash> Pacer<K> { 367 367 + pub fn new(rate: RateLimit) -> Self { 368 368 + Self { 369 369 + rate, 370 370 + inner: Mutex::new(Booked { 371 371 + turns: HashMap::new(), 372 372 + last_sweep: UnixMicros::new(0), 373 373 + }), 374 374 + } 375 375 + } 376 376 + 377 377 + pub fn reserve(&self, key: &K, now: UnixMicros) -> Duration { 378 378 + let mut booked = self.lock(); 379 379 + match booked.has_room_for(key, now) { 380 380 + false => Duration::from_micros(self.rate.interval().get()), 381 381 + true => { 382 382 + let wait = self.wait_for(&booked, key, now); 383 383 + self.claim_turn(&mut booked, key, now); 384 384 + wait 385 385 + } 386 386 + } 387 387 + } 388 388 + 389 389 + pub fn reserve_now(&self, key: &K, now: UnixMicros) -> bool { 390 390 + let mut booked = self.lock(); 391 391 + match booked.has_room_for(key, now) { 392 392 + false => false, 393 393 + true => match self.wait_for(&booked, key, now).is_zero() { 394 394 + false => false, 395 395 + true => { 396 396 + self.claim_turn(&mut booked, key, now); 397 397 + true 398 398 + } 399 399 + }, 400 400 + } 401 401 + } 402 402 + 403 403 + fn wait_for(&self, booked: &Booked<K>, key: &K, now: UnixMicros) -> Duration { 404 404 + let tolerance = self 405 405 + .rate 406 406 + .interval() 407 407 + .get() 408 408 + .saturating_mul(u64::from(self.rate.burst.get().saturating_sub(1))); 409 409 + Duration::from_micros( 410 410 + self.turn(booked, key, now) 411 411 + .get() 412 412 + .saturating_sub(tolerance) 413 413 + .saturating_sub(now.get()), 414 414 + ) 415 415 + } 416 416 + 417 417 + fn claim_turn(&self, booked: &mut Booked<K>, key: &K, now: UnixMicros) { 418 418 + let until = self 419 419 + .turn(booked, key, now) 420 420 + .get() 421 421 + .saturating_add(self.rate.interval().get()); 422 422 + booked.turns.insert(key.clone(), UnixMicros::new(until)); 423 423 + } 424 424 + 425 425 + fn turn(&self, booked: &Booked<K>, key: &K, now: UnixMicros) -> UnixMicros { 426 426 + booked 427 427 + .turns 428 428 + .get(key) 429 429 + .map_or(now, |until| UnixMicros::new(until.get().max(now.get()))) 430 430 + } 431 431 + 432 432 + fn lock(&self) -> std::sync::MutexGuard<'_, Booked<K>> { 433 433 + self.inner 434 434 + .lock() 435 435 + .unwrap_or_else(|poisoned| poisoned.into_inner()) 436 436 + } 437 437 + } 438 438 + 317 439 #[cfg(test)] 318 440 mod tests { 319 441 use super::*; ··· 556 678 assert!( 557 679 tracked <= MAX_TRACKED_PEERS, 558 680 "a flood of distinct source addresses mustn't grow the peer map past its limit, saw {tracked}" 681 681 + ); 682 682 + } 683 683 + 684 684 + #[test] 685 685 + fn a_pacer_spends_a_hosts_burst_at_once_then_spaces_it_and_leaves_every_other_host_alone() { 686 686 + let pacer = HostPacer::new(RateLimit { 687 687 + burst: Burst::new(3), 688 688 + refill: RefillMicros::new(100), 689 689 + }); 690 690 + let plc = HostKey::new("plc.directory"); 691 691 + let pds = HostKey::new("PDS.Nel.Pet"); 692 692 + let waits: Vec<u128> = (0..5) 693 693 + .map(|_| pacer.reserve(&plc, at(0)).as_micros()) 694 694 + .collect(); 695 695 + assert_eq!( 696 696 + waits, 697 697 + vec![0, 0, 0, 100, 200], 698 698 + "a cold host takes its whole burst without waiting. Every visit after that is one \ 699 699 + refill interval further out" 700 700 + ); 701 701 + assert_eq!( 702 702 + pacer.reserve(&pds, at(0)), 703 703 + Duration::ZERO, 704 704 + "a knot whose accounts spread over many PDSes must fill at the sum of their rates, \ 705 705 + so one busy host mustn't delay another" 706 706 + ); 707 707 + assert_eq!( 708 708 + HostKey::new("PDS.Nel.Pet"), 709 709 + HostKey::new("pds.nel.pet"), 710 710 + "a PDS endpoint written in mixed case is the same host and must share its schedule" 711 711 + ); 712 712 + assert_eq!( 713 713 + pacer.reserve(&plc, at(10_000)), 714 714 + Duration::ZERO, 715 715 + "a host the caller hasn't visited since its last turn is due immediately" 716 716 + ); 717 717 + } 718 718 + 719 719 + #[test] 720 720 + fn a_turn_that_isnt_due_is_refused_without_pushing_the_schedule_further_out() { 721 721 + let pacer = SubjectPacer::new(RateLimit { 722 722 + burst: Burst::new(1), 723 723 + refill: RefillMicros::new(1_000), 724 724 + }); 725 725 + let nel = SubjectKey::new("did:plc:nel"); 726 726 + assert!( 727 727 + pacer.reserve_now(&nel, at(0)), 728 728 + "a subject the caller hasn't read takes its turn straight away" 729 729 + ); 730 730 + assert!( 731 731 + !pacer.reserve_now(&nel, at(500)), 732 732 + "a caller that mustn't wait is refused inside the interval" 733 733 + ); 734 734 + assert!( 735 735 + !pacer.reserve_now(&nel, at(999)), 736 736 + "refusals must leave the booking alone, or whoever keeps trying pushes the turn \ 737 737 + further out every time and the knot never reads the subject again" 738 738 + ); 739 739 + assert!(pacer.reserve_now(&nel, at(1_000))); 740 740 + assert_eq!( 741 741 + SubjectKey::new("DID:PLC:NEL"), 742 742 + SubjectKey::new("did:plc:nel"), 743 743 + "a DID a client typed in mixed case is the same account and shares its schedule" 744 744 + ); 745 745 + } 746 746 + 747 747 + #[test] 748 748 + fn a_flood_of_distinct_hosts_doesnt_grow_the_schedule_past_its_limit_or_delay_a_newcomer() { 749 749 + let pacer = HostPacer::new(RateLimit { 750 750 + burst: Burst::new(1), 751 751 + refill: RefillMicros::new(1_000_000), 752 752 + }); 753 753 + (0..MAX_PACED_KEYS as u64 + 5_000).for_each(|index| { 754 754 + let _ = pacer.reserve(&HostKey::new(&format!("{index}.nel.pet")), at(0)); 755 755 + }); 756 756 + let tracked = pacer.lock().turns.len(); 757 757 + assert!( 758 758 + tracked <= MAX_PACED_KEYS, 759 759 + "a grant set spread over more did:web hosts than the schedule can track mustn't \ 760 760 + grow it past its limit, saw {tracked}" 761 761 + ); 762 762 + assert_eq!( 763 763 + pacer.reserve(&HostKey::new("plc.directory"), at(0)), 764 764 + Duration::from_micros(1_000_000), 765 765 + "a host the full schedule can't track waits one refill interval, so a caller that \ 766 766 + outgrows the schedule slows itself down" 767 767 + ); 768 768 + let settled = at(SWEEP_INTERVAL_MICROS + 2_000_000); 769 769 + assert_eq!( 770 770 + pacer.reserve(&HostKey::new("plc.directory"), settled), 771 771 + Duration::ZERO 772 772 + ); 773 773 + assert_eq!( 774 774 + pacer.lock().turns.len(), 775 775 + 1, 776 776 + "the sweep reclaims the schedule and tracks the newcomer once every booked turn \ 777 777 + has passed" 559 778 ); 560 779 } 561 780

+36 -35

knot2/crates/knot-resource/src/disk.rs

View file Reviewed

··· 19 19 } 20 20 21 21 #[derive(Debug)] 22 22 - pub enum ReserveError { 23 23 - BelowFloor { 24 24 - free: FreeBytes, 25 25 - floor: DiskFloorBytes, 26 26 - }, 27 27 - Probe(io::Error), 22 22 + pub struct BelowFloor { 23 23 + pub free: FreeBytes, 24 24 + pub floor: DiskFloorBytes, 28 25 } 29 26 30 27 struct Ledger { ··· 47 44 self.0.reserved.load(Ordering::SeqCst) 48 45 } 49 46 50 50 - pub fn reserve( 47 47 + pub fn reserve_against( 51 48 &self, 52 52 - path: &Path, 49 49 + free: FreeBytes, 53 50 bytes: ReserveBytes, 54 54 - ) -> Result<DiskReservation, ReserveError> { 51 51 + ) -> Result<DiskReservation, BelowFloor> { 55 52 let amount = bytes.get(); 56 53 let projected = self.0.reserved.fetch_add(amount, Ordering::SeqCst) + amount; 57 57 - let free = match free_bytes(path) { 58 58 - Ok(free) => free, 59 59 - Err(source) => { 60 60 - self.0.reserved.fetch_sub(amount, Ordering::SeqCst); 61 61 - return Err(ReserveError::Probe(source)); 62 62 - } 63 63 - }; 64 54 if free.get() < self.0.floor.get().saturating_add(projected) { 65 55 self.0.reserved.fetch_sub(amount, Ordering::SeqCst); 66 66 - return Err(ReserveError::BelowFloor { 56 56 + return Err(BelowFloor { 67 57 free, 68 58 floor: self.0.floor, 69 59 }); ··· 103 93 104 94 #[test] 105 95 fn a_reservation_holds_bytes_until_it_drops() { 106 106 - let dir = std::env::temp_dir(); 96 96 + let free = FreeBytes::new(1 << 20); 107 97 let governor = DiskGovernor::new(DiskFloorBytes::new(0)); 108 98 assert_eq!(governor.reserved_bytes(), 0); 109 99 { 110 110 - let _held = governor.reserve(&dir, ReserveBytes::new(4_096)).unwrap(); 100 100 + let _held = governor 101 101 + .reserve_against(free, ReserveBytes::new(4_096)) 102 102 + .unwrap(); 111 103 assert_eq!(governor.reserved_bytes(), 4_096); 112 112 - let _also = governor.reserve(&dir, ReserveBytes::new(1_024)).unwrap(); 104 104 + let _also = governor 105 105 + .reserve_against(free, ReserveBytes::new(1_024)) 106 106 + .unwrap(); 113 107 assert_eq!(governor.reserved_bytes(), 5_120); 114 108 } 115 109 assert_eq!(governor.reserved_bytes(), 0); ··· 117 111 118 112 #[test] 119 113 fn concurrent_reservations_cannot_jointly_punch_through_the_floor() { 120 120 - let dir = std::env::temp_dir(); 121 121 - let free = free_bytes(&dir).unwrap(); 122 122 - let floor = DiskFloorBytes::new(free.get().saturating_sub(6_144)); 123 123 - let governor = DiskGovernor::new(floor); 124 124 - let first = governor.reserve(&dir, ReserveBytes::new(4_096)).unwrap(); 125 125 - let denied = governor.reserve(&dir, ReserveBytes::new(4_096)); 114 114 + let free = FreeBytes::new(10_240); 115 115 + let governor = DiskGovernor::new(DiskFloorBytes::new(4_096)); 116 116 + let first = governor 117 117 + .reserve_against(free, ReserveBytes::new(4_096)) 118 118 + .unwrap(); 119 119 + let denied = governor.reserve_against(free, ReserveBytes::new(4_096)); 126 120 assert!( 127 127 - matches!(denied, Err(ReserveError::BelowFloor { .. })), 121 121 + denied.is_err(), 128 122 "the second reservation must see the first still held" 129 123 ); 130 124 assert_eq!(governor.reserved_bytes(), 4_096); 131 125 drop(first); 132 126 assert_eq!(governor.reserved_bytes(), 0); 133 133 - assert!(governor.reserve(&dir, ReserveBytes::new(4_096)).is_ok()); 127 127 + assert!( 128 128 + governor 129 129 + .reserve_against(free, ReserveBytes::new(4_096)) 130 130 + .is_ok(), 131 131 + "a free-space reading the floor leaves room in must admit a lone reservation, or \ 132 132 + the refusal above was the floor rather than the reservation still being held" 133 133 + ); 134 134 } 135 135 136 136 #[test] 137 137 - fn a_probe_fault_leaves_the_ledger_untouched() { 138 138 - let governor = DiskGovernor::new(DiskFloorBytes::new(0)); 139 139 - let fault = governor.reserve( 140 140 - Path::new("/definitely/not/a/mounted/path"), 141 141 - ReserveBytes::new(4_096), 137 137 + fn a_refused_reservation_leaves_the_ledger_untouched() { 138 138 + let governor = DiskGovernor::new(DiskFloorBytes::new(4_096)); 139 139 + let refused = governor.reserve_against(FreeBytes::new(4_096), ReserveBytes::new(1)); 140 140 + assert!(refused.is_err()); 141 141 + assert_eq!( 142 142 + governor.reserved_bytes(), 143 143 + 0, 144 144 + "a refusal that left its bytes on the ledger would deny every later upload too" 142 145 ); 143 143 - assert!(matches!(fault, Err(ReserveError::Probe(_)))); 144 144 - assert_eq!(governor.reserved_bytes(), 0); 145 146 } 146 147 }

+3 -3

knot2/crates/knot-resource/src/lib.rs

View file Reviewed

··· 6 6 mod slots; 7 7 8 8 pub use admission::{ 9 9 - AdmitGuard, Burst, GlobalInflight, LimitConfig, PerPeerInflight, PreAuthLimiter, RateLimit, 10 10 - RefillMicros, Refusal, 9 9 + AdmitGuard, Burst, GlobalInflight, HostKey, HostPacer, LimitConfig, PeerPacer, PerPeerInflight, 10 10 + PreAuthLimiter, RateLimit, RefillMicros, Refusal, SubjectKey, SubjectPacer, 11 11 }; 12 12 pub use cpu::{Saturate, ThreadCount, gix_thread_limit, map_chunks, map_spans, saturate, threads}; 13 13 pub use disk::{ 14 14 - DiskFloorBytes, DiskGovernor, DiskReservation, FreeBytes, ReserveBytes, ReserveError, 14 14 + BelowFloor, DiskFloorBytes, DiskGovernor, DiskReservation, FreeBytes, ReserveBytes, 15 15 free_bytes as disk_free_bytes, 16 16 }; 17 17 pub use fsio::{