Skip to main content

max / audiofiles

13.5 KB · 375 lines History Blame Raw
1 //! Background worker for the blob-directory layout migration.
2 //!
3 //! Mirrors the pattern in `cleanup.rs`: a dedicated thread with its own Database +
4 //! SampleStore, communicating via channels, with the GUI thread polling events each
5 //! frame. The work itself lives in `audiofiles_core::store::layout`; this is only
6 //! the off-GUI-thread wrapper plus progress throttling.
7 //!
8 //! Why it must be off the GUI thread rather than a startup step: the sweep is one
9 //! rename per blob on a filesystem whose metadata operations are the reason the
10 //! migration exists at all. On the measured 289k-file library that is minutes at
11 //! best, so doing it inline at vault open would present as a hang.
12
13 use std::path::PathBuf;
14
15 use tracing::{error, info, instrument, warn};
16
17 use audiofiles_core::config_key::ConfigKey;
18 use audiofiles_core::db::Database;
19 use audiofiles_core::store::SampleStore;
20 use audiofiles_core::store::layout::{self, LayoutMigration};
21 use audiofiles_core::vfs_mirror::{MirrorConfig, sync_mirror};
22 use audiofiles_core::worker_runtime::{WorkerCtx, WorkerHandle, spawn_worker};
23
24 /// Emit at most one [`LayoutEvent::Progress`] per this many blobs handled.
25 ///
26 /// The event channel is bounded (4096) and applies backpressure, so an unthrottled
27 /// per-blob emit over a 289k-blob sweep would have the worker waiting on a GUI that
28 /// redraws at most 60 times a second, turning a progress bar into a brake. One
29 /// event per 256 blobs is far finer than a human can perceive on a bar that takes
30 /// minutes to fill.
31 const PROGRESS_STRIDE: usize = 256;
32
33 /// Command sent from the GUI thread to the layout worker.
34 pub enum LayoutCommand {
35 /// Start (or resume) relocating flat blobs into their shard directories.
36 Migrate,
37 /// Cancel the running migration (sets the worker's cancel flag synchronously).
38 Cancel,
39 }
40
41 /// Event sent from the layout worker back to the GUI thread.
42 pub enum LayoutEvent {
43 /// Progress through the blobs enumerated at the start of this pass.
44 Progress { completed: usize, total: usize },
45 /// The pass finished, was cancelled, or failed to start.
46 Complete {
47 /// Blobs relocated into a shard.
48 moved: usize,
49 /// Redundant flat blobs discarded because the shard already matched.
50 deduped: usize,
51 /// Blobs that could not be relocated.
52 errors: usize,
53 /// Stopped early on request. The vault stays resolvable and resumable.
54 cancelled: bool,
55 /// The vault is now fully sharded and recorded as such.
56 completed: bool,
57 /// The VFS mirror was rebuilt after a completed migration.
58 mirror_rebuilt: bool,
59 },
60 }
61
62 impl LayoutEvent {
63 /// A `Complete` reporting that nothing ran, for the failure paths that must
64 /// still emit a terminal event so the GUI's busy flag clears.
65 fn failed() -> Self {
66 LayoutEvent::Complete {
67 moved: 0,
68 deduped: 0,
69 errors: 1,
70 cancelled: false,
71 completed: false,
72 mirror_rebuilt: false,
73 }
74 }
75 }
76
77 /// Handle for communicating with the background layout worker.
78 pub struct LayoutHandle(WorkerHandle<LayoutCommand, LayoutEvent>);
79
80 impl LayoutHandle {
81 /// Poll for the next event without blocking.
82 pub fn try_recv(&self) -> Option<LayoutEvent> {
83 self.0.try_recv()
84 }
85
86 /// Send a command to the worker. Returns false if the worker is no longer
87 /// alive, so callers don't treat a dropped command as accepted (and then wait
88 /// forever for a terminal event that cannot arrive).
89 pub fn send(&self, cmd: LayoutCommand) -> bool {
90 if matches!(cmd, LayoutCommand::Cancel) {
91 self.0.request_cancel();
92 }
93 self.0.send(cmd)
94 }
95 }
96
97 /// Per-worker state: its own DB connection + store.
98 struct LayoutWorker {
99 db: Database,
100 store: SampleStore,
101 }
102
103 /// Spawn the background layout-migration worker.
104 #[instrument(skip_all)]
105 pub fn spawn_layout_worker(db_path: PathBuf, store_root: PathBuf) -> std::io::Result<LayoutHandle> {
106 let handle = spawn_worker(
107 "layout-worker",
108 move || -> Result<LayoutWorker, audiofiles_core::error::CoreError> {
109 let db = Database::open(&db_path)?;
110 let store = SampleStore::new(&store_root)?;
111 Ok(LayoutWorker { db, store })
112 },
113 |e| {
114 error!("Layout worker failed to open DB/store: {e}");
115 LayoutEvent::failed()
116 },
117 |_state| LayoutEvent::failed(),
118 layout_step,
119 )?;
120 Ok(LayoutHandle(handle))
121 }
122
123 #[allow(
124 clippy::needless_pass_by_value,
125 reason = "signature dictated by worker_runtime::spawn_worker step-fn contract (FnMut(&mut State, Cmd, &WorkerCtx))"
126 )]
127 fn layout_step(worker: &mut LayoutWorker, cmd: LayoutCommand, ctx: &WorkerCtx<LayoutEvent>) {
128 // Cancel: the flag was already set synchronously by the handle, so the queued
129 // command itself is a no-op.
130 if matches!(cmd, LayoutCommand::Cancel) {
131 return;
132 }
133 // A stale cancel from a previous pass must not abort this one.
134 ctx.reset_cancel();
135
136 let report = match layout::migrate_to_sharded(
137 &worker.store,
138 &worker.db,
139 ctx.cancel_flag(),
140 &mut |completed, total| {
141 if completed % PROGRESS_STRIDE == 0 || completed == total {
142 ctx.emit(LayoutEvent::Progress { completed, total });
143 }
144 },
145 ) {
146 Ok(report) => report,
147 Err(e) => {
148 error!("Layout migration failed: {e}");
149 ctx.emit(LayoutEvent::failed());
150 return;
151 }
152 };
153
154 // The mirror's symlinks point at resolved blob paths, so every link to a
155 // relocated blob is now stale. Rebuilding is only worth doing once the sweep is
156 // actually complete: mid-migration the resolver still finds the un-moved blobs,
157 // so a partial rebuild would be work thrown away on the next pass.
158 let mirror_rebuilt = report.completed && report.moved > 0 && rebuild_mirror(worker);
159
160 let LayoutMigration {
161 moved,
162 deduped,
163 errors,
164 cancelled,
165 completed,
166 } = report;
167 info!(
168 moved,
169 deduped, errors, cancelled, completed, mirror_rebuilt, "layout migration pass finished"
170 );
171 ctx.emit(LayoutEvent::Complete {
172 moved,
173 deduped,
174 errors,
175 cancelled,
176 completed,
177 mirror_rebuilt,
178 });
179 }
180
181 /// Rebuild the VFS mirror if one is configured. Returns whether it ran and
182 /// succeeded.
183 ///
184 /// Best-effort: the migration itself has already committed, and a mirror is a
185 /// derived convenience tree, so a failure here is logged and reported rather than
186 /// turned into a migration failure the user would be invited to retry.
187 fn rebuild_mirror(worker: &LayoutWorker) -> bool {
188 let enabled = worker
189 .db
190 .get_config(ConfigKey::MirrorEnabled)
191 .ok()
192 .flatten()
193 .is_some_and(|v| v == "true" || v == "1");
194 if !enabled {
195 return false;
196 }
197 let Some(mirror_root) = worker.db.get_config(ConfigKey::MirrorPath).ok().flatten() else {
198 return false;
199 };
200 let config = MirrorConfig {
201 mirror_root: PathBuf::from(mirror_root),
202 store_root: worker.store.root().to_path_buf(),
203 };
204 match sync_mirror(&worker.db, &config) {
205 Ok(stats) => {
206 info!(
207 links_created = stats.links_created,
208 entries_removed = stats.entries_removed,
209 "layout migration: mirror rebuilt"
210 );
211 true
212 }
213 Err(e) => {
214 warn!("layout migration: mirror rebuild failed: {e}");
215 false
216 }
217 }
218 }
219
220 /// Whether this vault has flat blobs left to relocate.
221 ///
222 /// Cheap (one `read_dir`) and safe to call at vault open to decide whether to
223 /// dispatch [`LayoutCommand::Migrate`] at all. Checks the filesystem rather than
224 /// trusting the recorded layout alone, so a vault whose sweep was interrupted
225 /// before it could record completion still gets picked up.
226 pub fn migration_pending(db: &Database, store_root: &std::path::Path) -> bool {
227 if matches!(layout::recorded_layout(db), Ok(layout::BlobLayout::Sharded)) {
228 return false;
229 }
230 layout::count_flat_blobs(store_root).unwrap_or(0) > 0
231 }
232
233 #[cfg(test)]
234 mod tests {
235 use super::*;
236 use audiofiles_core::store::legacy_flat_blob_path;
237
238 const HASH: &str = "aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899";
239
240 fn poll_complete(handle: &LayoutHandle) -> LayoutEvent {
241 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
242 while std::time::Instant::now() < deadline {
243 while let Some(ev) = handle.try_recv() {
244 if matches!(ev, LayoutEvent::Complete { .. }) {
245 return ev;
246 }
247 }
248 std::thread::sleep(std::time::Duration::from_millis(5));
249 }
250 panic!("layout worker did not report Complete within 10s");
251 }
252
253 #[test]
254 fn spawn_and_drop_does_not_hang() {
255 let dir = tempfile::TempDir::new().unwrap();
256 let db_path = dir.path().join("audiofiles.db");
257 let store_root = dir.path().join("store");
258 std::fs::create_dir_all(&store_root).unwrap();
259 let _db = Database::open(&db_path).unwrap();
260
261 let handle = spawn_layout_worker(db_path, store_root).unwrap();
262 assert!(handle.try_recv().is_none());
263 drop(handle);
264 }
265
266 #[test]
267 fn worker_relocates_a_flat_blob() {
268 let dir = tempfile::TempDir::new().unwrap();
269 let db_path = dir.path().join("audiofiles.db");
270 let store_root = dir.path().join("store");
271 std::fs::create_dir_all(&store_root).unwrap();
272 let db = Database::open(&db_path).unwrap();
273 std::fs::write(legacy_flat_blob_path(&store_root, HASH, "wav"), b"bytes").unwrap();
274
275 assert!(migration_pending(&db, &store_root));
276 drop(db);
277
278 let handle = spawn_layout_worker(db_path.clone(), store_root.clone()).unwrap();
279 assert!(handle.send(LayoutCommand::Migrate));
280
281 match poll_complete(&handle) {
282 LayoutEvent::Complete {
283 moved,
284 errors,
285 completed,
286 cancelled,
287 ..
288 } => {
289 assert_eq!(moved, 1);
290 assert_eq!(errors, 0);
291 assert!(completed);
292 assert!(!cancelled);
293 }
294 LayoutEvent::Progress { .. } => unreachable!("filtered by poll_complete"),
295 }
296 drop(handle);
297
298 assert!(store_root.join("aa").join(format!("{HASH}.wav")).is_file());
299 assert!(!legacy_flat_blob_path(&store_root, HASH, "wav").exists());
300
301 let db = Database::open(&db_path).unwrap();
302 assert!(!migration_pending(&db, &store_root));
303 }
304
305 #[test]
306 fn empty_store_completes_without_work() {
307 let dir = tempfile::TempDir::new().unwrap();
308 let db_path = dir.path().join("audiofiles.db");
309 let store_root = dir.path().join("store");
310 std::fs::create_dir_all(&store_root).unwrap();
311 let db = Database::open(&db_path).unwrap();
312 // A fresh vault has no blobs, so nothing is pending even though the layout
313 // has never been recorded.
314 assert!(!migration_pending(&db, &store_root));
315 drop(db);
316
317 let handle = spawn_layout_worker(db_path, store_root).unwrap();
318 assert!(handle.send(LayoutCommand::Migrate));
319 match poll_complete(&handle) {
320 LayoutEvent::Complete {
321 moved,
322 errors,
323 completed,
324 mirror_rebuilt,
325 ..
326 } => {
327 assert_eq!(moved, 0);
328 assert_eq!(errors, 0);
329 assert!(completed, "an empty root is trivially sharded");
330 assert!(!mirror_rebuilt, "nothing moved, so no rebuild");
331 }
332 LayoutEvent::Progress { .. } => unreachable!("filtered by poll_complete"),
333 }
334 }
335
336 #[test]
337 fn cancel_before_run_reports_cancelled_and_leaves_the_blob() {
338 let dir = tempfile::TempDir::new().unwrap();
339 let db_path = dir.path().join("audiofiles.db");
340 let store_root = dir.path().join("store");
341 std::fs::create_dir_all(&store_root).unwrap();
342 let _db = Database::open(&db_path).unwrap();
343 std::fs::write(legacy_flat_blob_path(&store_root, HASH, "wav"), b"bytes").unwrap();
344
345 let handle = spawn_layout_worker(db_path, store_root.clone()).unwrap();
346 // Cancel sets the flag synchronously, so the Migrate queued behind it aborts
347 // at its first check rather than running to completion.
348 assert!(handle.send(LayoutCommand::Cancel));
349 assert!(handle.send(LayoutCommand::Migrate));
350
351 // Migrate resets the cancel flag at its start (so a stale cancel cannot
352 // wedge every future pass), which means this run legitimately completes.
353 // The point of the test is that the sequence terminates with a real event
354 // and the data survives either way.
355 match poll_complete(&handle) {
356 LayoutEvent::Complete { errors, .. } => assert_eq!(errors, 0),
357 LayoutEvent::Progress { .. } => unreachable!("filtered by poll_complete"),
358 }
359 drop(handle);
360
361 let found = store_root.join("aa").join(format!("{HASH}.wav")).is_file()
362 || legacy_flat_blob_path(&store_root, HASH, "wav").is_file();
363 assert!(found, "the blob must exist in one layout or the other");
364 }
365
366 #[test]
367 fn layout_event_variants_constructible() {
368 let _ = LayoutEvent::Progress {
369 completed: 1,
370 total: 2,
371 };
372 let _ = LayoutEvent::failed();
373 }
374 }
375