Skip to main content

max / audiofiles

Concurrency/Perf: seal worker spawns, off-thread forge import, bound caches Ultra-fuzz run-6 remediation, Concurrency & Performance axis. CHRONIC #1 (raw thread::spawn leaving stuck state, 3rd consecutive run): - StreamingLease RAII in preview.rs clears the `streaming` flag on EVERY exit of the streaming decode thread -- generation cancel, external stop, normal end, or a panic mid-decode (previously cleared by hand on the three normal paths only; a panic left it stuck true, a preview "playing but silent" with no recovery). - worker_runtime::spawn_detached: sanctioned panic-isolated spawn for detached threads that restore state via Drop and report no event. Streaming preview decode, file-dialog, and OAuth listener now route through it; instrument decode keeps its per-request catch_unwind (explicit allow). - clippy.toml disallowed-methods bans bare thread::spawn / Builder::spawn, so a new worker spawn is a build error -- the constructive impossibility the prior "migrate the one known site" fixes lacked. Docstring overclaim corrected. Findings: - analysis backfill: run the whole-library par_iter in a pool capped to MAX_CONCURRENT_ANALYSIS (6) so peak full-file decode RAM is bounded independent of core count; small files still parallelise. - forge import (chop/conform blob copy + DB writes) moves off the GUI thread onto a spawn_job with its own WAL connection; poll_workers drains the outcome on a later frame, so the import never holds the GUI's db.lock(). - search VP-tree caches: at most one index (similarity OR fingerprint) resident at a time -- using one evicts the other; documented ceiling. No result truncation. - forge chop_to_temp: RunDirGuard RAII sweeps the unique run dir (all staged slices) on any early exit including panic, replacing two manual cleanup loops. - classifier job replacement detaches the old job to a reaper thread instead of joining it on the GUI thread. Workspace tests pass (forge poll test switched to a file db to match the worker own-connection pattern); clippy clean with the new spawn lint. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 16:11 UTC
Commit: 6c331d6e60c13d2f3ed8be3c0d0cc15d66757f4a
Parent: 354a271
9 files changed, +302 insertions, -83 deletions
A clippy.toml +11
@@ -0,0 +1,11 @@
1 + # Worker threads must route through worker_runtime's panic-isolated seals
2 + # (spawn_worker / spawn_job / spawn_detached), never a bare thread::spawn. A bare
3 + # spawn is how the "stuck flag on panic" failure recurred across three ultra-fuzz
4 + # runs (loose_files, cleanup, streaming preview decode): each fix migrated the one
5 + # known site and the next run found another. Denying the primitive makes a new
6 + # bare worker spawn a build error; the legitimate spawn points carry an explicit
7 + # #[allow(clippy::disallowed_methods)] so they stay visible in review.
8 + disallowed-methods = [
9 + { path = "std::thread::spawn", reason = "route worker threads through worker_runtime::{spawn_worker, spawn_job, spawn_detached} for panic isolation" },
10 + { path = "std::thread::Builder::spawn", reason = "route worker threads through worker_runtime::{spawn_worker, spawn_job, spawn_detached} for panic isolation" },
11 + ]
M crates/audiofiles-browser/src/backend/direct.rs +153 -43
@@ -55,6 +55,80 @@ enum ForgePending {
55 55 },
56 56 }
57 57
58 + /// GUI-ready result of an off-thread forge import job. The import (blob copy + DB
59 + /// writes) runs on its own WAL connection so it never holds the GUI's `db.lock()`;
60 + /// `poll_workers` drains this and emits the matching `BackendEvent` on a later frame.
61 + enum ForgeImportOutcome {
62 + Chop { slice_count: usize },
63 + Conform {
64 + sample_rate: u32,
65 + bit_depth: u16,
66 + overshoot: Option<super::ForgeOvershoot>,
67 + },
68 + Error { error: String },
69 + }
70 +
71 + /// Open a fresh WAL connection + store and import chop slices. Runs on a job
72 + /// thread (the established per-worker connection pattern) so the GUI's db.lock()
73 + /// is never held during the blob copies + row writes.
74 + fn forge_import_chop(
75 + data_dir: &Path,
76 + root: &Path,
77 + vfs_id: VfsId,
78 + parent_id: Option<NodeId>,
79 + staging: audiofiles_core::forge::ChopStaging,
80 + ) -> ForgeImportOutcome {
81 + let db = match Database::open(data_dir.join("audiofiles.db")) {
82 + Ok(d) => d,
83 + Err(e) => return ForgeImportOutcome::Error { error: e.to_string() },
84 + };
85 + let store = match SampleStore::new(root) {
86 + Ok(s) => s,
87 + Err(e) => return ForgeImportOutcome::Error { error: e.to_string() },
88 + };
89 + match audiofiles_core::forge::import_chop(&store, &db, vfs_id, parent_id, staging) {
90 + Ok(r) => ForgeImportOutcome::Chop { slice_count: r.slice_count },
91 + Err(e) => ForgeImportOutcome::Error { error: e.to_string() },
92 + }
93 + }
94 +
95 + /// Off-thread counterpart for conform import (see [`forge_import_chop`]). Maps the
96 + /// core overshoot result into the GUI-facing `ForgeOvershoot` here so the outcome
97 + /// is ready to emit.
98 + #[allow(clippy::too_many_arguments)]
99 + fn forge_import_conform(
100 + data_dir: &Path,
101 + root: &Path,
102 + vfs_id: VfsId,
103 + parent_id: Option<NodeId>,
104 + sample_rate: u32,
105 + bit_depth: u16,
106 + staging: audiofiles_core::forge::ConformStaging,
107 + ) -> ForgeImportOutcome {
108 + let db = match Database::open(data_dir.join("audiofiles.db")) {
109 + Ok(d) => d,
110 + Err(e) => return ForgeImportOutcome::Error { error: e.to_string() },
111 + };
112 + let store = match SampleStore::new(root) {
113 + Ok(s) => s,
114 + Err(e) => return ForgeImportOutcome::Error { error: e.to_string() },
115 + };
116 + match audiofiles_core::forge::import_conform(&store, &db, vfs_id, parent_id, staging) {
117 + Ok(r) => {
118 + let overshoot = r.overshoot.map(|o| match o.action {
119 + audiofiles_core::forge::OvershootAction::Trimmed { gain_db } => {
120 + super::ForgeOvershoot::Trimmed { peak_dbfs: o.peak_dbfs, gain_db }
121 + }
122 + audiofiles_core::forge::OvershootAction::Flagged => {
123 + super::ForgeOvershoot::Flagged { peak_dbfs: o.peak_dbfs }
124 + }
125 + });
126 + ForgeImportOutcome::Conform { sample_rate, bit_depth, overshoot }
127 + }
128 + Err(e) => ForgeImportOutcome::Error { error: e.to_string() },
129 + }
130 + }
131 +
58 132 /// Direct backend: talks to SQLite and the sample store in-process.
59 133 pub struct DirectBackend {
60 134 db: Mutex<Database>,
@@ -70,8 +144,11 @@ pub struct DirectBackend {
70 144 forge_worker: Mutex<Option<audiofiles_core::forge::ForgeWorkerHandle>>,
71 145 classifier_worker: Mutex<Option<crate::classifier_worker::ClassifierHandle>>,
72 146 // Import context for the in-flight forge op, applied when its CPU stage
73 - // completes (the worker produces temp files; the GUI thread imports them).
147 + // completes (the worker produces temp files; an import job imports them).
74 148 forge_pending: Mutex<Option<ForgePending>>,
149 + // Off-thread forge import (blob copy + DB writes on its own WAL connection),
150 + // so the import never stalls the GUI frame under db.lock().
151 + forge_import_job: Mutex<Option<audiofiles_core::worker_runtime::JobHandle<ForgeImportOutcome>>>,
75 152 // Search worker: owns the similarity / near-duplicate VP-trees and builds +
76 153 // queries them off the GUI thread. Spawned lazily on first search,
77 154 // invalidated on new analysis.
@@ -106,6 +183,7 @@ impl DirectBackend {
106 183 forge_worker: Mutex::new(None),
107 184 classifier_worker: Mutex::new(None),
108 185 forge_pending: Mutex::new(None),
186 + forge_import_job: Mutex::new(None),
109 187 search_worker: Mutex::new(None),
110 188 #[cfg(feature = "device-profiles")]
111 189 plugin_registry: audiofiles_rhai::create_registry().unwrap_or_else(|e| {
@@ -771,8 +849,17 @@ impl Backend for DirectBackend {
771 849
772 850 fn start_classifier_job(&self, job: super::ClassifierJob) -> BackendResult<()> {
773 851 let db_path = self.data_dir.join("audiofiles.db");
774 - // Drop (join) any prior job before starting a new one — the UI gates to one at a time.
775 - *self.classifier_worker.lock() = None;
852 + // Detach any prior job instead of dropping (joining) it here: the one-shot
853 + // classifier job can run for minutes, so joining on the GUI thread would
854 + // freeze the UI. Hand the old handle to a reaper thread that joins it off
855 + // the render thread. The UI gates to one active job, so at most one stale
856 + // job is reaping at a time.
857 + if let Some(old) = self.classifier_worker.lock().take() {
858 + let _ = audiofiles_core::worker_runtime::spawn_detached(
859 + "classifier-reaper",
860 + move || drop(old),
861 + );
862 + }
776 863 let handle = crate::classifier_worker::spawn_classifier_job(db_path, job)
777 864 .map_err(|e| BackendError::Other(format!("failed to spawn classifier worker: {e}")))?;
778 865 *self.classifier_worker.lock() = Some(handle);
@@ -1758,17 +1845,19 @@ impl Backend for DirectBackend {
1758 1845 ForgeEvent::ChopComplete { staging } => {
1759 1846 let pending = self.forge_pending.lock().take();
1760 1847 if let Some(ForgePending::Chop { vfs_id, parent_id }) = pending {
1761 - let db = self.db.lock();
1762 - match audiofiles_core::forge::import_chop(
1763 - &self.store,
1764 - &db,
1765 - vfs_id,
1766 - parent_id,
1767 - staging,
1768 - ) {
1769 - Ok(r) => events.push(BackendEvent::ForgeChopComplete {
1770 - slice_count: r.slice_count,
1771 - }),
1848 + // Import off the GUI thread on its own WAL connection so the
1849 + // blob copies + DB writes never hold the GUI's db.lock().
1850 + let data_dir = self.data_dir.clone();
1851 + let root = self.store.root().to_path_buf();
1852 + let job = audiofiles_core::worker_runtime::spawn_job(
1853 + "forge-import-chop",
1854 + || ForgeImportOutcome::Error {
1855 + error: "forge chop import panicked".to_string(),
1856 + },
1857 + move || forge_import_chop(&data_dir, &root, vfs_id, parent_id, staging),
1858 + );
1859 + match job {
1860 + Ok(handle) => *self.forge_import_job.lock() = Some(handle),
1772 1861 Err(e) => events.push(BackendEvent::ForgeError {
1773 1862 error: e.to_string(),
1774 1863 }),
@@ -1790,34 +1879,23 @@ impl Backend for DirectBackend {
1790 1879 bit_depth,
1791 1880 }) = pending
1792 1881 {
1793 - let db = self.db.lock();
1794 - match audiofiles_core::forge::import_conform(
1795 - &self.store,
1796 - &db,
1797 - vfs_id,
1798 - parent_id,
1799 - staging,
1800 - ) {
1801 - Ok(r) => {
1802 - let overshoot = r.overshoot.map(|o| match o.action {
1803 - audiofiles_core::forge::OvershootAction::Trimmed { gain_db } => {
1804 - super::ForgeOvershoot::Trimmed {
1805 - peak_dbfs: o.peak_dbfs,
1806 - gain_db,
1807 - }
1808 - }
1809 - audiofiles_core::forge::OvershootAction::Flagged => {
1810 - super::ForgeOvershoot::Flagged {
1811 - peak_dbfs: o.peak_dbfs,
1812 - }
1813 - }
1814 - });
1815 - events.push(BackendEvent::ForgeConformComplete {
1816 - sample_rate,
1817 - bit_depth,
1818 - overshoot,
1819 - });
1820 - }
1882 + // Import off the GUI thread (see ChopComplete).
1883 + let data_dir = self.data_dir.clone();
1884 + let root = self.store.root().to_path_buf();
1885 + let job = audiofiles_core::worker_runtime::spawn_job(
1886 + "forge-import-conform",
1887 + || ForgeImportOutcome::Error {
1888 + error: "forge conform import panicked".to_string(),
1889 + },
1890 + move || {
1891 + forge_import_conform(
1892 + &data_dir, &root, vfs_id, parent_id, sample_rate, bit_depth,
1893 + staging,
1894 + )
1895 + },
1896 + );
1897 + match job {
1898 + Ok(handle) => *self.forge_import_job.lock() = Some(handle),
1821 1899 Err(e) => events.push(BackendEvent::ForgeError {
1822 1900 error: e.to_string(),
1823 1901 }),
@@ -1832,6 +1910,35 @@ impl Backend for DirectBackend {
1832 1910 }
1833 1911 }
1834 1912
1913 + // Drain the off-thread forge import job (chop/conform slices imported on
1914 + // their own connection), emitting its completion on a later frame.
1915 + let import_outcome = {
1916 + let guard = self.forge_import_job.lock();
1917 + let outcome = guard.as_ref().and_then(|h| h.try_recv());
1918 + drop(guard);
1919 + if outcome.is_some() {
1920 + *self.forge_import_job.lock() = None;
1921 + }
1922 + outcome
1923 + };
1924 + if let Some(outcome) = import_outcome {
1925 + match outcome {
1926 + ForgeImportOutcome::Chop { slice_count } => {
1927 + events.push(BackendEvent::ForgeChopComplete { slice_count })
1928 + }
1929 + ForgeImportOutcome::Conform { sample_rate, bit_depth, overshoot } => {
1930 + events.push(BackendEvent::ForgeConformComplete {
1931 + sample_rate,
1932 + bit_depth,
1933 + overshoot,
1934 + })
1935 + }
1936 + ForgeImportOutcome::Error { error } => {
1937 + events.push(BackendEvent::ForgeError { error })
1938 + }
1939 + }
1940 + }
1941 +
1835 1942 // Poll search worker (similarity / near-duplicate VP-tree build + query).
1836 1943 let search_events: Vec<SearchEvent> = match *self.search_worker.lock() {
1837 1944 Some(ref worker) => {
@@ -2109,7 +2216,10 @@ mod tests {
2109 2216 fn start_chop_runs_on_worker_and_imports_on_poll() {
2110 2217 use audiofiles_core::forge::ChopMethod;
2111 2218 let dir = tempfile::TempDir::new().unwrap();
2112 - let db = Database::open_in_memory().unwrap();
2219 + // File-based db: the off-thread forge import opens its own connection from
2220 + // data_dir/audiofiles.db (the same pattern every worker uses), so the test
2221 + // backend must share that file rather than an in-memory db.
2222 + let db = Database::open(dir.path().join("audiofiles.db")).unwrap();
2113 2223 let store = SampleStore::new(dir.path().join("store")).unwrap();
2114 2224 let backend = DirectBackend::new(db, store, dir.path().to_path_buf());
2115 2225
M crates/audiofiles-browser/src/backend/search_worker.rs +13
@@ -43,6 +43,15 @@ pub type SearchWorkerHandle = WorkerHandle<SearchCommand, SearchEvent>;
43 43
44 44 /// Persistent worker state: the read-only DB connection and the two cached
45 45 /// VP-tree indexes, reused across queries and dropped on `Invalidate`.
46 + ///
47 + /// Eviction policy: each index is bounded by library size (one VP-tree node per
48 + /// analyzed sample — order ~100 MB at a million samples), and **at most one is
49 + /// resident at a time**. A similarity query evicts the fingerprint index and vice
50 + /// versa, so peak index memory is a single tree, not both. The evicted index
51 + /// rebuilds on its next query from fresh data, so results are never truncated —
52 + /// only the cache is. Combined with `Invalidate` (on new analysis), this is the
53 + /// ceiling: ≤ one library-sized tree held, released when the other kind of search
54 + /// runs or when analysis lands.
46 55 struct SearchState {
47 56 db: Database,
48 57 similarity_index: Option<similarity::SimilarityIndex>,
@@ -76,6 +85,8 @@ fn search_step(state: &mut SearchState, cmd: SearchCommand, ctx: &WorkerCtx<Sear
76 85 SearchCommand::FindSimilar { hash, limit } => {
77 86 // A malformed feature row that panics is caught by the runtime and
78 87 // reported as Error; the worker survives for the next query.
88 + // Evict the other index: at most one VP-tree resident at a time.
89 + state.fingerprint_index = None;
79 90 let result: Result<Vec<String>, audiofiles_core::error::CoreError> = (|| {
80 91 if state.similarity_index.is_none() {
81 92 let data = similarity::SimilarityIndex::load_data(&state.db)?;
@@ -96,6 +107,8 @@ fn search_step(state: &mut SearchState, cmd: SearchCommand, ctx: &WorkerCtx<Sear
96 107 });
97 108 }
98 109 SearchCommand::FindNearDuplicates { hash, limit } => {
110 + // Evict the other index: at most one VP-tree resident at a time.
111 + state.similarity_index = None;
99 112 let result: Result<Vec<String>, audiofiles_core::error::CoreError> = (|| {
100 113 if state.fingerprint_index.is_none() {
101 114 let entries = fingerprint::FingerprintIndex::load_data(&state.db)?;
M crates/audiofiles-browser/src/preview.rs +28 -9
@@ -324,18 +324,22 @@ pub fn start_streaming_decode(
324 324 }
325 325
326 326 let shared = std::sync::Arc::clone(shared);
327 - std::thread::spawn(move || {
327 + // Detached + panic-isolated. The StreamingLease clears `streaming` on EVERY
328 + // exit — generation cancel, external stop, normal completion, or a panic in
329 + // the decode/convert path — instead of by-hand on the three normal paths
330 + // (which left the flag stuck true on panic: a preview "playing but silent"
331 + // with no recovery, ultra-fuzz run-6 CHRONIC #1).
332 + let _ = audiofiles_core::worker_runtime::spawn_detached("af-preview-stream", move || {
333 + let _lease = StreamingLease { shared: std::sync::Arc::clone(&shared) };
328 334 let mut total_frames = 0usize;
329 335 let mut started = false;
330 336 // Reuse SampleBuffer across packets to avoid per-packet allocation.
331 337 let mut sample_buf: Option<SampleBuffer<f32>> = None;
332 338
333 339 loop {
334 - // Check if a newer decode has started — if so, this thread exits
340 + // Check if a newer decode has started — if so, this thread exits.
335 341 if shared.decode_generation.load(Ordering::Acquire) != my_generation {
336 - let mut guard = shared.preview.lock();
337 - guard.streaming = false;
338 - return;
342 + return; // lease drop clears `streaming`
339 343 }
340 344
341 345 let packet = match format.next_packet() {
@@ -385,8 +389,7 @@ pub fn start_streaming_decode(
385 389 // Check cancellation: if playing was set to false externally (stop_preview),
386 390 // abort the decode.
387 391 if started && !guard.playing {
388 - guard.streaming = false;
389 - return;
392 + return; // lease drop clears `streaming`
390 393 }
391 394
392 395 if let Some(ref mut buf) = guard.buffer {
@@ -402,9 +405,8 @@ pub fn start_streaming_decode(
402 405 }
403 406 }
404 407
405 - // Decode complete
408 + // Decode complete (lease clears `streaming` when the closure returns).
406 409 let mut guard = shared.preview.lock();
407 - guard.streaming = false;
408 410 guard.decoded_frames = total_frames;
409 411 // If the file was very short and we never hit the prefill threshold, start now
410 412 if !started && total_frames > 0 {
@@ -415,6 +417,19 @@ pub fn start_streaming_decode(
415 417 Ok(())
416 418 }
417 419
420 + /// Clears the shared `streaming` flag when dropped, so every exit path of the
421 + /// streaming decode thread — normal completion, generation cancel, external stop,
422 + /// or a panic mid-decode — leaves `streaming = false`. Previously the flag was
423 + /// cleared by hand on the three normal paths only; a panic left it stuck true.
424 + struct StreamingLease {
425 + shared: std::sync::Arc<crate::state::SharedState>,
426 + }
427 + impl Drop for StreamingLease {
428 + fn drop(&mut self) {
429 + self.shared.preview.lock().streaming = false;
430 + }
431 + }
432 +
418 433 // ---------- Off-thread decode for instrument loading ----------
419 434
420 435 use std::path::PathBuf;
@@ -471,6 +486,10 @@ impl InstrumentDecodeWorker {
471 486 pub fn new() -> Self {
472 487 let (req_tx, req_rx) = std::sync::mpsc::channel::<DecodeRequest>();
473 488 let (out_tx, out_rx) = std::sync::mpsc::channel::<DecodeOutcome>();
489 + // Sanctioned isolated spawn: each request is wrapped in catch_unwind below,
490 + // so a corrupt file surfaces as a DecodeOutcome error rather than killing
491 + // the loop. (A custom request/outcome+generation shape, not the generic seal.)
492 + #[allow(clippy::disallowed_methods)]
474 493 let handle = std::thread::Builder::new()
475 494 .name("af-instrument-decode".into())
476 495 .spawn(move || {
M crates/audiofiles-browser/src/ui/dialog.rs +3 -6
@@ -70,12 +70,9 @@ impl DialogManager {
70 70 return;
71 71 }
72 72 let (tx, rx) = mpsc::channel();
73 - std::thread::Builder::new()
74 - .name("file-dialog".to_string())
75 - .spawn(move || {
76 - let _ = tx.send(pollster::block_on(run_dialog(kind)));
77 - })
78 - .ok();
73 + let _ = audiofiles_core::worker_runtime::spawn_detached("file-dialog", move || {
74 + let _ = tx.send(pollster::block_on(run_dialog(kind)));
75 + });
79 76 *slot = Some(Pending { rx, handler });
80 77 }
81 78
M crates/audiofiles-core/src/analysis/worker.rs +28 -5
@@ -17,6 +17,11 @@ use super::{analyze_sample, AnalysisResult};
17 17 use crate::worker_runtime::{spawn_worker as spawn_runtime_worker, WorkerCtx, WorkerHandle as RuntimeHandle};
18 18 use tracing::instrument;
19 19
20 + /// Upper bound on concurrent full-file decodes during a backfill batch. Caps peak
21 + /// decode memory independent of core count (each decode owns a whole-file mono
22 + /// buffer); small files still parallelise well since their decode is brief.
23 + const MAX_CONCURRENT_ANALYSIS: usize = 6;
24 +
20 25 /// Command sent from the GUI thread to the analysis worker.
21 26 pub enum WorkerCommand {
22 27 /// Analyze a batch of samples. Each tuple: (hash, file_extension, path_on_disk).
@@ -113,9 +118,21 @@ fn analysis_step(_state: &mut (), cmd: WorkerCommand, ctx: &WorkerCtx<WorkerEven
113 118 let total = samples.len();
114 119 let completed = Arc::new(AtomicUsize::new(0));
115 120
116 - // Process samples in parallel using rayon. The shared `&WorkerCtx` is Sync,
117 - // so each rayon task can emit events and read the cancel flag.
118 - samples.par_iter().for_each(|(hash, _ext, path)| {
121 + // Each task decodes a whole file to an owned mono buffer (capped per file at
122 + // ~660 MB). On the global rayon pool that is num_cpus concurrent decodes, so
123 + // peak RAM scaled with core count — a 16-core machine could spike to many GB
124 + // backfilling a library of large files. Run the batch in a pool capped to a
125 + // small fixed width so peak in-flight decode memory is bounded *independent of
126 + // core count*; small files (the common case) still parallelise fine since
127 + // their decode is fast. Fall back to the global pool if the pool fails to build.
128 + let width = std::thread::available_parallelism()
129 + .map(|n| n.get())
130 + .unwrap_or(MAX_CONCURRENT_ANALYSIS)
131 + .min(MAX_CONCURRENT_ANALYSIS);
132 + let run = || {
133 + // Process samples in parallel using rayon. The shared `&WorkerCtx` is Sync,
134 + // so each rayon task can emit events and read the cancel flag.
135 + samples.par_iter().for_each(|(hash, _ext, path)| {
119 136 // Check cancel flag before starting each sample.
120 137 if ctx.is_cancelled() {
121 138 return;
@@ -167,8 +184,14 @@ fn analysis_step(_state: &mut (), cmd: WorkerCommand, ctx: &WorkerCtx<WorkerEven
167 184 }
168 185 }
169 186
170 - completed.fetch_add(1, Ordering::Release);
171 - });
187 + completed.fetch_add(1, Ordering::Release);
188 + });
189 + };
190 +
191 + match rayon::ThreadPoolBuilder::new().num_threads(width).build() {
192 + Ok(pool) => pool.install(run),
193 + Err(_) => run(), // fall back to the global pool
194 + }
172 195
173 196 ctx.emit(WorkerEvent::BatchComplete);
174 197 }
M crates/audiofiles-core/src/forge/runner.rs +24 -11
@@ -103,6 +103,12 @@ pub fn chop_to_temp(
103 103 let temp_dir = forge_temp_dir()?;
104 104 let width = slice_index_width(slices.len());
105 105
106 + // RAII cleanup: the run dir is unique, so removing it on any early exit —
107 + // cancel, encode error, empty result, OR a panic in render_slice/encode_wav —
108 + // sweeps every staged slice. Disarmed only once a complete ChopStaging is
109 + // about to be returned; from then the import stage owns and removes the dir.
110 + let mut run_dir_guard = RunDirGuard { dir: temp_dir.clone(), armed: true };
111 +
106 112 // Number sequentially over slices actually written, so names carry no gaps
107 113 // even if a slice renders empty and the count never overstates the result.
108 114 let mut staged: Vec<(String, PathBuf)> = Vec::new();
@@ -110,10 +116,7 @@ pub fn chop_to_temp(
110 116 // Honor cancellation between slices: a chop of a long file into hundreds
111 117 // of slices must stop promptly, not encode every slice then discard.
112 118 if cancel.load(Ordering::Acquire) {
113 - for (_, p) in &staged {
114 - let _ = std::fs::remove_file(p);
115 - }
116 - return Err(CoreError::Cancelled);
119 + return Err(CoreError::Cancelled); // guard sweeps staged slices
117 120 }
118 121 let buf = render_slice(&decoded.samples, decoded.channels, slice);
119 122 if buf.is_empty() {
@@ -126,22 +129,32 @@ pub fn chop_to_temp(
126 129 };
127 130 let slice_name = format!("{stem}_{:0width$}.wav", staged.len() + 1, width = width);
128 131 let temp_path = temp_dir.join(&slice_name);
129 - if let Err(e) = encode_wav(&converted, 24, &temp_path) {
130 - // Clean up any temp files already staged before bailing.
131 - for (_, p) in &staged {
132 - let _ = std::fs::remove_file(p);
133 - }
134 - return Err(e);
135 - }
132 + encode_wav(&converted, 24, &temp_path)?; // guard sweeps on error
136 133 staged.push((slice_name, temp_path));
137 134 }
138 135
139 136 if staged.is_empty() {
140 137 return Err(CoreError::Internal("chop produced no slices".to_string()));
141 138 }
139 + run_dir_guard.armed = false; // staging complete; import stage owns the dir now
142 140 Ok(ChopStaging { stem, slices: staged })
143 141 }
144 142
143 + /// Removes the unique per-run forge temp dir (and every slice staged into it) on
144 + /// drop unless disarmed. Guards `chop_to_temp` against leaking staged WAVs when it
145 + /// bails or panics before handing a complete `ChopStaging` to the import stage.
146 + struct RunDirGuard {
147 + dir: PathBuf,
148 + armed: bool,
149 + }
150 + impl Drop for RunDirGuard {
151 + fn drop(&mut self) {
152 + if self.armed {
153 + let _ = std::fs::remove_dir_all(&self.dir);
154 + }
155 + }
156 + }
157 +
145 158 /// Import stage of chop: create the `"{stem}_slices"` directory and import each
146 159 /// staged temp WAV into the store/VFS. Runs on the GUI thread (fast DB writes;
147 160 /// the slow decode/encode already happened in [`chop_to_temp`]). Consumes the
M crates/audiofiles-core/src/worker_runtime.rs +38 -7
@@ -228,6 +228,7 @@ where
228 228 let cancel = Arc::new(AtomicBool::new(false));
229 229 let cancel_worker = Arc::clone(&cancel);
230 230
231 + #[allow(clippy::disallowed_methods)] // sanctioned spawn point (per-step catch_unwind)
231 232 let thread = thread::Builder::new()
232 233 .name(name.to_string())
233 234 .spawn(move || {
@@ -296,13 +297,12 @@ impl<Ev> Drop for JobHandle<Ev> {
296 297 /// isolation guarantee — so the GUI's `try_recv` always resolves a terminal event
297 298 /// instead of hanging on a silently-dead thread. Use this for any
298 299 /// run-once-then-exit thread (classifier jobs, one-shot library passes); use
299 - /// `spawn_worker` for a long-lived command loop. Two bare `thread::spawn` sites
300 - /// remain by design, both different shapes from a compute job: the instrument
301 - /// decode worker (its own per-request `catch_unwind`), and two recoverable
302 - /// threads — a generation-cancelled streaming preview decode (a panic just stops
303 - /// that one preview; the next play spawns a fresh generation) and an OAuth
304 - /// loopback listener (bounded by a timeout). Those recover by replacement rather
305 - /// than by emitting a terminal event, so the job seal does not apply.
300 + /// `spawn_worker` for a long-lived command loop, and [`spawn_detached`] for a
301 + /// thread that restores its own state via Drop and reports no event. Bare
302 + /// `thread::spawn` / `Builder::spawn` is denied by clippy (`disallowed-methods`):
303 + /// every worker thread must route through one of these three panic-isolated
304 + /// seals, so a new recv/stream loop cannot reintroduce the silent-stuck-state
305 + /// failure that recurred across three ultra-fuzz runs.
306 306 pub fn spawn_job<Ev, F>(
307 307 name: &str,
308 308 on_panic: impl FnOnce() -> Ev + Send + 'static,
@@ -313,6 +313,7 @@ where
313 313 F: FnOnce() -> Ev + Send + 'static,
314 314 {
315 315 let (tx, rx) = mpsc::channel::<Ev>();
316 + #[allow(clippy::disallowed_methods)] // sanctioned spawn point (panic-isolated)
316 317 let thread = thread::Builder::new()
317 318 .name(name.to_string())
318 319 .spawn(move || {
@@ -326,6 +327,36 @@ where
326 327 })
327 328 }
328 329
330 + /// Spawn a detached thread whose `body` restores its own invariants on exit
331 + /// (typically via an RAII guard whose `Drop` resets shared state) and does NOT
332 + /// report a terminal event.
333 + ///
334 + /// `body` runs inside `catch_unwind`, so a panic unwinds it — running its
335 + /// destructors, which is where the state restoration happens — and is then
336 + /// logged instead of silently killing the thread with state left wedged. This is
337 + /// the sanctioned spawn for the recoverable detached threads that recover by
338 + /// replacement rather than by emitting an event (the streaming preview decode,
339 + /// whose `StreamingLease::drop` clears the `streaming` flag on every exit path
340 + /// including panic). `spawn_job`/`spawn_worker` cover everything that reports a
341 + /// result. A bare `thread::spawn` is denied by clippy precisely so this seal
342 + /// (Drop + catch_unwind) can't be bypassed by a new recv/stream loop — the exact
343 + /// regression that recurred across three ultra-fuzz runs.
344 + pub fn spawn_detached(
345 + name: &str,
346 + body: impl FnOnce() + Send + 'static,
347 + ) -> std::io::Result<()> {
348 + let label = name.to_string();
349 + #[allow(clippy::disallowed_methods)] // sanctioned spawn point (panic-isolated)
350 + thread::Builder::new()
351 + .name(label.clone())
352 + .spawn(move || {
353 + if std::panic::catch_unwind(std::panic::AssertUnwindSafe(body)).is_err() {
354 + tracing::error!(thread = %label, "detached thread panicked; state restored via Drop");
355 + }
356 + })?;
357 + Ok(())
358 + }
359 +
329 360 #[cfg(test)]
330 361 mod tests {
331 362 use super::*;
M crates/audiofiles-sync/src/auth.rs +4 -2
@@ -67,8 +67,10 @@ pub fn start_auth(client: &SyncKitClient) -> Result<AuthSession> {
67 67 let thread_expected_state = state.clone();
68 68 let expected_state = state;
69 69
70 - // Spawn blocking listener thread
71 - std::thread::spawn(move || {
70 + // Spawn blocking listener thread (panic-isolated detached thread; on panic it
71 + // logs and the oneshot sender drops, so the awaiting caller sees a cancelled
72 + // channel rather than a silently dead listener).
73 + let _ = audiofiles_core::worker_runtime::spawn_detached("oauth-listener", move || {
72 74 listener
73 75 .set_nonblocking(false)
74 76 .ok();