alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::collections::{HashMap, HashSet};
2use std::io::{Read, Write};
3use std::path::{Path, PathBuf};
4use std::sync::Mutex;
5use std::time::{Duration, SystemTime};
6
7use knot_types::RepoDid;
8use sha2::{Digest, Sha256};
9
10use crate::types::RepoPrefix;
11use crate::{ClaimedSize, FreeSpaceFloor, LfsError, LfsOid, LfsSize, LfsStorePath, ObjectRelPath};
12
13pub trait LfsStore: Send + Sync {
14 fn put(
15 &self,
16 repo: &RepoDid,
17 oid: &LfsOid,
18 size: ClaimedSize,
19 body: &mut dyn Read,
20 ) -> Result<(), LfsError>;
21
22 fn read(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Box<dyn Read + Send>, LfsError>;
23
24 fn probe(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Option<LfsSize>, LfsError>;
25
26 fn touch(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Option<LfsSize>, LfsError>;
27}
28
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct StoredObject {
31 pub oid: LfsOid,
32 pub size: LfsSize,
33 pub mtime: SystemTime,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum Reclaimed {
38 Swept(LfsSize),
39 Spared,
40}
41
42#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
43pub struct OrphanSweep {
44 pub prefixes: usize,
45 pub objects: usize,
46 pub bytes: LfsSize,
47}
48
49pub(crate) fn expired(now: SystemTime, mtime: SystemTime, grace: Duration) -> bool {
50 now.duration_since(mtime)
51 .map(|age| age >= grace)
52 .unwrap_or(false)
53}
54
55const COPY_CHUNK: usize = 64 * 1024;
56
57fn io_at(op: &'static str, path: &Path) -> impl FnOnce(std::io::Error) -> LfsError {
58 let path = path.to_path_buf();
59 move |source| LfsError::Io { op, path, source }
60}
61
62pub(crate) fn for_each_chunk(
63 body: &mut dyn Read,
64 mut step: impl FnMut(&[u8]) -> Result<(), LfsError>,
65) -> Result<(), LfsError> {
66 let mut buffer = vec![0u8; COPY_CHUNK];
67 std::iter::from_fn(|| match body.read(&mut buffer) {
68 Ok(0) => None,
69 Ok(count) => Some(step(&buffer[..count])),
70 Err(source) if source.kind() == std::io::ErrorKind::Interrupted => Some(Ok(())),
71 Err(source) => Some(Err(LfsError::BodyRead { source })),
72 })
73 .try_for_each(std::convert::identity)
74}
75
76fn write_verified(
77 declared: &LfsOid,
78 size: ClaimedSize,
79 body: &mut dyn Read,
80 mut sink: impl FnMut(&[u8]) -> Result<(), LfsError>,
81) -> Result<(), LfsError> {
82 let mut hasher = Sha256::new();
83 let mut received: u64 = 0;
84 for_each_chunk(body, |chunk| {
85 received += chunk.len() as u64;
86 if received > size.get() {
87 return Err(LfsError::SizeMismatch {
88 declared: size,
89 received: LfsSize::new(received),
90 });
91 }
92 hasher.update(chunk);
93 sink(chunk)
94 })?;
95 if received != size.get() {
96 return Err(LfsError::SizeMismatch {
97 declared: size,
98 received: LfsSize::new(received),
99 });
100 }
101 let computed = LfsOid::from_digest(hasher.finalize().into());
102 match computed == *declared {
103 true => Ok(()),
104 false => Err(LfsError::HashMismatch {
105 declared: declared.clone(),
106 computed,
107 }),
108 }
109}
110
111fn fsync_dir(path: &Path) -> Result<(), LfsError> {
112 std::fs::File::open(path)
113 .and_then(|dir| dir.sync_all())
114 .map_err(io_at("sync dir", path))
115}
116
117fn fsync_chain(root: &Path, leaf: &Path) -> Result<(), LfsError> {
118 leaf.ancestors()
119 .take_while(|dir| dir.starts_with(root))
120 .try_for_each(fsync_dir)
121}
122
123const INCOMING_DIR: &str = ".incoming";
124const OID_LOCK_STRIPES: usize = 64;
125
126fn set_mtime_now(path: &Path) -> Result<(), LfsError> {
127 std::fs::OpenOptions::new()
128 .write(true)
129 .open(path)
130 .and_then(|file| file.set_modified(SystemTime::now()))
131 .map_err(io_at("touch mtime", path))
132}
133
134fn subdirs(path: &Path) -> Result<Vec<PathBuf>, LfsError> {
135 match std::fs::read_dir(path) {
136 Ok(entries) => entries
137 .map(|entry| {
138 entry
139 .map(|entry| entry.path())
140 .map_err(io_at("read dir", path))
141 })
142 .filter(|entry| entry.as_ref().map(|path| path.is_dir()).unwrap_or(true))
143 .collect(),
144 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
145 Err(source) => Err(io_at("read dir", path)(source)),
146 }
147}
148
149fn stored_object(path: &Path) -> Result<Option<StoredObject>, LfsError> {
150 let Some(oid) = path
151 .file_name()
152 .and_then(|name| name.to_str())
153 .and_then(|name| LfsOid::new(name).ok())
154 else {
155 return Ok(None);
156 };
157 let meta = std::fs::metadata(path).map_err(io_at("stat", path))?;
158 let mtime = meta.modified().map_err(io_at("read mtime", path))?;
159 Ok(Some(StoredObject {
160 oid,
161 size: LfsSize::new(meta.len()),
162 mtime,
163 }))
164}
165
166fn is_object_file(path: &Path) -> bool {
167 path.is_file()
168 && path
169 .file_name()
170 .and_then(|name| name.to_str())
171 .is_some_and(|name| LfsOid::new(name).is_ok())
172}
173
174fn is_shard_nibble(path: &Path) -> bool {
175 path.file_name()
176 .and_then(|name| name.to_str())
177 .is_some_and(|name| {
178 name.len() == 2
179 && name
180 .bytes()
181 .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
182 })
183}
184
185fn is_object_leaf(dir: &Path) -> bool {
186 is_shard_nibble(dir) && dir.parent().is_some_and(is_shard_nibble)
187}
188
189fn discover_prefixes(dir: &Path) -> Result<Vec<PathBuf>, LfsError> {
190 let entries = match std::fs::read_dir(dir) {
191 Ok(entries) => entries
192 .map(|entry| {
193 entry
194 .map(|entry| entry.path())
195 .map_err(io_at("read dir", dir))
196 })
197 .collect::<Result<Vec<_>, _>>()?,
198 Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
199 Err(source) => return Err(io_at("read dir", dir)(source)),
200 };
201 if is_object_leaf(dir) && entries.iter().any(|path| is_object_file(path)) {
202 return Ok(dir
203 .parent()
204 .and_then(|shard| shard.parent())
205 .map(Path::to_path_buf)
206 .into_iter()
207 .collect());
208 }
209 entries
210 .iter()
211 .filter(|path| path.is_dir())
212 .map(|sub| discover_prefixes(sub))
213 .collect::<Result<Vec<_>, _>>()
214 .map(|nested| nested.into_iter().flatten().collect())
215}
216
217fn enumerate_prefix(prefix: &Path) -> Result<Vec<StoredObject>, LfsError> {
218 subdirs(prefix)?
219 .iter()
220 .map(|shard| subdirs(shard))
221 .collect::<Result<Vec<_>, _>>()?
222 .into_iter()
223 .flatten()
224 .map(|nibble| match std::fs::read_dir(&nibble) {
225 Ok(entries) => entries
226 .map(|entry| {
227 entry
228 .map(|entry| entry.path())
229 .map_err(io_at("read dir", &nibble))
230 })
231 .collect::<Result<Vec<_>, _>>(),
232 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
233 Err(source) => Err(io_at("read dir", &nibble)(source)),
234 })
235 .collect::<Result<Vec<_>, _>>()?
236 .into_iter()
237 .flatten()
238 .filter_map(|file| stored_object(&file).transpose())
239 .collect()
240}
241
242pub struct DiskStore {
243 root: LfsStorePath,
244 locks: Box<[Mutex<()>]>,
245}
246
247impl DiskStore {
248 pub fn open(root: LfsStorePath) -> Result<Self, LfsError> {
249 let incoming = root.as_path().join(INCOMING_DIR);
250 std::fs::create_dir_all(&incoming).map_err(io_at("create dir", &incoming))?;
251 std::fs::read_dir(&incoming)
252 .map_err(io_at("read dir", &incoming))?
253 .try_for_each(|entry| {
254 let path = entry.map_err(io_at("read dir", &incoming))?.path();
255 match std::fs::remove_file(&path) {
256 Ok(()) => Ok(()),
257 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
258 Err(source) => Err(io_at("remove abandoned upload", &path)(source)),
259 }
260 })?;
261 let locks = std::iter::repeat_with(|| Mutex::new(()))
262 .take(OID_LOCK_STRIPES)
263 .collect();
264 Ok(Self { root, locks })
265 }
266
267 fn object_path(&self, repo: &RepoDid, oid: &LfsOid) -> Result<PathBuf, LfsError> {
268 Ok(self.root.object_path(&ObjectRelPath::new(repo, oid)?))
269 }
270
271 fn oid_lock(&self, oid: &LfsOid) -> &Mutex<()> {
272 let stripe = u8::from_str_radix(&oid.as_str()[0..2], 16).unwrap_or(0) as usize;
273 &self.locks[stripe % OID_LOCK_STRIPES]
274 }
275}
276
277impl LfsStore for DiskStore {
278 fn put(
279 &self,
280 repo: &RepoDid,
281 oid: &LfsOid,
282 size: ClaimedSize,
283 body: &mut dyn Read,
284 ) -> Result<(), LfsError> {
285 let target = self.object_path(repo, oid)?;
286 let incoming = self.root.as_path().join(INCOMING_DIR);
287 let mut temp = tempfile::Builder::new()
288 .prefix("put-")
289 .tempfile_in(&incoming)
290 .map_err(io_at("create temp under", &incoming))?;
291 let temp_path = temp.path().to_path_buf();
292 write_verified(oid, size, body, |chunk| {
293 temp.as_file_mut()
294 .write_all(chunk)
295 .map_err(io_at("write", &temp_path))
296 })?;
297 temp.as_file()
298 .sync_all()
299 .map_err(io_at("sync", &temp_path))?;
300 let parent = target
301 .parent()
302 .expect("object path always has a shard parent");
303 let _guard = self
304 .oid_lock(oid)
305 .lock()
306 .unwrap_or_else(std::sync::PoisonError::into_inner);
307 std::fs::create_dir_all(parent).map_err(io_at("create dir", parent))?;
308 temp.persist(&target).map_err(|fault| LfsError::Io {
309 op: "rename into",
310 path: target.clone(),
311 source: fault.error,
312 })?;
313 fsync_chain(self.root.as_path(), parent)
314 }
315
316 fn read(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Box<dyn Read + Send>, LfsError> {
317 let path = self.object_path(repo, oid)?;
318 match std::fs::File::open(&path) {
319 Ok(file) => Ok(Box::new(file)),
320 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
321 Err(LfsError::NotFound { oid: oid.clone() })
322 }
323 Err(source) => Err(io_at("open", &path)(source)),
324 }
325 }
326
327 fn probe(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Option<LfsSize>, LfsError> {
328 Ok(self.object_file(repo, oid)?.map(|(size, _)| size))
329 }
330
331 fn touch(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Option<LfsSize>, LfsError> {
332 let path = self.object_path(repo, oid)?;
333 let _guard = self
334 .oid_lock(oid)
335 .lock()
336 .unwrap_or_else(std::sync::PoisonError::into_inner);
337 match std::fs::metadata(&path) {
338 Ok(meta) => {
339 set_mtime_now(&path)?;
340 Ok(Some(LfsSize::new(meta.len())))
341 }
342 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(None),
343 Err(source) => Err(io_at("stat", &path)(source)),
344 }
345 }
346}
347
348impl DiskStore {
349 pub fn object_file(
350 &self,
351 repo: &RepoDid,
352 oid: &LfsOid,
353 ) -> Result<Option<(LfsSize, PathBuf)>, LfsError> {
354 let path = self.object_path(repo, oid)?;
355 match std::fs::metadata(&path) {
356 Ok(meta) => Ok(Some((LfsSize::new(meta.len()), path))),
357 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(None),
358 Err(source) => Err(io_at("stat", &path)(source)),
359 }
360 }
361
362 pub fn probe_ready(&self) -> Result<(), LfsError> {
363 let incoming = self.root.as_path().join(INCOMING_DIR);
364 tempfile::tempfile_in(&incoming)
365 .map(|_| ())
366 .map_err(io_at("probe writability under", &incoming))
367 }
368
369 pub fn remove_repo(&self, repo: &RepoDid) -> Result<(), LfsError> {
370 let prefix = self.root.as_path().join(RepoPrefix::new(repo)?.as_path());
371 match std::fs::remove_dir_all(&prefix) {
372 Ok(()) => Ok(()),
373 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
374 Err(source) => Err(io_at("remove repo prefix", &prefix)(source)),
375 }
376 }
377
378 pub fn enumerate(&self, repo: &RepoDid) -> Result<Vec<StoredObject>, LfsError> {
379 let prefix = self.root.as_path().join(RepoPrefix::new(repo)?.as_path());
380 enumerate_prefix(&prefix)
381 }
382
383 pub fn collect_expired(
384 &self,
385 repo: &RepoDid,
386 oid: &LfsOid,
387 grace: Duration,
388 now: SystemTime,
389 ) -> Result<Reclaimed, LfsError> {
390 let path = self.object_path(repo, oid)?;
391 let _guard = self
392 .oid_lock(oid)
393 .lock()
394 .unwrap_or_else(std::sync::PoisonError::into_inner);
395 let meta = match std::fs::metadata(&path) {
396 Ok(meta) => meta,
397 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
398 return Ok(Reclaimed::Spared);
399 }
400 Err(source) => return Err(io_at("stat", &path)(source)),
401 };
402 let mtime = meta.modified().map_err(io_at("read mtime", &path))?;
403 if !expired(now, mtime, grace) {
404 return Ok(Reclaimed::Spared);
405 }
406 match std::fs::remove_file(&path) {
407 Ok(()) => Ok(Reclaimed::Swept(LfsSize::new(meta.len()))),
408 Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(Reclaimed::Spared),
409 Err(source) => Err(io_at("remove object", &path)(source)),
410 }
411 }
412
413 pub fn sweep_orphans(
414 &self,
415 hosted: &HashSet<RepoDid>,
416 grace: Duration,
417 now: SystemTime,
418 ) -> Result<OrphanSweep, LfsError> {
419 let root = self.root.as_path();
420 let expected: HashSet<PathBuf> = hosted
421 .iter()
422 .filter_map(|repo| RepoPrefix::new(repo).ok())
423 .map(|prefix| prefix.as_path().to_path_buf())
424 .collect();
425 let orphans: HashSet<PathBuf> = discover_prefixes(root)?
426 .into_iter()
427 .filter(|prefix| {
428 prefix
429 .strip_prefix(root)
430 .ok()
431 .filter(|rel| matches!(rel.components().count(), 2 | 3))
432 .map(|rel| !expected.contains(rel))
433 .unwrap_or(false)
434 })
435 .collect();
436 orphans
437 .iter()
438 .map(|prefix| self.reclaim_orphan(prefix, grace, now))
439 .try_fold(OrphanSweep::default(), |acc, outcome| {
440 let outcome = outcome?;
441 Ok(OrphanSweep {
442 prefixes: acc.prefixes + outcome.prefixes,
443 objects: acc.objects + outcome.objects,
444 bytes: acc.bytes.saturating_add(outcome.bytes),
445 })
446 })
447 }
448
449 fn reclaim_orphan(
450 &self,
451 prefix: &Path,
452 grace: Duration,
453 now: SystemTime,
454 ) -> Result<OrphanSweep, LfsError> {
455 let objects = enumerate_prefix(prefix)?;
456 let live = objects
457 .iter()
458 .any(|object| !expired(now, object.mtime, grace));
459 if live {
460 return Ok(OrphanSweep::default());
461 }
462 match std::fs::remove_dir_all(prefix) {
463 Ok(()) => Ok(OrphanSweep {
464 prefixes: 1,
465 objects: objects.len(),
466 bytes: objects
467 .iter()
468 .map(|object| object.size)
469 .fold(LfsSize::new(0), LfsSize::saturating_add),
470 }),
471 Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
472 Ok(OrphanSweep::default())
473 }
474 Err(source) => Err(io_at("remove orphan prefix", prefix)(source)),
475 }
476 }
477}
478
479#[derive(Clone)]
480pub struct LfsHandle {
481 pub store: std::sync::Arc<DiskStore>,
482 pub admission: std::sync::Arc<crate::StoreAdmission>,
483}
484
485impl LfsHandle {
486 pub fn open(
487 root: LfsStorePath,
488 max_object: LfsSize,
489 free_space_floor: FreeSpaceFloor,
490 ) -> Result<Self, LfsError> {
491 let admission = crate::StoreAdmission::new(root.clone(), max_object, free_space_floor);
492 Ok(Self {
493 store: std::sync::Arc::new(DiskStore::open(root)?),
494 admission: std::sync::Arc::new(admission),
495 })
496 }
497}
498
499#[derive(Default)]
500pub struct MemoryStore {
501 objects: Mutex<HashMap<ObjectRelPath, Vec<u8>>>,
502}
503
504impl MemoryStore {
505 pub fn new() -> Self {
506 Self::default()
507 }
508
509 fn locked(&self) -> std::sync::MutexGuard<'_, HashMap<ObjectRelPath, Vec<u8>>> {
510 self.objects.lock().expect("lfs memory store lock poisoned")
511 }
512}
513
514impl LfsStore for MemoryStore {
515 fn put(
516 &self,
517 repo: &RepoDid,
518 oid: &LfsOid,
519 size: ClaimedSize,
520 body: &mut dyn Read,
521 ) -> Result<(), LfsError> {
522 let rel = ObjectRelPath::new(repo, oid)?;
523 let mut bytes = Vec::new();
524 write_verified(oid, size, body, |chunk| {
525 bytes.extend_from_slice(chunk);
526 Ok(())
527 })?;
528 self.locked().insert(rel, bytes);
529 Ok(())
530 }
531
532 fn read(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Box<dyn Read + Send>, LfsError> {
533 let rel = ObjectRelPath::new(repo, oid)?;
534 self.locked()
535 .get(&rel)
536 .cloned()
537 .map(|bytes| Box::new(std::io::Cursor::new(bytes)) as Box<dyn Read + Send>)
538 .ok_or_else(|| LfsError::NotFound { oid: oid.clone() })
539 }
540
541 fn probe(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Option<LfsSize>, LfsError> {
542 let rel = ObjectRelPath::new(repo, oid)?;
543 Ok(self
544 .locked()
545 .get(&rel)
546 .map(|bytes| LfsSize::new(bytes.len() as u64)))
547 }
548
549 fn touch(&self, repo: &RepoDid, oid: &LfsOid) -> Result<Option<LfsSize>, LfsError> {
550 self.probe(repo, oid)
551 }
552}
553
554#[cfg(test)]
555mod tests {
556 use super::*;
557
558 const MONTH: Duration = Duration::from_secs(30 * 86_400);
559 const GRACE: Duration = Duration::from_secs(14 * 86_400);
560
561 fn oid_of(bytes: &[u8]) -> LfsOid {
562 LfsOid::from_digest(Sha256::digest(bytes).into())
563 }
564
565 fn disk() -> (DiskStore, tempfile::TempDir) {
566 let dir = tempfile::tempdir().unwrap();
567 let store = DiskStore::open(LfsStorePath::new(dir.path())).unwrap();
568 (store, dir)
569 }
570
571 fn seed(store: &DiskStore, repo: &RepoDid, body: &[u8]) -> (LfsOid, LfsSize) {
572 let oid = oid_of(body);
573 let bytes = body.len() as u64;
574 store
575 .put(repo, &oid, ClaimedSize::new(bytes), &mut &body[..])
576 .unwrap();
577 (oid, LfsSize::new(bytes))
578 }
579
580 fn backdate(store: &DiskStore, repo: &RepoDid, oid: &LfsOid, past: Duration) {
581 let path = store.object_file(repo, oid).unwrap().unwrap().1;
582 std::fs::OpenOptions::new()
583 .write(true)
584 .open(&path)
585 .unwrap()
586 .set_modified(SystemTime::now() - past)
587 .unwrap();
588 }
589
590 fn read_back(store: &dyn LfsStore, repo: &RepoDid, oid: &LfsOid) -> Vec<u8> {
591 let mut out = Vec::new();
592 store
593 .read(repo, oid)
594 .unwrap()
595 .read_to_end(&mut out)
596 .unwrap();
597 out
598 }
599
600 fn store_contract(store: &dyn LfsStore) {
601 let repo = RepoDid::new("did:plc:squid").unwrap();
602 let body: &[u8] = b"lfs media bytes for the round trip";
603 let oid = oid_of(body);
604 let size = LfsSize::new(body.len() as u64);
605
606 assert_eq!(store.probe(&repo, &oid).unwrap(), None);
607 assert!(matches!(
608 store.read(&repo, &oid),
609 Err(LfsError::NotFound { .. })
610 ));
611
612 let claim = ClaimedSize::new(size.get());
613 store.put(&repo, &oid, claim, &mut &body[..]).unwrap();
614 store.put(&repo, &oid, claim, &mut &body[..]).unwrap();
615 assert_eq!(
616 store.probe(&repo, &oid).unwrap(),
617 Some(size),
618 "a re-put of identical bytes is idempotent"
619 );
620
621 let other = RepoDid::new("did:plc:limpet").unwrap();
622 assert_eq!(store.probe(&other, &oid).unwrap(), None);
623 assert!(matches!(
624 store.read(&other, &oid),
625 Err(LfsError::NotFound { .. })
626 ));
627 }
628
629 #[test]
630 fn every_store_honors_the_contract() {
631 store_contract(&MemoryStore::new());
632 let (store, _dir) = disk();
633 store_contract(&store);
634 }
635
636 #[test]
637 fn a_body_longer_than_its_declared_size_errors_before_the_end() {
638 let store = MemoryStore::new();
639 let repo = RepoDid::new("did:plc:squid").unwrap();
640 let mut endless = std::io::repeat(0x5a);
641 assert!(matches!(
642 store.put(
643 &repo,
644 &oid_of(b"whatever"),
645 ClaimedSize::new(8),
646 &mut endless
647 ),
648 Err(LfsError::SizeMismatch { .. })
649 ));
650 }
651
652 #[test]
653 fn disk_writes_are_sharded_and_the_boot_sweep_clears_only_temp_files() {
654 let (store, dir) = disk();
655 let repo = RepoDid::new("did:plc:squid").unwrap();
656 let (oid, size) = seed(&store, &repo, b"sharded placement");
657 let sharded = dir
658 .path()
659 .join("plc/sq/uid")
660 .join(&oid.as_str()[0..2])
661 .join(&oid.as_str()[2..4])
662 .join(oid.as_str());
663 assert_eq!(std::fs::read(&sharded).unwrap(), b"sharded placement");
664
665 let method = RepoDid::new("did:incoming:squid").unwrap();
666 seed(
667 &store,
668 &method,
669 b"a method named incoming mustn't alias the temp dir",
670 );
671
672 let tampered = oid_of(b"a different object");
673 assert!(matches!(
674 store.put(&repo, &tampered, ClaimedSize::new(4), &mut &b"nope"[..]),
675 Err(LfsError::HashMismatch { .. })
676 ));
677 assert_eq!(store.probe(&repo, &tampered).unwrap(), None);
678 let incoming = dir.path().join(INCOMING_DIR);
679 assert!(
680 std::fs::read_dir(&incoming).unwrap().next().is_none(),
681 "a failed put leaves no files in the incoming dir"
682 );
683
684 std::fs::write(incoming.join("put-torn4321"), b"partial bytes from a crash").unwrap();
685 let store = DiskStore::open(LfsStorePath::new(dir.path())).unwrap();
686 assert!(
687 std::fs::read_dir(&incoming).unwrap().next().is_none(),
688 "the boot sweep clears abandoned uploads"
689 );
690 assert_eq!(store.probe(&repo, &oid).unwrap(), Some(size));
691 assert_eq!(read_back(&store, &repo, &oid), b"sharded placement");
692 }
693
694 #[test]
695 fn remove_repo_reclaims_the_prefix_and_spares_shard_neighbors() {
696 let (store, _dir) = disk();
697 let doomed = RepoDid::new("did:plc:squid").unwrap();
698 let neighbor = RepoDid::new("did:plc:squirrel").unwrap();
699 let (oid, size) = seed(&store, &doomed, b"prefix removal");
700 seed(&store, &neighbor, b"prefix removal");
701 store.remove_repo(&doomed).unwrap();
702 assert_eq!(store.probe(&doomed, &oid).unwrap(), None);
703 assert_eq!(store.probe(&neighbor, &oid).unwrap(), Some(size));
704 store.remove_repo(&doomed).unwrap();
705 }
706
707 #[test]
708 fn collect_touch_and_enumerate_govern_the_sweep_per_object() {
709 let (store, _dir) = disk();
710 let repo = RepoDid::new("did:plc:squid").unwrap();
711 assert!(
712 store
713 .enumerate(&RepoDid::new("did:plc:limpet").unwrap())
714 .unwrap()
715 .is_empty(),
716 "a missing prefix enumerates to nothing"
717 );
718
719 let (stale, stale_size) = seed(&store, &repo, b"long unreferenced");
720 let (fresh, _) = seed(&store, &repo, b"still within grace");
721 let (vouched, vouched_size) = seed(&store, &repo, b"vouched for moments before the sweep");
722 backdate(&store, &repo, &stale, MONTH);
723 backdate(&store, &repo, &vouched, MONTH);
724 let now = SystemTime::now();
725
726 let listed: HashSet<LfsOid> = store
727 .enumerate(&repo)
728 .unwrap()
729 .into_iter()
730 .map(|object| object.oid)
731 .collect();
732 assert_eq!(
733 listed,
734 HashSet::from([stale.clone(), fresh.clone(), vouched.clone()])
735 );
736
737 assert_eq!(
738 store.collect_expired(&repo, &fresh, GRACE, now).unwrap(),
739 Reclaimed::Spared,
740 "a fresh object is inside its grace window"
741 );
742 assert_eq!(
743 store.collect_expired(&repo, &stale, GRACE, now).unwrap(),
744 Reclaimed::Swept(stale_size)
745 );
746 assert_eq!(store.probe(&repo, &stale).unwrap(), None);
747 assert_eq!(
748 store.collect_expired(&repo, &stale, GRACE, now).unwrap(),
749 Reclaimed::Spared,
750 "collecting an already-gone object is a no-op"
751 );
752
753 assert_eq!(store.touch(&repo, &vouched).unwrap(), Some(vouched_size));
754 assert_eq!(
755 store
756 .collect_expired(&repo, &vouched, GRACE, SystemTime::now())
757 .unwrap(),
758 Reclaimed::Spared,
759 "the touch bumped the mtime inside the grace window"
760 );
761 assert_eq!(store.probe(&repo, &vouched).unwrap(), Some(vouched_size));
762 }
763
764 #[test]
765 fn the_orphan_sweep_reclaims_unregistered_prefixes_and_spares_every_other_class() {
766 let (store, dir) = disk();
767 let hosted = RepoDid::new("did:plc:squid").unwrap();
768 let orphan = RepoDid::new("did:plc:limpet").unwrap();
769 let fresh = RepoDid::new("did:plc:cuttle").unwrap();
770 let short_hosted = RepoDid::new("did:web:ab").unwrap();
771 let short_orphan = RepoDid::new("did:web:cd").unwrap();
772 assert_eq!(
773 RepoPrefix::new(&short_hosted)
774 .unwrap()
775 .as_path()
776 .components()
777 .count(),
778 2,
779 "a short method-specific-id shards to a two-component prefix"
780 );
781
782 let (kept, kept_size) = seed(&store, &hosted, b"belongs to a live repo");
783 let (doomed, doomed_size) = seed(&store, &orphan, b"repo was deleted");
784 let (spared, spared_size) = seed(&store, &fresh, b"deleted repo, but only just");
785 let (short_kept, short_kept_size) = seed(&store, &short_hosted, b"live short-did object");
786 let (short_doomed, short_doomed_size) =
787 seed(&store, &short_orphan, b"orphaned short-did media");
788 [
789 (&hosted, &kept),
790 (&orphan, &doomed),
791 (&short_hosted, &short_kept),
792 (&short_orphan, &short_doomed),
793 ]
794 .iter()
795 .for_each(|(did, oid)| backdate(&store, did, oid, MONTH));
796
797 std::fs::write(
798 dir.path().join(doomed.as_str()),
799 b"stray at the wrong depth",
800 )
801 .unwrap();
802
803 let registry = HashSet::from([hosted.clone(), short_hosted.clone()]);
804 let sweep = store
805 .sweep_orphans(®istry, GRACE, SystemTime::now())
806 .unwrap();
807
808 assert_eq!(sweep.prefixes, 2, "both past-grace orphans are reclaimed");
809 assert_eq!(sweep.objects, 2, "one object under each reclaimed prefix");
810 assert_eq!(sweep.bytes, doomed_size.saturating_add(short_doomed_size));
811 assert_eq!(store.probe(&orphan, &doomed).unwrap(), None);
812 assert_eq!(store.probe(&short_orphan, &short_doomed).unwrap(), None);
813 assert_eq!(
814 store.probe(&hosted, &kept).unwrap(),
815 Some(kept_size),
816 "a hosted prefix is never an orphan"
817 );
818 assert_eq!(
819 store.probe(&short_hosted, &short_kept).unwrap(),
820 Some(short_kept_size),
821 "a hosted repo shallower than the oid shards survives too"
822 );
823 assert_eq!(
824 store.probe(&fresh, &spared).unwrap(),
825 Some(spared_size),
826 "a fresh orphan is held by the grace window"
827 );
828 }
829}