Skip to main content

max / audiofiles

16.8 KB · 408 lines History Blame Raw
1 //! Blob-directory layout: the flat-to-sharded migration and its bookkeeping.
2 //!
3 //! Vaults created before 2026-07-29 keep every blob in one flat directory. That
4 //! collapses at scale: measured on a 289k-file library, import throughput fell
5 //! about 90% between an empty vault and a 40,000-entry one, and a *dedup* pass
6 //! doing strictly less work per file (no blob write, no fsync) still ran five
7 //! times slower than a full write into an empty directory. The cost is kernel
8 //! filesystem metadata work, so it cannot be optimised away on the read side; the
9 //! directory has to stop being flat. Numbers and method in wiki `af-benchmarks`.
10 //!
11 //! [`migrate_to_sharded`] relocates a vault forward. It is resumable and
12 //! idempotent, because on the filesystems that need it most a full sweep of a
13 //! large library takes long enough to be interrupted: every step is a rename to a
14 //! path derived from the blob's own content, so re-running continues rather than
15 //! repeating, and an interrupted sweep leaves a vault that still resolves (reads
16 //! check both layouts, see [`super::existing_blob_path`]).
17 //!
18 //! <!-- wiki: af-benchmarks -->
19
20 use std::path::Path;
21 use std::sync::atomic::{AtomicBool, Ordering};
22
23 use tracing::{instrument, warn};
24
25 use super::{SampleStore, blob_shard, store_blob_path};
26 use crate::config_key::ConfigKey;
27 use crate::db::Database;
28 use crate::error::{Result, io_err};
29
30 /// On-disk layout of a vault's blob directory.
31 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
32 pub enum BlobLayout {
33 /// Every blob directly in the store root: `{root}/{hash}.{ext}`. Pre-2026-07-29.
34 Flat,
35 /// Blobs under a hash-prefix shard: `{root}/{ab}/{hash}.{ext}`.
36 Sharded,
37 }
38
39 impl BlobLayout {
40 /// The `blob_layout` config value.
41 #[must_use]
42 pub const fn as_str(self) -> &'static str {
43 match self {
44 BlobLayout::Flat => "flat",
45 BlobLayout::Sharded => "sharded",
46 }
47 }
48 }
49
50 /// The layout recorded for this vault.
51 ///
52 /// Absent or unrecognised reads as [`BlobLayout::Flat`], which is the fail-safe
53 /// direction: it schedules a sweep that finds nothing on an already-sharded vault
54 /// (one `read_dir`), whereas defaulting to `Sharded` would leave a genuinely flat
55 /// vault permanently unmigrated and paying the cost this module exists to remove.
56 ///
57 /// Deliberately not a schema migration. The key's only job is to hold a default,
58 /// and a migration whose entire body inserts one default row is more moving parts
59 /// than reading `None` as `Flat`.
60 pub fn recorded_layout(db: &Database) -> Result<BlobLayout> {
61 Ok(match db.get_config(ConfigKey::BlobLayout)?.as_deref() {
62 Some(v) if v == BlobLayout::Sharded.as_str() => BlobLayout::Sharded,
63 _ => BlobLayout::Flat,
64 })
65 }
66
67 /// Outcome of a [`migrate_to_sharded`] pass.
68 #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
69 pub struct LayoutMigration {
70 /// Blobs relocated into a shard directory.
71 pub moved: usize,
72 /// Flat blobs discarded because the shard already held the same content.
73 pub deduped: usize,
74 /// Blobs that could not be relocated. Each is logged.
75 pub errors: usize,
76 /// The sweep stopped early because cancellation was requested.
77 pub cancelled: bool,
78 /// The root is clean and the vault is now recorded as [`BlobLayout::Sharded`].
79 pub completed: bool,
80 }
81
82 /// Split a store-root filename into `(hash, ext)`, or `None` if it is not a blob.
83 ///
84 /// Strict on purpose. The store root also holds shard directories, and can hold
85 /// `{hash}.{ext}.{pid}.tmp` leftovers from an import that died between create and
86 /// rename. Requiring exactly one dot and a 64-char lowercase-hex stem rejects both
87 /// a temp file (two dots) and anything a user dropped in by hand, so the sweep
88 /// only ever renames files it is certain are blobs.
89 fn parse_blob_name(name: &str) -> Option<(&str, &str)> {
90 let (hash, ext) = match name.split_once('.') {
91 Some((hash, ext)) => (hash, ext),
92 None => (name, ""),
93 };
94 let is_hash = hash.len() == 64
95 && hash
96 .bytes()
97 .all(|b| b.is_ascii_digit() || b.is_ascii_lowercase() && b.is_ascii_hexdigit());
98 if !is_hash || ext.contains('.') {
99 return None;
100 }
101 Some((hash, ext))
102 }
103
104 /// Every flat blob presently in the store root, as `(hash, ext)`.
105 ///
106 /// Non-recursive: shard directories are entries of the root and are skipped, so
107 /// this counts exactly the work a sweep still has left.
108 fn flat_blobs(store_root: &Path) -> Result<Vec<(String, String)>> {
109 let mut out = Vec::new();
110 let entries = std::fs::read_dir(store_root).map_err(|e| io_err(store_root, e))?;
111 for entry in entries.flatten() {
112 // A directory here is a shard (or something a user made); never a blob.
113 if !entry.file_type().is_ok_and(|t| t.is_file()) {
114 continue;
115 }
116 let name = entry.file_name();
117 let Some(name) = name.to_str() else { continue };
118 if let Some((hash, ext)) = parse_blob_name(name) {
119 out.push((hash.to_string(), ext.to_string()));
120 }
121 }
122 Ok(out)
123 }
124
125 /// How many flat blobs are still in the store root.
126 ///
127 /// Cheap enough to call before deciding whether to start a sweep: one `read_dir`.
128 pub fn count_flat_blobs(store_root: &Path) -> Result<usize> {
129 Ok(flat_blobs(store_root)?.len())
130 }
131
132 /// Relocate every flat blob in the store root into its hash-prefix shard.
133 ///
134 /// Resumable, idempotent and cancellable. `on_progress` is called with
135 /// `(done, total)` as each blob is handled. On a clean pass (nothing cancelled, no
136 /// errors, root verified empty of blobs afterwards) the vault is recorded as
137 /// [`BlobLayout::Sharded`] and [`LayoutMigration::completed`] is set; otherwise the
138 /// recorded layout is left alone so the next run picks up the remainder.
139 ///
140 /// No per-blob fsync. The relocation is a pure rename of content-addressed data, so
141 /// an untimely crash can only lose the *dirent update*, never bytes: the blob is
142 /// still at one of the two paths a read checks, and re-running the sweep finishes
143 /// the job. Paying a metadata flush per file would reproduce the per-file fsync cost
144 /// that the same measurement found was worth about 27% of import time.
145 #[instrument(skip_all)]
146 pub fn migrate_to_sharded(
147 store: &SampleStore,
148 db: &Database,
149 cancel: &AtomicBool,
150 on_progress: &mut dyn FnMut(usize, usize),
151 ) -> Result<LayoutMigration> {
152 let root = store.root();
153 let pending = flat_blobs(root)?;
154 let total = pending.len();
155 let mut report = LayoutMigration::default();
156
157 for (done, (hash, ext)) in pending.iter().enumerate() {
158 if cancel.load(Ordering::Acquire) {
159 report.cancelled = true;
160 break;
161 }
162 let src = super::legacy_flat_blob_path(root, hash, ext);
163 let dest = store_blob_path(root, hash, ext);
164 let shard_dir = root.join(blob_shard(hash));
165
166 if let Err(e) = std::fs::create_dir_all(&shard_dir) {
167 warn!(shard = %shard_dir.display(), "layout: shard create failed: {e}");
168 report.errors += 1;
169 continue;
170 }
171
172 // A blob already at the destination is the same content by construction,
173 // so the flat copy is redundant and should go. Size is still checked
174 // first: if they differ, one of them is a truncated blob from a
175 // pre-atomic-rename crash, and silently deleting either could destroy the
176 // intact one. That case is left alone for a human and counted as an error.
177 match (std::fs::metadata(&dest), std::fs::metadata(&src)) {
178 (Ok(d), Ok(s)) if d.len() == s.len() => match std::fs::remove_file(&src) {
179 Ok(()) => report.deduped += 1,
180 Err(e) => {
181 warn!(path = %src.display(), "layout: redundant flat blob unlink failed: {e}");
182 report.errors += 1;
183 }
184 },
185 (Ok(d), Ok(s)) => {
186 warn!(
187 path = %src.display(),
188 flat_len = s.len(),
189 sharded_len = d.len(),
190 "layout: size mismatch between flat and sharded blob, leaving both for inspection"
191 );
192 report.errors += 1;
193 }
194 _ => match std::fs::rename(&src, &dest) {
195 Ok(()) => report.moved += 1,
196 Err(e) => {
197 warn!(path = %src.display(), "layout: rename into shard failed: {e}");
198 report.errors += 1;
199 }
200 },
201 }
202 on_progress(done + 1, total);
203 }
204
205 // One directory fsync for the whole sweep, so the renames are durable without
206 // paying a metadata flush per blob. Best-effort: not every filesystem supports
207 // it, and failure here only weakens durability, never correctness.
208 if let Ok(d) = std::fs::File::open(root) {
209 let _ = d.sync_all();
210 }
211
212 // Only claim completion against a re-scan. A blob could have been written
213 // flat by another process between the enumeration and here, and recording
214 // `Sharded` over one would strand it: reads would still find it via the
215 // fallback, but nothing would ever move it.
216 if !report.cancelled && report.errors == 0 && count_flat_blobs(root)? == 0 {
217 db.set_config(ConfigKey::BlobLayout, BlobLayout::Sharded.as_str())?;
218 report.completed = true;
219 }
220
221 Ok(report)
222 }
223
224 #[cfg(test)]
225 mod tests {
226 use super::*;
227 use crate::SampleHash;
228
229 const HASH_A: &str = "aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899";
230 const HASH_B: &str = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff";
231
232 fn store_with_flat_blob(root: &Path, hash: &str, ext: &str, bytes: &[u8]) {
233 std::fs::write(super::super::legacy_flat_blob_path(root, hash, ext), bytes).unwrap();
234 }
235
236 #[test]
237 fn parse_blob_name_accepts_a_blob_with_and_without_extension() {
238 assert_eq!(parse_blob_name(HASH_A), Some((HASH_A, "")));
239 assert_eq!(
240 parse_blob_name(&format!("{HASH_A}.wav")),
241 Some((HASH_A, "wav"))
242 );
243 }
244
245 #[test]
246 fn parse_blob_name_rejects_temp_files_and_non_blobs() {
247 // The exact shape import leaves behind when it dies before the rename.
248 assert_eq!(parse_blob_name(&format!("{HASH_A}.wav.12345.tmp")), None);
249 assert_eq!(parse_blob_name("audiofiles.db"), None);
250 assert_eq!(parse_blob_name("notes.txt"), None);
251 // Uppercase hex is not the spelling the store writes.
252 assert_eq!(parse_blob_name(&HASH_A.to_uppercase()), None);
253 // Right charset, wrong length.
254 assert_eq!(parse_blob_name("aabbcc.wav"), None);
255 }
256
257 #[test]
258 fn sweep_relocates_flat_blobs_and_records_the_layout() {
259 let dir = tempfile::TempDir::new().unwrap();
260 let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
261 let root = dir.path().join("samples");
262 let store = SampleStore::new(&root).unwrap();
263
264 store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");
265 store_with_flat_blob(&root, HASH_B, "flac", b"bbbbbb");
266
267 assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Flat);
268 assert_eq!(count_flat_blobs(&root).unwrap(), 2);
269
270 let cancel = AtomicBool::new(false);
271 let mut seen = Vec::new();
272 let report =
273 migrate_to_sharded(&store, &db, &cancel, &mut |d, t| seen.push((d, t))).unwrap();
274
275 assert_eq!(report.moved, 2);
276 assert_eq!(report.deduped, 0);
277 assert_eq!(report.errors, 0);
278 assert!(report.completed);
279 assert_eq!(seen, vec![(1, 2), (2, 2)]);
280
281 // Bytes are where the sharded layout says, and gone from the flat one.
282 assert!(store_blob_path(&root, HASH_A, "wav").exists());
283 assert!(root.join("aa").is_dir());
284 assert!(!super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
285 assert_eq!(count_flat_blobs(&root).unwrap(), 0);
286 assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Sharded);
287
288 // And the store resolves them.
289 let resolved = store
290 .sample_path(&SampleHash::from_trusted(HASH_A.to_string()), "wav")
291 .unwrap();
292 assert_eq!(resolved, store_blob_path(&root, HASH_A, "wav"));
293 }
294
295 #[test]
296 fn sweep_is_idempotent() {
297 let dir = tempfile::TempDir::new().unwrap();
298 let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
299 let root = dir.path().join("samples");
300 let store = SampleStore::new(&root).unwrap();
301 store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");
302
303 let cancel = AtomicBool::new(false);
304 let first = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
305 let second = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
306
307 assert_eq!(first.moved, 1);
308 assert_eq!(second.moved, 0);
309 assert!(
310 second.completed,
311 "a clean already-sharded vault stays sharded"
312 );
313 assert!(store_blob_path(&root, HASH_A, "wav").exists());
314 }
315
316 #[test]
317 fn sweep_discards_a_redundant_flat_blob_matching_its_shard() {
318 let dir = tempfile::TempDir::new().unwrap();
319 let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
320 let root = dir.path().join("samples");
321 let store = SampleStore::new(&root).unwrap();
322
323 // Same content in both layouts: an interrupted sweep plus a repair write.
324 store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");
325 std::fs::create_dir_all(root.join("aa")).unwrap();
326 std::fs::write(store_blob_path(&root, HASH_A, "wav"), b"aaaa").unwrap();
327
328 let cancel = AtomicBool::new(false);
329 let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
330
331 assert_eq!(report.deduped, 1);
332 assert_eq!(report.moved, 0);
333 assert_eq!(report.errors, 0);
334 assert!(report.completed);
335 assert!(!super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
336 }
337
338 #[test]
339 fn sweep_leaves_a_size_mismatch_alone_and_does_not_complete() {
340 let dir = tempfile::TempDir::new().unwrap();
341 let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
342 let root = dir.path().join("samples");
343 let store = SampleStore::new(&root).unwrap();
344
345 // One of these is a truncated blob from a pre-atomic-rename crash. Deleting
346 // either could destroy the intact copy, so both must survive the sweep.
347 store_with_flat_blob(&root, HASH_A, "wav", b"aaaaaaaa");
348 std::fs::create_dir_all(root.join("aa")).unwrap();
349 std::fs::write(store_blob_path(&root, HASH_A, "wav"), b"aa").unwrap();
350
351 let cancel = AtomicBool::new(false);
352 let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
353
354 assert_eq!(report.errors, 1);
355 assert_eq!(report.moved, 0);
356 assert_eq!(report.deduped, 0);
357 assert!(
358 !report.completed,
359 "an unresolved blob must not record Sharded"
360 );
361 assert!(super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
362 assert!(store_blob_path(&root, HASH_A, "wav").exists());
363 assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Flat);
364 }
365
366 #[test]
367 fn cancelled_sweep_keeps_the_flat_layout_recorded() {
368 let dir = tempfile::TempDir::new().unwrap();
369 let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
370 let root = dir.path().join("samples");
371 let store = SampleStore::new(&root).unwrap();
372 store_with_flat_blob(&root, HASH_A, "wav", b"aaaa");
373
374 let cancel = AtomicBool::new(true);
375 let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
376
377 assert!(report.cancelled);
378 assert_eq!(report.moved, 0);
379 assert!(!report.completed);
380 assert_eq!(recorded_layout(&db).unwrap(), BlobLayout::Flat);
381 // The blob is untouched and still resolvable, so a cancel is not data loss.
382 assert!(super::super::legacy_flat_blob_path(&root, HASH_A, "wav").exists());
383 }
384
385 #[test]
386 fn sweep_ignores_temp_leftovers_and_shard_directories() {
387 let dir = tempfile::TempDir::new().unwrap();
388 let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
389 let root = dir.path().join("samples");
390 let store = SampleStore::new(&root).unwrap();
391
392 let tmp = root.join(format!("{HASH_A}.wav.999.tmp"));
393 std::fs::write(&tmp, b"partial").unwrap();
394 store_with_flat_blob(&root, HASH_B, "wav", b"bbbb");
395
396 let cancel = AtomicBool::new(false);
397 let report = migrate_to_sharded(&store, &db, &cancel, &mut |_, _| {}).unwrap();
398
399 assert_eq!(report.moved, 1, "only the real blob moves");
400 assert_eq!(report.errors, 0);
401 assert!(tmp.exists(), "a temp leftover is not the sweep's business");
402 assert!(
403 report.completed,
404 "a temp file is not a flat blob, so it must not block completion"
405 );
406 }
407 }
408