Skip to main content

max / audiofiles

Performance: forge cancellation, virtualized file list, audio + index fixes (ultra-fuzz) - Forge chop/conform now honor cancellation mid-job: a CoreError::Cancelled propagates from the slice/encode loop (chop) and around the resample (conform), cleaning staged temps; the worker treats it as a clean stop. Previously a cancelled job ran to completion then discarded its output. - start_chop/start_conform reuse the persistent forge worker (Cancel + queue) instead of drop+respawn, which joined the old thread ON the egui frame thread — a visible stall on a long in-flight job. - File list uses egui's virtualized body.rows: a folder with thousands of children no longer re-lays-out every row each frame. (Folder browse keeps no silent LIMIT — the query runs once per navigation, not per frame, and a hard cap would hide samples.) - Audio mix buffer is pre-sized off the cpal callback (Fixed size when known, generous ceiling otherwise) so the realtime thread never allocates. - The dead O(n) free find_similar is now #[cfg(test)]-gated as the index's brute-force oracle — no caller can wire the linear scan back into the UI. - M028 indexes audio_analysis filter columns (classification, musical_key, bpm, peak_db, duration) so a text-less global filter seeks instead of full-scanning. Tests: chop/conform cancellation honored + temps cleaned; migration version 28. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-18 21:27 UTC
Commit: 0626ba805a5191607dff6af644d73984f269974f
Parent: d0aeb6d
8 files changed, +170 insertions, -55 deletions
M crates/audiofiles-app/src/audio.rs +11 -2
@@ -82,7 +82,15 @@ fn build_stream<T: cpal::SizedSample + cpal::FromSample<f32>>(
82 82 channels: usize,
83 83 device_sample_rate: u32,
84 84 ) -> Result<Stream, AudioError> {
85 - let mut mix_buf: Vec<f32> = Vec::new();
85 + // Pre-size the mix buffer off the audio thread. Allocating inside the cpal
86 + // callback is a real-time violation (it can block and cause an xrun). A
87 + // Fixed config tells us the exact block size; otherwise reserve a generous
88 + // ceiling so the in-callback resize below never fires in practice.
89 + let initial_frames = match config.buffer_size {
90 + cpal::BufferSize::Fixed(n) => n as usize,
91 + cpal::BufferSize::Default => 8192,
92 + };
93 + let mut mix_buf: Vec<f32> = vec![0.0; initial_frames.saturating_mul(channels.max(1))];
86 94
87 95 let stream = device
88 96 .build_output_stream(
@@ -90,7 +98,8 @@ fn build_stream<T: cpal::SizedSample + cpal::FromSample<f32>>(
90 98 move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
91 99 let num_samples = data.len();
92 100
93 - // Resize mix buffer if needed (no per-callback allocation after first call)
101 + // Backstop for an unexpectedly large block (pre-sized above to
102 + // avoid allocating here on the steady-state path).
94 103 if mix_buf.len() < num_samples {
95 104 mix_buf.resize(num_samples, 0.0);
96 105 }
M crates/audiofiles-browser/src/backend/direct.rs +40 -25
@@ -1059,19 +1059,28 @@ impl Backend for DirectBackend {
1059 1059 audiofiles_core::store::resolve_file_path(&self.store, &db, hash, ext)?
1060 1060 };
1061 1061
1062 - if let Some(old) = self.forge_worker.lock().take() {
1063 - old.send(ForgeCommand::Cancel);
1064 - drop(old);
1062 + // Reuse the persistent worker. Sending Cancel sets the cancel flag so any
1063 + // in-flight job bails cooperatively; the queued Chop then runs. Dropping
1064 + // the old handle here would join its thread ON the GUI thread — a visible
1065 + // frame stall on a long in-flight job. The worker resets the flag when it
1066 + // dequeues the new command.
1067 + {
1068 + let mut guard = self.forge_worker.lock();
1069 + if guard.is_none() {
1070 + *guard = Some(
1071 + audiofiles_core::forge::spawn_forge_worker().map_err(|e| {
1072 + BackendError::Other(format!("failed to spawn forge worker: {e}"))
1073 + })?,
1074 + );
1075 + }
1076 + let handle = guard.as_ref().expect("forge worker present");
1077 + handle.send(ForgeCommand::Cancel);
1078 + handle.send(ForgeCommand::Chop {
1079 + source_path: path,
1080 + base_name: name.to_string(),
1081 + method: method.clone(),
1082 + });
1065 1083 }
1066 -
1067 - let handle = audiofiles_core::forge::spawn_forge_worker()
1068 - .map_err(|e| BackendError::Other(format!("failed to spawn forge worker: {e}")))?;
1069 - handle.send(ForgeCommand::Chop {
1070 - source_path: path,
1071 - base_name: name.to_string(),
1072 - method: method.clone(),
1073 - });
1074 - *self.forge_worker.lock() = Some(handle);
1075 1084 *self.forge_pending.lock() = Some(ForgePending::Chop { vfs_id, parent_id });
1076 1085 Ok(())
1077 1086 }
@@ -1101,20 +1110,26 @@ impl Backend for DirectBackend {
1101 1110 (path, auto_trim)
1102 1111 };
1103 1112
1104 - if let Some(old) = self.forge_worker.lock().take() {
1105 - old.send(ForgeCommand::Cancel);
1106 - drop(old);
1113 + // Reuse the persistent worker (see start_chop): Cancel + queue, never
1114 + // drop+join on the GUI thread.
1115 + {
1116 + let mut guard = self.forge_worker.lock();
1117 + if guard.is_none() {
1118 + *guard = Some(
1119 + audiofiles_core::forge::spawn_forge_worker().map_err(|e| {
1120 + BackendError::Other(format!("failed to spawn forge worker: {e}"))
1121 + })?,
1122 + );
1123 + }
1124 + let handle = guard.as_ref().expect("forge worker present");
1125 + handle.send(ForgeCommand::Cancel);
1126 + handle.send(ForgeCommand::Conform {
1127 + source_path: path,
1128 + base_name: name.to_string(),
1129 + target: target.clone(),
1130 + auto_trim,
1131 + });
1107 1132 }
1108 -
1109 - let handle = audiofiles_core::forge::spawn_forge_worker()
1110 - .map_err(|e| BackendError::Other(format!("failed to spawn forge worker: {e}")))?;
1111 - handle.send(ForgeCommand::Conform {
1112 - source_path: path,
1113 - base_name: name.to_string(),
1114 - target: target.clone(),
1115 - auto_trim,
1116 - });
1117 - *self.forge_worker.lock() = Some(handle);
1118 1133 *self.forge_pending.lock() = Some(ForgePending::Conform {
1119 1134 vfs_id,
1120 1135 parent_id,
M crates/audiofiles-browser/src/ui/file_list.rs +15 -12
@@ -283,10 +283,15 @@ pub fn draw_file_list(
283 283 ui.label(egui::RichText::new("Play").color(theme::content_muted()));
284 284 });
285 285 })
286 - .body(|mut body| {
287 - // ".." parent entry
288 - if has_parent {
289 - body.row(row_height, |mut row| {
286 + .body(|body| {
287 + // Virtualized: egui lays out only the visible rows, so a folder with
288 + // thousands of children no longer re-lays-out every row each frame.
289 + // Display index 0 is the ".." parent entry when present; the rest map
290 + // into `contents`.
291 + let parent_rows = if has_parent { 1usize } else { 0 };
292 + body.rows(row_height, parent_rows + contents.len(), |mut row| {
293 + let display_idx = row.index();
294 + if has_parent && display_idx == 0 {
290 295 let selected = state.selection.contains(0);
291 296 row.set_selected(selected);
292 297 row.col(|ui| {
@@ -312,12 +317,10 @@ pub fn draw_file_list(
312 317 if show_peak_db { row.col(|_ui| {}); }
313 318 if show_tags { row.col(|_ui| {}); }
314 319 row.col(|_ui| {});
315 - });
316 - }
317 -
318 - for (i, node) in contents.iter().enumerate() {
319 - let row_idx = i + offset;
320 - body.row(row_height, |mut row| {
320 + } else {
321 + let i = display_idx - parent_rows;
322 + let node = &contents[i];
323 + let row_idx = i + offset;
321 324 let selected = state.selection.contains(row_idx);
322 325 row.set_selected(selected);
323 326
@@ -380,8 +383,8 @@ pub fn draw_file_list(
380 383 }
381 384 }
382 385 });
383 - });
384 - }
386 + }
387 + });
385 388 });
386 389
387 390 // Apply sort toggle after the table is fully drawn, so we don't conflict
M crates/audiofiles-core/src/db.rs +20 -6
@@ -1449,6 +1449,19 @@ CREATE TRIGGER IF NOT EXISTS vfs_nodes_fts_update AFTER UPDATE OF name ON vfs_no
1449 1449 END;
1450 1450 "#;
1451 1451
1452 + const MIGRATION_028: &str = r#"
1453 + -- Indexes for the search filter columns on audio_analysis. A text-less global
1454 + -- filter (e.g. BPM range or a classification, with no name query) otherwise
1455 + -- full-scans audio_analysis; these let the planner seek instead. classification
1456 + -- and musical_key are equality/IN filters (most selective); bpm/peak_db/duration
1457 + -- are range filters. All additive and idempotent.
1458 + CREATE INDEX IF NOT EXISTS idx_analysis_classification ON audio_analysis(classification);
1459 + CREATE INDEX IF NOT EXISTS idx_analysis_musical_key ON audio_analysis(musical_key);
1460 + CREATE INDEX IF NOT EXISTS idx_analysis_bpm ON audio_analysis(bpm);
1461 + CREATE INDEX IF NOT EXISTS idx_analysis_peak_db ON audio_analysis(peak_db);
1462 + CREATE INDEX IF NOT EXISTS idx_analysis_duration ON audio_analysis(duration);
1463 + "#;
1464 +
1452 1465 /// Register `hash_row_id(salt, key) -> TEXT` as a deterministic SQLite
1453 1466 /// function on the given connection. Used by the M018 sync triggers so the
1454 1467 /// `sync_changelog.row_id` field never carries cleartext content (tag strings,
@@ -1567,6 +1580,7 @@ impl Database {
1567 1580 MIGRATION_025,
1568 1581 MIGRATION_026,
1569 1582 MIGRATION_027,
1583 + MIGRATION_028,
1570 1584 ];
1571 1585
1572 1586 for (i, sql) in MIGRATIONS.iter().enumerate() {
@@ -1778,7 +1792,7 @@ mod tests {
1778 1792 .conn()
1779 1793 .query_row("PRAGMA user_version", [], |row| row.get(0))
1780 1794 .unwrap();
1781 - assert_eq!(version, 27);
1795 + assert_eq!(version, 28);
1782 1796 }
1783 1797
1784 1798 #[test]
@@ -1789,7 +1803,7 @@ mod tests {
1789 1803 .conn()
1790 1804 .query_row("PRAGMA user_version", [], |row| row.get(0))
1791 1805 .unwrap();
1792 - assert_eq!(version, 27);
1806 + assert_eq!(version, 28);
1793 1807 }
1794 1808
1795 1809 /// Open a fresh file-backed DB, close, reopen. The second open re-enters
@@ -1808,7 +1822,7 @@ mod tests {
1808 1822 .conn()
1809 1823 .query_row("PRAGMA user_version", [], |row| row.get(0))
1810 1824 .unwrap();
1811 - assert_eq!(version, 27);
1825 + assert_eq!(version, 28);
1812 1826 }
1813 1827
1814 1828 /// Simulates the worst-case recovery path: a prior partial migration left
@@ -1852,7 +1866,7 @@ mod tests {
1852 1866 .conn()
1853 1867 .query_row("PRAGMA user_version", [], |row| row.get(0))
1854 1868 .unwrap();
1855 - assert_eq!(version, 27);
1869 + assert_eq!(version, 28);
1856 1870 }
1857 1871
1858 1872 /// M018 contract: the `sync_changelog.row_id` for sensitive tables must
@@ -2074,7 +2088,7 @@ mod tests {
2074 2088 let initial_version: i32 = conn
2075 2089 .query_row("PRAGMA user_version", [], |row| row.get(0))
2076 2090 .unwrap();
2077 - assert_eq!(initial_version, 27);
2091 + assert_eq!(initial_version, 28);
2078 2092
2079 2093 let batch = format!(
2080 2094 "BEGIN;\n{}\nPRAGMA user_version = 999;\nCOMMIT;",
@@ -2137,7 +2151,7 @@ mod tests {
2137 2151 .conn()
2138 2152 .query_row("PRAGMA user_version", [], |row| row.get(0))
2139 2153 .unwrap();
2140 - assert_eq!(version, 27);
2154 + assert_eq!(version, 28);
2141 2155 }
2142 2156
2143 2157 #[test]
M crates/audiofiles-core/src/error.rs +6
@@ -83,6 +83,12 @@ pub enum CoreError {
83 83 /// Message is shown to the user verbatim (no prefix).
84 84 #[error("{0}")]
85 85 Incompatible(String),
86 +
87 + /// The operation was cancelled cooperatively (e.g. a forge job the user
88 + /// aborted mid-flight). Callers should treat this as a clean stop, not a
89 + /// failure to surface.
90 + #[error("operation cancelled")]
91 + Cancelled,
86 92 }
87 93
88 94 /// Typed errors for the audio analysis/decode pipeline.
M crates/audiofiles-core/src/forge/runner.rs +62 -2
@@ -6,6 +6,7 @@
6 6 //! device" actions. Originals are never touched — every result is a new sample.
7 7
8 8 use std::path::{Path, PathBuf};
9 + use std::sync::atomic::{AtomicBool, Ordering};
9 10
10 11 use crate::db::Database;
11 12 use crate::error::{io_err, CoreError};
@@ -90,6 +91,7 @@ pub fn chop_to_temp(
90 91 source_path: &Path,
91 92 base_name: &str,
92 93 method: &ChopMethod,
94 + cancel: &AtomicBool,
93 95 ) -> Result<ChopStaging, CoreError> {
94 96 let decoded = decode_multichannel(source_path)?;
95 97 let slices = compute_slices(&decoded.samples, decoded.channels, decoded.sample_rate, method)?;
@@ -105,6 +107,14 @@ pub fn chop_to_temp(
105 107 // even if a slice renders empty and the count never overstates the result.
106 108 let mut staged: Vec<(String, PathBuf)> = Vec::new();
107 109 for slice in slices.iter() {
110 + // Honor cancellation between slices: a chop of a long file into hundreds
111 + // of slices must stop promptly, not encode every slice then discard.
112 + if cancel.load(Ordering::Acquire) {
113 + for (_, p) in &staged {
114 + let _ = std::fs::remove_file(p);
115 + }
116 + return Err(CoreError::Cancelled);
117 + }
108 118 let buf = render_slice(&decoded.samples, decoded.channels, slice);
109 119 if buf.is_empty() {
110 120 continue;
@@ -176,7 +186,7 @@ pub fn chop_to_vfs(
176 186 parent_id: Option<NodeId>,
177 187 method: &ChopMethod,
178 188 ) -> Result<ChopResult, CoreError> {
179 - let staging = chop_to_temp(source_path, base_name, method)?;
189 + let staging = chop_to_temp(source_path, base_name, method, &AtomicBool::new(false))?;
180 190 import_chop(store, db, vfs_id, parent_id, staging)
181 191 }
182 192
@@ -201,9 +211,17 @@ pub fn conform_to_temp(
201 211 base_name: &str,
202 212 target: &ConformTarget,
203 213 auto_trim: bool,
214 + cancel: &AtomicBool,
204 215 ) -> Result<ConformStaging, CoreError> {
205 216 let decoded = decode_multichannel(source_path)?;
217 + if cancel.load(Ordering::Acquire) {
218 + return Err(CoreError::Cancelled);
219 + }
206 220 let mut conformed = conform(&decoded.samples, decoded.channels, decoded.sample_rate, target)?;
221 + // Bail before the encode (the last expensive step) if cancelled mid-resample.
222 + if cancel.load(Ordering::Acquire) {
223 + return Err(CoreError::Cancelled);
224 + }
207 225
208 226 // Integer targets clamp at the encoder; 32-bit float is a lossless f32
209 227 // passthrough with no clamp, so only the integer path can overshoot.
@@ -256,7 +274,7 @@ pub fn conform_to_vfs(
256 274 target: &ConformTarget,
257 275 auto_trim: bool,
258 276 ) -> Result<ConformResult, CoreError> {
259 - let staging = conform_to_temp(source_path, base_name, target, auto_trim)?;
277 + let staging = conform_to_temp(source_path, base_name, target, auto_trim, &AtomicBool::new(false))?;
260 278 import_conform(store, db, vfs_id, parent_id, staging)
261 279 }
262 280
@@ -396,6 +414,48 @@ mod tests {
396 414 }
397 415
398 416 #[test]
417 + fn chop_honors_cancellation_and_cleans_temps() {
418 + let dir = tempfile::tempdir().unwrap();
419 + let samples: Vec<f32> = (0..4000).map(|i| ((i % 50) as f32 / 50.0) - 0.5).collect();
420 + let src = dir.path().join("loop.wav");
421 + write_test_wav(&src, 1, 44100, &samples);
422 +
423 + // Pre-cancelled: chop must bail before staging any slice.
424 + let cancel = AtomicBool::new(true);
425 + let result = chop_to_temp(&src, "loop.wav", &ChopMethod::EqualDivisions(8), &cancel);
426 + assert!(matches!(result, Err(CoreError::Cancelled)));
427 +
428 + // No orphaned temp WAVs left behind for this stem.
429 + if let Ok(temp_dir) = forge_temp_dir() {
430 + let leaked = std::fs::read_dir(&temp_dir)
431 + .map(|rd| {
432 + rd.filter_map(|e| e.ok())
433 + .filter(|e| e.file_name().to_string_lossy().starts_with("loop_"))
434 + .count()
435 + })
436 + .unwrap_or(0);
437 + assert_eq!(leaked, 0, "cancelled chop must not leak temp slices");
438 + }
439 + }
440 +
441 + #[test]
442 + fn conform_honors_cancellation() {
443 + let dir = tempfile::tempdir().unwrap();
444 + let samples: Vec<f32> = (0..2000).map(|i| ((i % 50) as f32 / 50.0) - 0.5).collect();
445 + let src = dir.path().join("one.wav");
446 + write_test_wav(&src, 1, 44100, &samples);
447 +
448 + let cancel = AtomicBool::new(true);
449 + let target = ConformTarget {
450 + sample_rate: 22050,
451 + bit_depth: 16,
452 + channels: crate::export::ExportChannels::Original,
453 + };
454 + let result = conform_to_temp(&src, "one.wav", &target, false, &cancel);
455 + assert!(matches!(result, Err(CoreError::Cancelled)));
456 + }
457 +
458 + #[test]
399 459 fn conform_creates_sibling_sample() {
400 460 let dir = tempfile::tempdir().unwrap();
401 461 let db = Database::open_in_memory().unwrap();
M crates/audiofiles-core/src/forge/worker.rs +7 -2
@@ -13,6 +13,7 @@ use std::thread;
13 13
14 14 use tracing::instrument;
15 15
16 + use crate::error::CoreError;
16 17 use super::chop::ChopMethod;
17 18 use super::conform::ConformTarget;
18 19 use super::runner::{chop_to_temp, compute_preview_marks, conform_to_temp, ChopStaging, ConformStaging};
@@ -135,7 +136,7 @@ fn forge_worker_loop(
135 136 // Isolate panics: a bad file must report an error, not unwind
136 137 // the worker thread and silently stop all future forge jobs.
137 138 let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
138 - chop_to_temp(&source_path, &base_name, &method)
139 + chop_to_temp(&source_path, &base_name, &method, &cancel_flag)
139 140 }));
140 141 match outcome {
141 142 Ok(Ok(staging)) => {
@@ -149,6 +150,8 @@ fn forge_worker_loop(
149 150 }
150 151 let _ = event_tx.send(ForgeEvent::ChopComplete { staging });
151 152 }
153 + // Cooperative cancel mid-chop: clean stop, no error surfaced.
154 + Ok(Err(CoreError::Cancelled)) => continue,
152 155 Ok(Err(e)) => {
153 156 let _ = event_tx.send(ForgeEvent::Error {
154 157 error: e.to_string(),
@@ -173,7 +176,7 @@ fn forge_worker_loop(
173 176 continue;
174 177 }
175 178 let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
176 - conform_to_temp(&source_path, &base_name, &target, auto_trim)
179 + conform_to_temp(&source_path, &base_name, &target, auto_trim, &cancel_flag)
177 180 }));
178 181 match outcome {
179 182 Ok(Ok(staging)) => {
@@ -183,6 +186,8 @@ fn forge_worker_loop(
183 186 }
184 187 let _ = event_tx.send(ForgeEvent::ConformComplete { staging });
185 188 }
189 + // Cooperative cancel mid-conform: clean stop, no error surfaced.
190 + Ok(Err(CoreError::Cancelled)) => continue,
186 191 Ok(Err(e)) => {
187 192 let _ = event_tx.send(ForgeEvent::Error {
188 193 error: e.to_string(),
M crates/audiofiles-core/src/similarity.rs +9 -6
@@ -177,11 +177,12 @@ pub fn load_features(db: &Database, hash: &str) -> Result<FeatureVector> {
177 177
178 178 /// Find samples similar to the given reference hash, ranked by feature distance (linear scan).
179 179 ///
180 - /// Loads all analysed samples, normalizes features across the dataset, and returns
181 - /// the top `limit` results (excluding the reference itself).
182 - ///
183 - /// O(n) linear scan with per-query normalization. For indexed sub-linear queries,
184 - /// use [`SimilarityIndex`].
180 + /// Brute-force O(n) similarity scan, retained only as the correctness oracle for
181 + /// the indexed path's tests. The production code path is the cached, sub-linear
182 + /// [`SimilarityIndex`] (built once per session, invalidated on library change).
183 + /// This is `#[cfg(test)]`-gated so no caller can accidentally wire the O(n)
184 + /// per-query scan back into the UI.
185 + #[cfg(test)]
185 186 #[instrument(skip_all, fields(hash = %hash))]
186 187 pub fn find_similar(db: &Database, hash: &str, limit: usize) -> Result<Vec<SimilarResult>> {
187 188 let ref_features = load_features(db, hash)?;
@@ -240,7 +241,9 @@ pub fn find_similar(db: &Database, hash: &str, limit: usize) -> Result<Vec<Simil
240 241 Ok(results)
241 242 }
242 243
243 - /// Compute min/max ranges for each feature across the reference and all other samples.
244 + /// Compute min/max ranges for each feature across the reference and all other
245 + /// samples. Only used by the brute-force oracle [`find_similar`].
246 + #[cfg(test)]
244 247 fn compute_ranges(reference: &FeatureVector, others: &[(String, FeatureVector)]) -> NormRanges {
245 248 let mut ranges = NormRanges::default();
246 249