1use std::collections::HashMap;
9use std::fmt;
10use std::sync::atomic::{AtomicU64, Ordering};
11use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
12
13use crate::runtime::fs::FileIdentity;
14
15#[derive(Debug)]
17pub struct DiskQuota {
18 limit: u64,
19 used: AtomicU64,
20 unmeasured: bool,
22 open: Mutex<HashMap<FileIdentity, OpenFile>>,
24}
25
26#[derive(Debug, Default)]
27struct OpenFile {
28 readers: u32,
29 writer: bool,
30 pending: u64,
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum Holder {
38 Reader,
39 Writer,
40}
41
42impl DiskQuota {
43 pub fn new(limit: u64, used: u64) -> Self {
44 Self {
45 limit,
46 used: AtomicU64::new(used),
47 unmeasured: false,
48 open: Mutex::default(),
49 }
50 }
51
52 pub fn unmeasured(limit: u64) -> Self {
55 Self {
56 unmeasured: true,
57 ..Self::new(limit, 0)
58 }
59 }
60
61 pub fn limit(&self) -> u64 {
62 self.limit
63 }
64
65 pub fn used(&self) -> u64 {
66 self.used.load(Ordering::Acquire)
67 }
68
69 pub fn is_unmeasured(&self) -> bool {
70 self.unmeasured
71 }
72
73 pub fn reserve(&self, bytes: u64) -> Result<(), QuotaExceeded> {
77 if bytes == 0 {
78 return Ok(());
79 }
80 if self.unmeasured {
81 return Err(QuotaExceeded::Unmeasured { limit: self.limit });
82 }
83 let mut used = self.used.load(Ordering::Acquire);
84 loop {
85 let Some(next) = used.checked_add(bytes).filter(|next| *next <= self.limit) else {
86 return Err(QuotaExceeded::Over {
87 limit: self.limit,
88 used,
89 requested: bytes,
90 });
91 };
92 match self
93 .used
94 .compare_exchange_weak(used, next, Ordering::AcqRel, Ordering::Acquire)
95 {
96 Ok(_) => return Ok(()),
97 Err(actual) => used = actual,
98 }
99 }
100 }
101
102 pub fn record(&self, bytes: u64) {
104 let _ = self
105 .used
106 .fetch_update(Ordering::AcqRel, Ordering::Acquire, |used| {
107 Some(used.saturating_add(bytes))
108 });
109 }
110
111 pub fn release(&self, bytes: u64) {
112 let _ = self
113 .used
114 .fetch_update(Ordering::AcqRel, Ordering::Acquire, |used| {
115 Some(used.saturating_sub(bytes))
116 });
117 }
118
119 pub fn hold(self: &Arc<Self>, file: FileIdentity, holder: Holder) -> OpenFileGuard {
121 let mut open = self.open_files();
122 let entry = open.entry(file).or_default();
123 match holder {
124 Holder::Reader => entry.readers = entry.readers.saturating_add(1),
125 Holder::Writer => entry.writer = true,
126 }
127 OpenFileGuard {
128 quota: Arc::clone(self),
129 file,
130 holder,
131 }
132 }
133
134 pub fn is_held(&self, file: FileIdentity) -> bool {
136 self.open_files().contains_key(&file)
137 }
138
139 fn is_writers_own(&self, file: FileIdentity) -> bool {
141 self.open_files().get(&file).is_some_and(|held| held.writer)
142 }
143
144 pub fn release_file(&self, file: FileIdentity, bytes: u64) {
148 let mut open = self.open_files();
149 match open.get_mut(&file) {
150 Some(held) if held.writer => {}
151 Some(held) => held.pending = held.pending.saturating_add(bytes),
152 None => {
153 drop(open);
154 self.release(bytes);
155 }
156 }
157 }
158
159 fn release_writers_own_file(&self, file: FileIdentity, bytes: u64) {
163 let mut open = self.open_files();
164 match open.get_mut(&file) {
165 Some(held) if held.readers > 0 => held.pending = held.pending.saturating_add(bytes),
166 _ => {
167 drop(open);
168 self.release(bytes);
169 }
170 }
171 }
172
173 fn open_files(&self) -> MutexGuard<'_, HashMap<FileIdentity, OpenFile>> {
174 self.open.lock().unwrap_or_else(PoisonError::into_inner)
177 }
178}
179
180#[derive(Debug)]
182pub struct OpenFileGuard {
183 quota: Arc<DiskQuota>,
184 file: FileIdentity,
185 holder: Holder,
186}
187
188impl Drop for OpenFileGuard {
189 fn drop(&mut self) {
190 let mut open = self.quota.open_files();
191 let Some(held) = open.get_mut(&self.file) else {
192 return;
193 };
194 match self.holder {
195 Holder::Reader => held.readers = held.readers.saturating_sub(1),
196 Holder::Writer => held.writer = false,
197 }
198 if held.readers == 0 && !held.writer {
199 let pending = held.pending;
200 open.remove(&self.file);
201 drop(open);
202 self.quota.release(pending);
203 }
204 }
205}
206
207#[derive(Debug)]
223pub struct QuotaCharge {
224 quota: Option<Arc<DiskQuota>>,
225 allowance: u64,
226 drawn: u64,
228 reserved: u64,
229 replaces: Replaces,
230}
231
232#[derive(Debug, Clone, Copy)]
234enum Replaces {
235 Nothing,
238 File(FileIdentity, u64),
241 InPlace,
243}
244
245impl Replaces {
246 fn file(replaced: Option<(FileIdentity, u64)>) -> Self {
247 replaced.map_or(Self::Nothing, |(file, bytes)| Self::File(file, bytes))
248 }
249}
250
251impl QuotaCharge {
252 pub fn new(quota: Option<Arc<DiskQuota>>, replaced: Option<(FileIdentity, u64)>) -> Self {
254 let allowance = match ("a, replaced) {
255 (Some(quota), Some((file, bytes))) if !quota.is_held(file) => bytes,
256 _ => 0,
257 };
258 Self {
259 quota,
260 allowance,
261 drawn: 0,
262 reserved: 0,
263 replaces: Replaces::file(replaced),
264 }
265 }
266
267 pub fn in_place(quota: Option<Arc<DiskQuota>>, rewritten: Option<(FileIdentity, u64)>) -> Self {
272 let (allowance, replaces) = match ("a, rewritten) {
273 (Some(quota), Some((file, bytes))) if !quota.is_writers_own(file) => {
274 (bytes, Replaces::InPlace)
275 }
276 _ => (0, Replaces::Nothing),
277 };
278 Self {
279 quota,
280 allowance,
281 drawn: 0,
282 reserved: 0,
283 replaces,
284 }
285 }
286
287 pub fn reserve(&mut self, bytes: u64) -> Result<(), QuotaExceeded> {
290 let from_allowance = bytes.min(self.allowance.saturating_sub(self.drawn));
291 let beyond = bytes.saturating_sub(from_allowance);
292 if let Some(quota) = &self.quota {
293 quota.reserve(beyond)?;
294 }
295 self.drawn = self.drawn.saturating_add(from_allowance);
296 self.reserved = self.reserved.saturating_add(beyond);
297 Ok(())
298 }
299
300 pub fn unreserve(&mut self, bytes: u64) {
303 let from_reserved = bytes.min(self.reserved);
304 if let Some(quota) = &self.quota {
305 quota.release(from_reserved);
306 }
307 self.reserved = self.reserved.saturating_sub(from_reserved);
308 self.drawn = self
309 .drawn
310 .saturating_sub(bytes.saturating_sub(from_reserved));
311 }
312
313 pub fn cover(&mut self, replaced: Option<(FileIdentity, u64)>) -> Result<(), QuotaExceeded> {
317 let replaced_bytes = replaced.map_or(0, |(_, bytes)| bytes);
318 let uncovered = self.drawn.saturating_sub(replaced_bytes);
319 if let Some(quota) = &self.quota {
320 quota.reserve(uncovered)?;
321 }
322 self.drawn = self.drawn.saturating_sub(uncovered);
323 self.reserved = self.reserved.saturating_add(uncovered);
324 self.replaces = Replaces::file(replaced);
325 Ok(())
326 }
327
328 pub fn commit(mut self) {
332 let Some(quota) = self.quota.take() else {
333 return;
334 };
335 match self.replaces {
336 Replaces::Nothing => {}
337 Replaces::File(file, bytes) => {
338 quota.release_file(file, bytes.saturating_sub(self.drawn));
339 }
340 Replaces::InPlace => quota.release(self.allowance.saturating_sub(self.drawn)),
341 }
342 }
343
344 pub fn settle_at(mut self, left: u64) {
348 let Some(quota) = self.quota.take() else {
349 return;
350 };
351 match self.replaces {
352 Replaces::InPlace => {
353 let counted = self.allowance.saturating_add(self.reserved);
354 if left < counted {
355 quota.release(counted - left);
356 } else {
357 quota.record(left - counted);
358 }
359 }
360 Replaces::Nothing | Replaces::File(..) => {
361 quota.release(self.reserved.saturating_sub(left));
362 }
363 }
364 }
365
366 pub fn keep(mut self) {
369 self.quota = None;
370 }
371
372 pub fn release_when_closed(mut self, written: FileIdentity) {
375 if let Some(quota) = self.quota.take() {
376 quota.release_writers_own_file(written, self.reserved);
377 }
378 }
379}
380
381impl Drop for QuotaCharge {
382 fn drop(&mut self) {
383 if let Some(quota) = self.quota.take() {
384 quota.release(self.reserved);
385 }
386 }
387}
388
389#[derive(Debug, Clone, Copy, PartialEq, Eq)]
392pub enum QuotaExceeded {
393 Over {
395 limit: u64,
396 used: u64,
397 requested: u64,
398 },
399 Unmeasured { limit: u64 },
402}
403
404impl fmt::Display for QuotaExceeded {
405 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
406 match self {
407 Self::Over {
408 limit,
409 used,
410 requested,
411 } => write!(
412 f,
413 "the filesystem's size limit of {limit} bytes would be exceeded: {used} bytes are in use and this needs {requested} more"
414 ),
415 Self::Unmeasured { limit } => write!(
416 f,
417 "the filesystem couldn't be measured against its size limit of {limit} bytes, so it is treated as full until a run can measure it"
418 ),
419 }
420 }
421}
422
423impl std::error::Error for QuotaExceeded {}
424
425#[cfg(test)]
426mod tests {
427 use super::*;
428
429 #[test]
430 fn quota_reserves_within_the_limit_and_refuses_past_it() {
431 let quota = DiskQuota::new(100, 40);
432 quota.reserve(60).expect("fits exactly");
433 let refused = quota.reserve(1).expect_err("one byte over");
434 assert_eq!(
435 refused,
436 QuotaExceeded::Over {
437 limit: 100,
438 used: 100,
439 requested: 1,
440 }
441 );
442 quota.release(30);
443 assert_eq!(quota.used(), 70);
444 quota.release(1_000);
445 assert_eq!(quota.used(), 0, "release never underflows");
446 }
447
448 #[test]
449 fn a_dropped_charge_gives_back_what_it_reserved() {
450 let quota = Arc::new(DiskQuota::new(100, 0));
451 let mut charge = QuotaCharge::new(Some(Arc::clone("a)), None);
452 charge.reserve(70).expect("fits");
453 assert_eq!(quota.used(), 70);
454 drop(charge);
455 assert_eq!(quota.used(), 0);
456 }
457
458 #[test]
459 fn a_charge_covers_a_replaced_file_that_shrank() {
460 let quota = Arc::new(DiskQuota::new(100, 90));
461 let file = identity();
462 let mut charge = QuotaCharge::new(Some(Arc::clone("a)), Some((file, 90)));
463 charge.reserve(90).expect("drawn from the allowance");
464 assert_eq!(quota.used(), 90);
465 quota.release(90);
468 charge
469 .cover(Some((file, 0)))
470 .expect("fits once the file is empty");
471 charge.commit();
472 assert_eq!(quota.used(), 90);
473 let mut again = QuotaCharge::new(Some(Arc::clone("a)), Some((file, 90)));
474 again.reserve(90).expect("drawn from the allowance");
475 assert!(
476 again.cover(Some((file, 0))).is_err(),
477 "the file shrank and nothing freed its bytes"
478 );
479 }
480
481 #[test]
482 fn a_held_file_gives_no_allowance() {
483 let quota = Arc::new(DiskQuota::new(100, 60));
484 let file = identity();
485 let reader = quota.hold(file, Holder::Reader);
486 let mut charge = QuotaCharge::new(Some(Arc::clone("a)), Some((file, 60)));
487 assert!(
488 charge.reserve(60).is_err(),
489 "the replaced file stays on disk while the reader holds it"
490 );
491 charge.reserve(40).expect("the growth beyond it fits");
492 charge.commit();
493 assert_eq!(quota.used(), 100, "the old file is still counted");
494 drop(reader);
495 assert_eq!(quota.used(), 40, "and freed when the reader closes");
496 }
497
498 #[test]
499 fn a_file_rewritten_in_place_frees_its_old_bytes_while_held() {
500 let quota = Arc::new(DiskQuota::new(100, 60));
501 let file = identity();
502 let _reader = quota.hold(file, Holder::Reader);
503 let mut charge = QuotaCharge::in_place(Some(Arc::clone("a)), Some((file, 60)));
504 charge.reserve(30).expect("drawn from the old contents");
505 charge.commit();
506 assert_eq!(quota.used(), 30, "the old contents are gone, held or not");
507 }
508
509 #[test]
510 fn a_failed_rewrite_in_place_counts_what_it_left() {
511 let quota = Arc::new(DiskQuota::new(100, 40));
512 let mut charge = QuotaCharge::in_place(Some(Arc::clone("a)), Some((identity(), 40)));
513 charge.reserve(50).expect("room for the new contents");
514 assert_eq!(quota.used(), 50);
515 charge.settle_at(20);
516 assert_eq!(
517 quota.used(),
518 20,
519 "the old contents are gone, 20 new bytes landed"
520 );
521
522 let untouched = Arc::new(DiskQuota::new(100, 40));
523 let mut charge =
524 QuotaCharge::in_place(Some(Arc::clone(&untouched)), Some((identity(), 40)));
525 charge.reserve(10).expect("drawn from the old contents");
526 charge.settle_at(40);
527 assert_eq!(
528 untouched.used(),
529 40,
530 "a copy that failed before truncating left the file"
531 );
532 }
533
534 #[test]
535 fn a_writers_own_file_rewritten_in_place_is_left_for_the_writer() {
536 let quota = Arc::new(DiskQuota::new(100, 60));
537 let file = identity();
538 let _writer = quota.hold(file, Holder::Writer);
539 let mut charge = QuotaCharge::in_place(Some(Arc::clone("a)), Some((file, 60)));
540 charge.reserve(30).expect("room for the new contents");
541 charge.commit();
542 assert_eq!(
543 quota.used(),
544 90,
545 "the writer's charge still holds the old contents"
546 );
547 }
548
549 #[test]
550 fn a_removed_file_held_open_stays_counted_until_it_closes() {
551 let quota = Arc::new(DiskQuota::new(100, 60));
552 let file = identity();
553 let reader = quota.hold(file, Holder::Reader);
554 quota.release_file(file, 60);
555 assert_eq!(quota.used(), 60, "an open file's bytes are still on disk");
556 drop(reader);
557 assert_eq!(quota.used(), 0);
558 }
559
560 #[test]
561 fn a_writers_own_file_is_left_for_the_writer_to_settle() {
562 let quota = Arc::new(DiskQuota::new(100, 0));
563 let file = identity();
564 let writer = quota.hold(file, Holder::Writer);
565 quota.record(40);
566 quota.release_file(file, 40);
567 assert_eq!(quota.used(), 40);
568 drop(writer);
569 assert_eq!(
570 quota.used(),
571 40,
572 "releasing it is the writer's charge's job"
573 );
574 }
575
576 fn identity() -> FileIdentity {
578 let file = tempfile::tempfile().expect("tempfile");
579 let metadata = cap_std::fs::Metadata::from_file(&file).expect("metadata");
580 FileIdentity::of(&metadata).expect("identity")
581 }
582}