Skip to main content

max / audiofiles

Fix content-address poisoning on import and move forge off the UI thread AF-1: import() wrote blobs with a bare fs::copy guarded by dest.exists(), so a crash mid-copy left a truncated blob at the canonical {hash}.{ext} path that dedup then trusted forever. Write to a temp path and atomically rename, and repair an already-present wrong-size blob via a cheap size check (no re-hash) - closing both the partial-write and trust-existing halves. Mirrors the atomic temp+rename the sync download path uses. AF-2: Sample Forge chop/conform ran synchronously on the egui render thread holding the DB mutex, freezing the headline feature on long files. Add audiofiles_core::forge::worker (mirrors edit::worker); split runner.rs into CPU chop_to_temp/conform_to_temp + GUI-side import_chop/import_conform. DirectBackend gains start_chop/start_conform and a forge worker; poll_events runs the fast import on the GUI thread when the worker signals completion. CPU work and the DB lock are now off the render thread. Also fix a pre-existing deref-in-borrowing-pattern hard error in audiofiles-train so cargo build --workspace is green. Tests: import_repairs_a_truncated_preexisting_blob, start_chop_runs_on_worker_and_imports_on_poll. 695 tests pass, clippy clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-12 19:59 UTC
Commit: 39d9f635288316c99b6e02536a9a05f517f20cf8
Parent: 4bec0ef
9 files changed, +751 insertions, -82 deletions
M crates/audiofiles-browser/src/backend/direct.rs +244
@@ -34,6 +34,24 @@ use crate::export::{ExportCommand, ExportHandle};
34 34 use crate::import::{ImportCommand, ImportEvent, ImportHandle, ImportStrategy};
35 35
36 36 use audiofiles_core::analysis::worker::{WorkerCommand, WorkerEvent, WorkerHandle};
37 + use audiofiles_core::forge::{ForgeCommand, ForgeEvent};
38 +
39 + /// Import context for the in-flight forge operation. The worker does the CPU
40 + /// work (decode/slice/conform/encode → temp files); this carries the VFS
41 + /// destination so the GUI thread can import the staged files when the worker
42 + /// signals completion.
43 + enum ForgePending {
44 + Chop {
45 + vfs_id: VfsId,
46 + parent_id: Option<NodeId>,
47 + },
48 + Conform {
49 + vfs_id: VfsId,
50 + parent_id: Option<NodeId>,
51 + sample_rate: u32,
52 + bit_depth: u16,
53 + },
54 + }
37 55
38 56 /// Direct backend: talks to SQLite and the sample store in-process.
39 57 pub struct DirectBackend {
@@ -46,6 +64,10 @@ pub struct DirectBackend {
46 64 export_worker: Mutex<Option<ExportHandle>>,
47 65 cleanup_worker: Mutex<Option<CleanupHandle>>,
48 66 edit_worker: Mutex<Option<EditWorkerHandle>>,
67 + forge_worker: Mutex<Option<audiofiles_core::forge::ForgeWorkerHandle>>,
68 + // Import context for the in-flight forge op, applied when its CPU stage
69 + // completes (the worker produces temp files; the GUI thread imports them).
70 + forge_pending: Mutex<Option<ForgePending>>,
49 71 // VP-tree indexes for fast search (lazy, invalidated on new analysis)
50 72 fingerprint_index: Mutex<Option<fingerprint::FingerprintIndex>>,
51 73 similarity_index: Mutex<Option<similarity::SimilarityIndex>>,
@@ -66,6 +88,8 @@ impl DirectBackend {
66 88 export_worker: Mutex::new(None),
67 89 cleanup_worker: Mutex::new(None),
68 90 edit_worker: Mutex::new(None),
91 + forge_worker: Mutex::new(None),
92 + forge_pending: Mutex::new(None),
69 93 fingerprint_index: Mutex::new(None),
70 94 similarity_index: Mutex::new(None),
71 95 #[cfg(feature = "device-profiles")]
@@ -819,6 +843,88 @@ impl Backend for DirectBackend {
819 843 )?)
820 844 }
821 845
846 + #[instrument(skip_all)]
847 + fn start_chop(
848 + &self,
849 + vfs_id: VfsId,
850 + hash: &str,
851 + ext: &str,
852 + name: &str,
853 + parent_id: Option<NodeId>,
854 + method: &ChopMethod,
855 + ) -> BackendResult<()> {
856 + // Resolve the source path under a brief lock; the heavy work runs on the
857 + // worker, and the import lands when ForgeEvent::ChopComplete is drained.
858 + let path = {
859 + let db = self.db.lock();
860 + audiofiles_core::store::resolve_file_path(&self.store, &db, hash, ext)?
861 + };
862 +
863 + if let Some(old) = self.forge_worker.lock().take() {
864 + old.send(ForgeCommand::Cancel);
865 + drop(old);
866 + }
867 +
868 + let handle = audiofiles_core::forge::spawn_forge_worker()
869 + .map_err(|e| BackendError::Other(format!("failed to spawn forge worker: {e}")))?;
870 + handle.send(ForgeCommand::Chop {
871 + source_path: path,
872 + base_name: name.to_string(),
873 + method: method.clone(),
874 + });
875 + *self.forge_worker.lock() = Some(handle);
876 + *self.forge_pending.lock() = Some(ForgePending::Chop { vfs_id, parent_id });
877 + Ok(())
878 + }
879 +
880 + #[instrument(skip_all)]
881 + fn start_conform(
882 + &self,
883 + vfs_id: VfsId,
884 + hash: &str,
885 + ext: &str,
886 + name: &str,
887 + parent_id: Option<NodeId>,
888 + target: &ConformTarget,
889 + ) -> BackendResult<()> {
890 + let (path, auto_trim) = {
891 + let db = self.db.lock();
892 + let auto_trim = db
893 + .conn()
894 + .query_row(
895 + "SELECT value FROM user_config WHERE key = ?1",
896 + [super::FORGE_AUTO_TRIM_OVERSHOOT_KEY],
897 + |row| row.get::<_, String>(0),
898 + )
899 + .ok()
900 + .is_some_and(|v| v == "true");
901 + let path = audiofiles_core::store::resolve_file_path(&self.store, &db, hash, ext)?;
902 + (path, auto_trim)
903 + };
904 +
905 + if let Some(old) = self.forge_worker.lock().take() {
906 + old.send(ForgeCommand::Cancel);
907 + drop(old);
908 + }
909 +
910 + let handle = audiofiles_core::forge::spawn_forge_worker()
911 + .map_err(|e| BackendError::Other(format!("failed to spawn forge worker: {e}")))?;
912 + handle.send(ForgeCommand::Conform {
913 + source_path: path,
914 + base_name: name.to_string(),
915 + target: target.clone(),
916 + auto_trim,
917 + });
918 + *self.forge_worker.lock() = Some(handle);
919 + *self.forge_pending.lock() = Some(ForgePending::Conform {
920 + vfs_id,
921 + parent_id,
922 + sample_rate: target.sample_rate,
923 + bit_depth: target.bit_depth,
924 + });
925 + Ok(())
926 + }
927 +
822 928 // --- Config ---
823 929
824 930 fn get_config(&self, key: &str) -> BackendResult<Option<String>> {
@@ -1264,6 +1370,101 @@ impl Backend for DirectBackend {
1264 1370 }
1265 1371 }
1266 1372
1373 + // Poll forge worker. Collect events first (releasing the worker lock),
1374 + // then run the import stage here on the GUI thread — the worker only did
1375 + // the CPU work (decode/slice/conform/encode → temp files), so the DB
1376 + // write is fast and never blocked the render thread.
1377 + let forge_events: Vec<ForgeEvent> = match *self.forge_worker.lock() {
1378 + Some(ref worker) => {
1379 + let mut evs = Vec::new();
1380 + while let Some(ev) = worker.try_recv() {
1381 + evs.push(ev);
1382 + }
1383 + evs
1384 + }
1385 + None => Vec::new(),
1386 + };
1387 + for ev in forge_events {
1388 + match ev {
1389 + ForgeEvent::Started => events.push(BackendEvent::ForgeStarted),
1390 + ForgeEvent::Error { error } => {
1391 + self.forge_pending.lock().take();
1392 + events.push(BackendEvent::ForgeError { error });
1393 + }
1394 + ForgeEvent::ChopComplete { staging } => {
1395 + let pending = self.forge_pending.lock().take();
1396 + if let Some(ForgePending::Chop { vfs_id, parent_id }) = pending {
1397 + let db = self.db.lock();
1398 + match audiofiles_core::forge::import_chop(
1399 + &self.store,
1400 + &db,
1401 + vfs_id,
1402 + parent_id,
1403 + staging,
1404 + ) {
1405 + Ok(r) => events.push(BackendEvent::ForgeChopComplete {
1406 + slice_count: r.slice_count,
1407 + }),
1408 + Err(e) => events.push(BackendEvent::ForgeError {
1409 + error: e.to_string(),
1410 + }),
1411 + }
1412 + } else {
1413 + // No matching pending context (cancelled/superseded) —
1414 + // drop the staged temp files rather than leak them.
1415 + for (_, p) in &staging.slices {
1416 + let _ = std::fs::remove_file(p);
1417 + }
1418 + }
1419 + }
1420 + ForgeEvent::ConformComplete { staging } => {
1421 + let pending = self.forge_pending.lock().take();
1422 + if let Some(ForgePending::Conform {
1423 + vfs_id,
1424 + parent_id,
1425 + sample_rate,
1426 + bit_depth,
1427 + }) = pending
1428 + {
1429 + let db = self.db.lock();
1430 + match audiofiles_core::forge::import_conform(
1431 + &self.store,
1432 + &db,
1433 + vfs_id,
1434 + parent_id,
1435 + staging,
1436 + ) {
1437 + Ok(r) => {
1438 + let overshoot = r.overshoot.map(|o| match o.action {
1439 + audiofiles_core::forge::OvershootAction::Trimmed { gain_db } => {
1440 + super::ForgeOvershoot::Trimmed {
1441 + peak_dbfs: o.peak_dbfs,
1442 + gain_db,
1443 + }
1444 + }
1445 + audiofiles_core::forge::OvershootAction::Flagged => {
1446 + super::ForgeOvershoot::Flagged {
1447 + peak_dbfs: o.peak_dbfs,
1448 + }
1449 + }
1450 + });
1451 + events.push(BackendEvent::ForgeConformComplete {
1452 + sample_rate,
1453 + bit_depth,
1454 + overshoot,
1455 + });
1456 + }
1457 + Err(e) => events.push(BackendEvent::ForgeError {
1458 + error: e.to_string(),
1459 + }),
1460 + }
1461 + } else {
1462 + let _ = std::fs::remove_file(&staging.temp_path);
1463 + }
1464 + }
1465 + }
1466 + }
1467 +
1267 1468 events
1268 1469 }
1269 1470 }
@@ -1481,6 +1682,49 @@ mod tests {
1481 1682 }
1482 1683
1483 1684 #[test]
1685 + fn start_chop_runs_on_worker_and_imports_on_poll() {
1686 + use audiofiles_core::forge::ChopMethod;
1687 + let dir = tempfile::TempDir::new().unwrap();
1688 + let db = Database::open_in_memory().unwrap();
1689 + let store = SampleStore::new(dir.path().join("store")).unwrap();
1690 + let backend = DirectBackend::new(db, store, dir.path().to_path_buf());
1691 +
1692 + let vfs_id = backend.create_vfs("Test").unwrap();
1693 + let samples: Vec<f32> = (0..2000).map(|i| ((i % 40) as f32 / 40.0) - 0.5).collect();
1694 + let src = dir.path().join("loop.wav");
1695 + write_float_wav(&src, 1, 44100, &samples);
1696 + let hash = backend.import_file(&src).unwrap();
1697 +
1698 + // Dispatch to the worker; returns immediately (no UI block).
1699 + backend
1700 + .start_chop(vfs_id, &hash, "wav", "loop.wav", None, &ChopMethod::EqualDivisions(4))
1701 + .unwrap();
1702 +
1703 + // Poll until the chop completes (worker does CPU; poll imports).
1704 + let mut slice_count = None;
1705 + for _ in 0..200 {
1706 + for ev in backend.poll_events() {
1707 + match ev {
1708 + BackendEvent::ForgeChopComplete { slice_count: n } => slice_count = Some(n),
1709 + BackendEvent::ForgeError { error } => panic!("forge error: {error}"),
1710 + _ => {}
1711 + }
1712 + }
1713 + if slice_count.is_some() {
1714 + break;
1715 + }
1716 + std::thread::sleep(std::time::Duration::from_millis(10));
1717 + }
1718 + assert_eq!(slice_count, Some(4), "chop should report 4 slices");
1719 +
1720 + // The slices landed in the VFS under "loop_slices".
1721 + let roots = backend.list_children(vfs_id, None).unwrap();
1722 + let slice_dir = roots.iter().find(|n| n.name == "loop_slices").unwrap();
1723 + let slices = backend.list_children(vfs_id, Some(slice_dir.id)).unwrap();
1724 + assert_eq!(slices.len(), 4);
1725 + }
1726 +
1727 + #[test]
1484 1728 #[cfg(feature = "device-profiles")]
1485 1729 fn device_conform_target_resolves_bundled() {
1486 1730 let backend = setup();
M crates/audiofiles-browser/src/backend/mod.rs +50
@@ -122,6 +122,30 @@ pub enum BackendEvent {
122 122 hash: String,
123 123 error: String,
124 124 },
125 +
126 + // Forge events (chop/conform run on a worker thread; the GUI imports the
127 + // staged result when the CPU stage completes)
128 + ForgeStarted,
129 + ForgeChopComplete {
130 + slice_count: usize,
131 + },
132 + ForgeConformComplete {
133 + sample_rate: u32,
134 + bit_depth: u16,
135 + overshoot: Option<ForgeOvershoot>,
136 + },
137 + ForgeError {
138 + error: String,
139 + },
140 + }
141 +
142 + /// Serializable summary of a conform overshoot, for the completion event.
143 + #[derive(Debug, serde::Serialize, serde::Deserialize)]
144 + pub enum ForgeOvershoot {
145 + /// Scaled down by `gain_db` so the true peak sits at full scale.
146 + Trimmed { peak_dbfs: f32, gain_db: f32 },
147 + /// Left untouched; the encoder clamp will hard-clip on store/playback.
148 + Flagged { peak_dbfs: f32 },
125 149 }
126 150
127 151 /// Serializable description of an imported folder (for IPC).
@@ -454,6 +478,32 @@ pub trait Backend: Send + Sync {
454 478 target: &ConformTarget,
455 479 ) -> BackendResult<ConformResult>;
456 480
481 + /// Start a chop on the forge worker thread. The slow decode/encode runs off
482 + /// the GUI thread; a [`BackendEvent::ForgeChopComplete`] (or `ForgeError`)
483 + /// arrives via [`Backend::poll_events`] once the slices are imported.
484 + fn start_chop(
485 + &self,
486 + vfs_id: VfsId,
487 + hash: &str,
488 + ext: &str,
489 + name: &str,
490 + parent_id: Option<NodeId>,
491 + method: &ChopMethod,
492 + ) -> BackendResult<()>;
493 +
494 + /// Start a conform on the forge worker thread. Emits
495 + /// [`BackendEvent::ForgeConformComplete`] (or `ForgeError`) via
496 + /// [`Backend::poll_events`] once the conformed sample is imported.
497 + fn start_conform(
498 + &self,
499 + vfs_id: VfsId,
500 + hash: &str,
501 + ext: &str,
502 + name: &str,
503 + parent_id: Option<NodeId>,
504 + target: &ConformTarget,
505 + ) -> BackendResult<()>;
506 +
457 507 // --- Config ---
458 508
459 509 /// Get a user config value by key.
M crates/audiofiles-browser/src/state/forge.rs +20 -40
@@ -94,6 +94,10 @@ impl BrowserState {
94 94 }
95 95
96 96 /// Chop the loaded sample into slices written into a new VFS folder.
97 + ///
98 + /// Dispatches the CPU work (decode/slice/encode) to the forge worker thread
99 + /// and returns immediately so the UI keeps painting; the result lands via
100 + /// `BackendEvent::ForgeChopComplete` in `poll_workers`.
97 101 pub fn forge_apply_chop(&mut self) {
98 102 let Some(hash) = self.forge.hash.clone() else { return };
99 103 let Some(vfs_id) = self.current_vfs_id() else {
@@ -106,17 +110,13 @@ impl BrowserState {
106 110 let parent_id = self.current_dir;
107 111
108 112 self.forge.busy = true;
109 - let result = self
113 + self.status = "Chopping...".to_string();
114 + if let Err(e) = self
110 115 .backend
111 - .chop_sample(vfs_id, &hash, &ext, &name, parent_id, &method);
112 - self.forge.busy = false;
113 -
114 - match result {
115 - Ok(count) => {
116 - self.status = format!("Chopped into {count} slices");
117 - self.refresh_contents();
118 - }
119 - Err(e) => self.status = format!("Chop failed: {e}"),
116 + .start_chop(vfs_id, &hash, &ext, &name, parent_id, &method)
117 + {
118 + self.forge.busy = false;
119 + self.status = format!("Chop failed: {e}");
120 120 }
121 121 }
122 122
@@ -147,37 +147,17 @@ impl BrowserState {
147 147 let parent_id = self.current_dir;
148 148
149 149 self.forge.busy = true;
150 - let result = self
150 + // Remember the device so the completion handler can name it in the
151 + // status message (the conform runs off-thread; the result/overshoot
152 + // arrive via BackendEvent::ForgeConformComplete in poll_workers).
153 + self.forge.conform_device = Some(device_name.to_string());
154 + self.status = format!("Conforming for {device_name}...");
155 + if let Err(e) = self
151 156 .backend
152 - .conform_sample(vfs_id, &hash, &ext, &name, parent_id, &target);
153 - self.forge.busy = false;
154 -
155 - match result {
156 - Ok(conformed) => {
157 - let mut msg = format!(
158 - "Conformed for {device_name} ({} Hz, {}-bit)",
159 - target.sample_rate, target.bit_depth
160 - );
161 - // Surface any true-peak overshoot so the encoder's clamp is never
162 - // a silent loss: either it was trimmed (opt-in) or it will clip.
163 - if let Some(o) = conformed.overshoot {
164 - use audiofiles_core::forge::OvershootAction;
165 - match o.action {
166 - OvershootAction::Trimmed { gain_db } => msg.push_str(&format!(
167 - " — true-peak +{:.1} dB trimmed {:.1} dB to avoid clipping",
168 - o.peak_dbfs,
169 - gain_db.abs()
170 - )),
171 - OvershootAction::Flagged => msg.push_str(&format!(
172 - " — warning: true-peak +{:.1} dB will clip (enable auto-trim in Settings to prevent)",
173 - o.peak_dbfs
174 - )),
175 - }
176 - }
177 - self.status = msg;
178 - self.refresh_contents();
179 - }
180 - Err(e) => self.status = format!("Conform failed: {e}"),
157 + .start_conform(vfs_id, &hash, &ext, &name, parent_id, &target)
158 + {
159 + self.forge.busy = false;
160 + self.status = format!("Conform failed: {e}");
181 161 }
182 162 }
183 163
M crates/audiofiles-browser/src/state/import_workflow.rs +45
@@ -660,6 +660,51 @@ impl BrowserState {
660 660 self.edit.in_progress = false;
661 661 self.status = format!("Edit failed: {error}");
662 662 }
663 +
664 + // --- Forge events (chop/conform now run off the GUI thread) ---
665 + BackendEvent::ForgeStarted => {
666 + self.forge.busy = true;
667 + }
668 + BackendEvent::ForgeChopComplete { slice_count } => {
669 + self.forge.busy = false;
670 + self.status = format!("Chopped into {slice_count} slices");
671 + self.refresh_contents();
672 + }
673 + BackendEvent::ForgeConformComplete {
674 + sample_rate,
675 + bit_depth,
676 + overshoot,
677 + } => {
678 + self.forge.busy = false;
679 + let device = self.forge.conform_device.clone().unwrap_or_default();
680 + let mut msg =
681 + format!("Conformed for {device} ({sample_rate} Hz, {bit_depth}-bit)");
682 + // Surface any true-peak overshoot so the encoder's clamp is
683 + // never a silent loss: either it was trimmed (opt-in) or it
684 + // will clip.
685 + if let Some(o) = overshoot {
686 + use crate::backend::ForgeOvershoot;
687 + match o {
688 + ForgeOvershoot::Trimmed { peak_dbfs, gain_db } => msg.push_str(
689 + &format!(
690 + " — true-peak +{:.1} dB trimmed {:.1} dB to avoid clipping",
691 + peak_dbfs,
692 + gain_db.abs()
693 + ),
694 + ),
695 + ForgeOvershoot::Flagged { peak_dbfs } => msg.push_str(&format!(
696 + " — warning: true-peak +{:.1} dB will clip (enable auto-trim in Settings to prevent)",
697 + peak_dbfs
698 + )),
699 + }
700 + }
701 + self.status = msg;
702 + self.refresh_contents();
703 + }
704 + BackendEvent::ForgeError { error } => {
705 + self.forge.busy = false;
706 + self.status = format!("Forge failed: {error}");
707 + }
663 708 }
664 709 }
665 710
M crates/audiofiles-core/src/forge/mod.rs +6 -1
@@ -20,6 +20,7 @@ pub mod batch;
20 20 pub mod chop;
21 21 pub mod conform;
22 22 pub mod runner;
23 + pub mod worker;
23 24
24 25 pub use batch::{find_content_bounds, trim_silence};
25 26 pub use chop::{compute_slices, detect_bpm, render_slice, ChopMethod, Slice};
@@ -27,4 +28,8 @@ pub use conform::{
27 28 conform, peak_amplitude, resolve_overshoot, ConformTarget, ConformedAudio, OvershootAction,
28 29 OvershootReport,
29 30 };
30 - pub use runner::{chop_to_vfs, conform_to_vfs, ChopResult, ConformResult};
31 + pub use runner::{
32 + chop_to_temp, chop_to_vfs, conform_to_temp, conform_to_vfs, import_chop, import_conform,
33 + ChopResult, ChopStaging, ConformResult, ConformStaging,
34 + };
35 + pub use worker::{spawn_forge_worker, ForgeCommand, ForgeEvent, ForgeWorkerHandle};
M crates/audiofiles-core/src/forge/runner.rs +127 -36
@@ -39,20 +39,39 @@ pub struct ConformResult {
39 39 pub overshoot: Option<OvershootReport>,
40 40 }
41 41
42 - /// Decode `source_path`, chop it with `method`, and write each slice as a new
43 - /// sample into a new `"{base_name}_slices"` directory under `parent_id`.
44 - ///
45 - /// Slices are encoded as 24-bit WAV (lossless headroom for further forging).
46 - /// Returns the directory node and slice count.
47 - pub fn chop_to_vfs(
48 - store: &SampleStore,
49 - db: &Database,
50 - vfs_id: VfsId,
42 + /// The CPU output of a chop: slices already decoded, rendered, and encoded to
43 + /// temp WAV files, ready to be imported into the store/VFS. Produced off the GUI
44 + /// thread (no DB access) so the heavy decode/encode never holds the DB lock.
45 + #[derive(Debug)]
46 + pub struct ChopStaging {
47 + /// Sanitized base name, used to build the `"{stem}_slices"` directory.
48 + pub stem: String,
49 + /// `(sample name, temp WAV path)` for each non-empty slice, numbered
50 + /// sequentially with no gaps.
51 + pub slices: Vec<(String, PathBuf)>,
52 + }
53 +
54 + /// The CPU output of a conform: the conformed signal already encoded to a temp
55 + /// WAV file, ready to import. Produced off the GUI thread.
56 + #[derive(Debug)]
57 + pub struct ConformStaging {
58 + /// Output sample name (`"{stem}_{rate}_{bits}bit.wav"`).
59 + pub out_name: String,
60 + /// Temp WAV path holding the conformed audio.
61 + pub temp_path: PathBuf,
62 + /// Overshoot report when an integer target pushed the true peak past full
63 + /// scale (see [`conform_to_vfs`]).
64 + pub overshoot: Option<OvershootReport>,
65 + }
66 +
67 + /// CPU stage of chop: decode `source_path`, slice it with `method`, and encode
68 + /// each non-empty slice to a temp WAV. No store/DB access — safe to run on a
69 + /// worker thread. The import stage is [`import_chop`].
70 + pub fn chop_to_temp(
51 71 source_path: &Path,
52 72 base_name: &str,
53 - parent_id: Option<NodeId>,
54 73 method: &ChopMethod,
55 - ) -> Result<ChopResult, CoreError> {
74 + ) -> Result<ChopStaging, CoreError> {
56 75 let decoded = decode_multichannel(source_path)?;
57 76 let slices = compute_slices(&decoded.samples, decoded.channels, decoded.sample_rate, method)?;
58 77 if slices.is_empty() {
@@ -60,16 +79,12 @@ pub fn chop_to_vfs(
60 79 }
61 80
62 81 let stem = sanitize_stem(base_name);
63 - let dir_name = format!("{stem}_slices");
64 - let dir_node = vfs::create_directory(db, vfs_id, parent_id, &dir_name)?;
65 -
66 82 let temp_dir = forge_temp_dir()?;
67 83 let width = slice_index_width(slices.len());
68 84
69 - // Count only slices actually written, and number them sequentially, so the
70 - // reported count never overstates what landed in the VFS and the file names
71 - // carry no gaps even if a slice renders empty.
72 - let mut written = 0usize;
85 + // Number sequentially over slices actually written, so names carry no gaps
86 + // even if a slice renders empty and the count never overstates the result.
87 + let mut staged: Vec<(String, PathBuf)> = Vec::new();
73 88 for slice in slices.iter() {
74 89 let buf = render_slice(&decoded.samples, decoded.channels, slice);
75 90 if buf.is_empty() {
@@ -80,22 +95,73 @@ pub fn chop_to_vfs(
80 95 sample_rate: decoded.sample_rate,
81 96 channels: decoded.channels,
82 97 };
83 - let slice_name = format!("{stem}_{:0width$}.wav", written + 1, width = width);
98 + let slice_name = format!("{stem}_{:0width$}.wav", staged.len() + 1, width = width);
84 99 let temp_path = temp_dir.join(&slice_name);
85 - encode_wav(&converted, 24, &temp_path)?;
100 + if let Err(e) = encode_wav(&converted, 24, &temp_path) {
101 + // Clean up any temp files already staged before bailing.
102 + for (_, p) in &staged {
103 + let _ = std::fs::remove_file(p);
104 + }
105 + return Err(e);
106 + }
107 + staged.push((slice_name, temp_path));
108 + }
109 +
110 + if staged.is_empty() {
111 + return Err(CoreError::Internal("chop produced no slices".to_string()));
112 + }
113 + Ok(ChopStaging { stem, slices: staged })
114 + }
86 115
87 - let import = store.import(&temp_path, db);
88 - let _ = std::fs::remove_file(&temp_path);
116 + /// Import stage of chop: create the `"{stem}_slices"` directory and import each
117 + /// staged temp WAV into the store/VFS. Runs on the GUI thread (fast DB writes;
118 + /// the slow decode/encode already happened in [`chop_to_temp`]). Consumes the
119 + /// staging and removes each temp file after import.
120 + pub fn import_chop(
121 + store: &SampleStore,
122 + db: &Database,
123 + vfs_id: VfsId,
124 + parent_id: Option<NodeId>,
125 + staging: ChopStaging,
126 + ) -> Result<ChopResult, CoreError> {
127 + let dir_name = format!("{}_slices", staging.stem);
128 + let dir_node = vfs::create_directory(db, vfs_id, parent_id, &dir_name)?;
129 +
130 + let mut written = 0usize;
131 + for (slice_name, temp_path) in &staging.slices {
132 + let import = store.import(temp_path, db);
133 + let _ = std::fs::remove_file(temp_path);
89 134 let hash = import?;
90 - vfs::create_sample_link(db, vfs_id, Some(dir_node), &slice_name, &hash)?;
135 + vfs::create_sample_link(db, vfs_id, Some(dir_node), slice_name, &hash)?;
91 136 written += 1;
92 137 }
93 138
94 139 Ok(ChopResult { dir_node, slice_count: written })
95 140 }
96 141
97 - /// Decode `source_path`, conform it to `target`, and write the result as a new
98 - /// sibling sample under `parent_id`.
142 + /// Decode `source_path`, chop it with `method`, and write each slice as a new
143 + /// sample into a new `"{base_name}_slices"` directory under `parent_id`.
144 + ///
145 + /// Slices are encoded as 24-bit WAV (lossless headroom for further forging).
146 + /// Returns the directory node and slice count. Convenience composition of
147 + /// [`chop_to_temp`] + [`import_chop`] for synchronous callers/tests; the GUI
148 + /// runs the two stages on separate threads.
149 + pub fn chop_to_vfs(
150 + store: &SampleStore,
151 + db: &Database,
152 + vfs_id: VfsId,
153 + source_path: &Path,
154 + base_name: &str,
155 + parent_id: Option<NodeId>,
156 + method: &ChopMethod,
157 + ) -> Result<ChopResult, CoreError> {
158 + let staging = chop_to_temp(source_path, base_name, method)?;
159 + import_chop(store, db, vfs_id, parent_id, staging)
160 + }
161 +
162 + /// CPU stage of conform: decode `source_path`, conform to `target`, resolve any
163 + /// overshoot per `auto_trim`, and encode to a temp WAV. No store/DB access. The
164 + /// import stage is [`import_conform`].
99 165 ///
100 166 /// The conformed f32 signal is carried pristine to the encode boundary. Resampling
101 167 /// can push inter-sample (true) peaks past `±1.0`; at an integer target the
@@ -109,17 +175,12 @@ pub fn chop_to_vfs(
109 175 ///
110 176 /// 32-bit float targets store the f32 verbatim and never clip, so the check is
111 177 /// skipped for them.
112 - #[allow(clippy::too_many_arguments)] // store/db/vfs context + target + the overshoot policy flag
113 - pub fn conform_to_vfs(
114 - store: &SampleStore,
115 - db: &Database,
116 - vfs_id: VfsId,
178 + pub fn conform_to_temp(
117 179 source_path: &Path,
118 180 base_name: &str,
119 - parent_id: Option<NodeId>,
120 181 target: &ConformTarget,
121 182 auto_trim: bool,
122 - ) -> Result<ConformResult, CoreError> {
183 + ) -> Result<ConformStaging, CoreError> {
123 184 let decoded = decode_multichannel(source_path)?;
124 185 let mut conformed = conform(&decoded.samples, decoded.channels, decoded.sample_rate, target)?;
125 186
@@ -139,11 +200,41 @@ pub fn conform_to_vfs(
139 200 let temp_path = temp_dir.join(&out_name);
140 201 encode_wav(&conformed.audio, conformed.bit_depth, &temp_path)?;
141 202
142 - let import = store.import(&temp_path, db);
143 - let _ = std::fs::remove_file(&temp_path);
203 + Ok(ConformStaging { out_name, temp_path, overshoot })
204 + }
205 +
206 + /// Import stage of conform: import the staged temp WAV into the store and link
207 + /// it as a new sibling sample under `parent_id`. Runs on the GUI thread.
208 + pub fn import_conform(
209 + store: &SampleStore,
210 + db: &Database,
211 + vfs_id: VfsId,
212 + parent_id: Option<NodeId>,
213 + staging: ConformStaging,
214 + ) -> Result<ConformResult, CoreError> {
215 + let import = store.import(&staging.temp_path, db);
216 + let _ = std::fs::remove_file(&staging.temp_path);
144 217 let hash = import?;
145 - vfs::create_sample_link(db, vfs_id, parent_id, &out_name, &hash)?;
146 - Ok(ConformResult { hash, overshoot })
218 + vfs::create_sample_link(db, vfs_id, parent_id, &staging.out_name, &hash)?;
219 + Ok(ConformResult { hash, overshoot: staging.overshoot })
220 + }
221 +
222 + /// Decode `source_path`, conform it to `target`, and write the result as a new
223 + /// sibling sample under `parent_id`. Convenience composition of
224 + /// [`conform_to_temp`] + [`import_conform`] for synchronous callers/tests.
225 + #[allow(clippy::too_many_arguments)] // store/db/vfs context + target + the overshoot policy flag
226 + pub fn conform_to_vfs(
227 + store: &SampleStore,
228 + db: &Database,
229 + vfs_id: VfsId,
230 + source_path: &Path,
231 + base_name: &str,
232 + parent_id: Option<NodeId>,
233 + target: &ConformTarget,
234 + auto_trim: bool,
235 + ) -> Result<ConformResult, CoreError> {
236 + let staging = conform_to_temp(source_path, base_name, target, auto_trim)?;
237 + import_conform(store, db, vfs_id, parent_id, staging)
147 238 }
148 239
149 240 /// Drop the extension and strip path-unfriendly characters from a base name.
A crates/audiofiles-core/src/forge/worker.rs +206
@@ -0,0 +1,206 @@
1 + //! Background forge worker thread: runs chop/conform CPU work off the GUI thread.
2 + //!
3 + //! Mirrors `edit::worker`. The heavy decode → slice/conform → encode runs here
4 + //! and produces temp WAV files ([`ChopStaging`]/[`ConformStaging`]); the GUI
5 + //! thread imports those into the store/VFS (a fast DB write) when the matching
6 + //! event arrives. This keeps the slow CPU work — and the DB lock — off the
7 + //! render thread, so the forge no longer freezes the UI on long files.
8 +
9 + use std::path::PathBuf;
10 + use std::sync::atomic::{AtomicBool, Ordering};
11 + use std::sync::{mpsc, Arc, Mutex};
12 + use std::thread;
13 +
14 + use tracing::instrument;
15 +
16 + use super::chop::ChopMethod;
17 + use super::conform::ConformTarget;
18 + use super::runner::{chop_to_temp, conform_to_temp, ChopStaging, ConformStaging};
19 +
20 + /// Command sent from the GUI thread to the forge worker.
21 + pub enum ForgeCommand {
22 + /// Chop a sample into slices (CPU stage only; GUI imports the result).
23 + Chop {
24 + source_path: PathBuf,
25 + base_name: String,
26 + method: ChopMethod,
27 + },
28 + /// Conform a sample to a device target (CPU stage only).
29 + Conform {
30 + source_path: PathBuf,
31 + base_name: String,
32 + target: ConformTarget,
33 + auto_trim: bool,
34 + },
35 + /// Cancel the current operation.
36 + Cancel,
37 + /// Shut down the worker thread.
38 + Shutdown,
39 + }
40 +
41 + /// Event sent from the forge worker back to the GUI thread.
42 + pub enum ForgeEvent {
43 + /// A forge operation started.
44 + Started,
45 + /// Chop CPU work finished; slices are staged on disk awaiting import.
46 + ChopComplete { staging: ChopStaging },
47 + /// Conform CPU work finished; the conformed file is staged awaiting import.
48 + ConformComplete { staging: ConformStaging },
49 + /// A forge operation failed.
50 + Error { error: String },
51 + }
52 +
53 + /// Handle for communicating with the background forge worker.
54 + pub struct ForgeWorkerHandle {
55 + cmd_tx: mpsc::Sender<ForgeCommand>,
56 + event_rx: Mutex<mpsc::Receiver<ForgeEvent>>,
57 + cancel_flag: Arc<AtomicBool>,
58 + thread: Option<thread::JoinHandle<()>>,
59 + }
60 +
61 + impl ForgeWorkerHandle {
62 + /// Poll for the next event without blocking.
63 + pub fn try_recv(&self) -> Option<ForgeEvent> {
64 + self.event_rx.lock().ok()?.try_recv().ok()
65 + }
66 +
67 + /// Send a command to the worker.
68 + pub fn send(&self, cmd: ForgeCommand) {
69 + if matches!(cmd, ForgeCommand::Cancel) {
70 + self.cancel_flag.store(true, Ordering::Release);
71 + }
72 + let _ = self.cmd_tx.send(cmd);
73 + }
74 + }
75 +
76 + impl Drop for ForgeWorkerHandle {
77 + fn drop(&mut self) {
78 + self.cancel_flag.store(true, Ordering::Release);
79 + let _ = self.cmd_tx.send(ForgeCommand::Shutdown);
80 + if let Some(handle) = self.thread.take() {
81 + let _ = handle.join();
82 + }
83 + }
84 + }
85 +
86 + /// Spawn the background forge worker thread.
87 + #[instrument(skip_all)]
88 + pub fn spawn_forge_worker() -> std::io::Result<ForgeWorkerHandle> {
89 + let (cmd_tx, cmd_rx) = mpsc::channel::<ForgeCommand>();
90 + let (event_tx, event_rx) = mpsc::channel::<ForgeEvent>();
91 + let cancel_flag = Arc::new(AtomicBool::new(false));
92 + let cancel_clone = Arc::clone(&cancel_flag);
93 +
94 + let thread = thread::Builder::new()
95 + .name("forge-worker".to_string())
96 + .spawn(move || {
97 + forge_worker_loop(cmd_rx, event_tx, cancel_clone);
98 + })?;
99 +
100 + Ok(ForgeWorkerHandle {
101 + cmd_tx,
102 + event_rx: Mutex::new(event_rx),
103 + cancel_flag,
104 + thread: Some(thread),
105 + })
106 + }
107 +
108 + fn forge_worker_loop(
109 + cmd_rx: mpsc::Receiver<ForgeCommand>,
110 + event_tx: mpsc::Sender<ForgeEvent>,
111 + cancel_flag: Arc<AtomicBool>,
112 + ) {
113 + while let Ok(cmd) = cmd_rx.recv() {
114 + match cmd {
115 + ForgeCommand::Shutdown => break,
116 + ForgeCommand::Cancel => continue,
117 + ForgeCommand::Chop {
118 + source_path,
119 + base_name,
120 + method,
121 + } => {
122 + cancel_flag.store(false, Ordering::Release);
123 + let _ = event_tx.send(ForgeEvent::Started);
124 + if cancel_flag.load(Ordering::Acquire) {
125 + continue;
126 + }
127 + match chop_to_temp(&source_path, &base_name, &method) {
128 + Ok(staging) => {
129 + // If cancelled mid-flight, drop the staged temp files
130 + // rather than handing them to the GUI.
131 + if cancel_flag.load(Ordering::Acquire) {
132 + for (_, p) in &staging.slices {
133 + let _ = std::fs::remove_file(p);
134 + }
135 + continue;
136 + }
137 + let _ = event_tx.send(ForgeEvent::ChopComplete { staging });
138 + }
139 + Err(e) => {
140 + let _ = event_tx.send(ForgeEvent::Error {
141 + error: e.to_string(),
142 + });
143 + }
144 + }
145 + }
146 + ForgeCommand::Conform {
147 + source_path,
148 + base_name,
149 + target,
150 + auto_trim,
151 + } => {
152 + cancel_flag.store(false, Ordering::Release);
153 + let _ = event_tx.send(ForgeEvent::Started);
154 + if cancel_flag.load(Ordering::Acquire) {
155 + continue;
156 + }
157 + match conform_to_temp(&source_path, &base_name, &target, auto_trim) {
158 + Ok(staging) => {
159 + if cancel_flag.load(Ordering::Acquire) {
160 + let _ = std::fs::remove_file(&staging.temp_path);
161 + continue;
162 + }
163 + let _ = event_tx.send(ForgeEvent::ConformComplete { staging });
164 + }
165 + Err(e) => {
166 + let _ = event_tx.send(ForgeEvent::Error {
167 + error: e.to_string(),
168 + });
169 + }
170 + }
171 + }
172 + }
173 + }
174 + }
175 +
176 + #[cfg(test)]
177 + mod tests {
178 + use super::*;
179 +
180 + #[test]
181 + fn forge_command_variants() {
182 + let _cancel = ForgeCommand::Cancel;
183 + let _shutdown = ForgeCommand::Shutdown;
184 + let _chop = ForgeCommand::Chop {
185 + source_path: PathBuf::from("/test.wav"),
186 + base_name: "test.wav".to_string(),
187 + method: ChopMethod::EqualDivisions(4),
188 + };
189 + }
190 +
191 + #[test]
192 + fn spawn_and_drop() {
193 + let handle = spawn_forge_worker().unwrap();
194 + assert!(handle.try_recv().is_none());
195 + drop(handle);
196 + }
197 +
198 + #[test]
199 + fn cancel_flag_set_on_cancel() {
200 + let handle = spawn_forge_worker().unwrap();
201 + handle.send(ForgeCommand::Cancel);
202 + std::thread::sleep(std::time::Duration::from_millis(10));
203 + assert!(handle.cancel_flag.load(Ordering::Relaxed));
204 + drop(handle);
205 + }
206 + }
M crates/audiofiles-core/src/store.rs +51 -3
@@ -162,10 +162,29 @@ impl SampleStore {
162 162 // Probe duration from file headers (cheap, no full decode)
163 163 let duration = probe_duration(path);
164 164
165 - // Copy file to store if not already present
165 + // Copy file to store if not already present *and intact*. Write to a
166 + // temp path and atomically rename: a crash / full disk mid-copy must not
167 + // leave a truncated blob at the canonical content-addressed path,
168 + // because dedup would then trust that corrupt blob under this hash
169 + // forever. We also repair an already-present blob whose size doesn't
170 + // match the source (a partial left by a pre-atomic-fix crash): since the
171 + // store is content-addressed, the correct blob's size is exactly the
172 + // source's, so a cheap size check catches truncation without re-hashing.
173 + // Mirrors the atomic temp+rename the sync download path uses. The pid
174 + // suffix keeps two concurrent imports of the same hash from colliding on
175 + // the temp file.
166 176 let dest = self.sample_path(&hash, &ext)?;
167 - if !dest.exists() {
168 - fs::copy(path, &dest).map_err(|e| io_err(&dest, e))?;
177 + let needs_write = match fs::metadata(&dest) {
178 + Ok(m) => m.len() != file_size as u64,
179 + Err(_) => true,
180 + };
181 + if needs_write {
182 + let tmp = dest.with_file_name(format!("{hash}.{ext}.{}.tmp", std::process::id()));
183 + fs::copy(path, &tmp).map_err(|e| io_err(&tmp, e))?;
184 + if let Err(e) = fs::rename(&tmp, &dest) {
185 + let _ = fs::remove_file(&tmp);
186 + return Err(io_err(&dest, e));
187 + }
169 188 }
170 189
171 190 // Insert into DB (ignore if hash already exists)
@@ -744,6 +763,35 @@ mod tests {
744 763 }
745 764
746 765 #[test]
766 + fn import_repairs_a_truncated_preexisting_blob() {
767 + // Reproduces the corrupt-blob trap: a crash mid-copy (or any partial
768 + // write) can leave a truncated file at the canonical content-addressed
769 + // path. A content-addressed store must never trust it — import must
770 + // detect the size mismatch and rewrite the correct bytes.
771 + let (dir, db, store) = setup();
772 + let content = b"the genuine full sample payload";
773 + let src = create_test_file(&dir, "kick.wav", content);
774 +
775 + // Pre-place a truncated blob at the canonical path the real import targets.
776 + let hash = format!("{:x}", Sha256::digest(content));
777 + let dest = store.sample_path(&hash, "wav").unwrap();
778 + fs::create_dir_all(dest.parent().unwrap()).unwrap();
779 + fs::write(&dest, b"trunc").unwrap();
780 +
781 + let imported = store.import(&src, &db).unwrap();
782 + assert_eq!(imported, hash);
783 +
784 + // The stored blob now matches the source bytes exactly (hash verified).
785 + let stored = fs::read(&dest).unwrap();
786 + assert_eq!(stored, content, "truncated blob must be repaired on import");
787 + assert_eq!(
788 + format!("{:x}", Sha256::digest(&stored)),
789 + hash,
790 + "repaired blob hashes back to its content address"
791 + );
792 + }
793 +
794 + #[test]
747 795 fn remove_deletes_file_and_row() {
748 796 let (dir, db, store) = setup();
749 797 let src = create_test_file(&dir, "snare.wav", b"snare data");
M crates/audiofiles-train/src/main.rs +2 -2
@@ -136,7 +136,7 @@ fn majority_class(labels: &[u8], num_classes: usize) -> u8 {
136 136 counts
137 137 .iter()
138 138 .enumerate()
139 - .max_by_key(|(_, &c)| c)
139 + .max_by_key(|(_, c)| **c)
140 140 .map(|(i, _)| i as u8)
141 141 .unwrap_or(0)
142 142 }
@@ -335,7 +335,7 @@ fn predict_forest(trees: &[TreeNode], num_classes: usize, features: &[f64; NUM_F
335 335 let (best_class, &best_count) = votes
336 336 .iter()
337 337 .enumerate()
338 - .max_by_key(|(_, &c)| c)
338 + .max_by_key(|(_, c)| **c)
339 339 .unwrap_or((0, &0));
340 340
341 341 (best_class as u8, best_count as f64 / total)