max / audiofiles

Perf: seal worker panic isolation behind worker_runtime Chronic finding (ultra-fuzz runs 1-4): the "every worker wraps its command in catch_unwind" invariant was enforced per call site and drifted — loose_files_worker (added by run 1's own remediation) omitted it, so a panic in relocate/purge silently killed the thread and dropped all future commands. Structural fix: a generic spawn_worker factory (worker_runtime) owns the recv loop and catch_unwinds every command. The Receiver never leaves the factory, so a worker cannot write its own unguarded loop — the drift is unrepresentable. Migrate all 7 recv-loop workers (analysis, edit, forge, import, export, search, loose-files) onto it; delete the per-command catch_unwind band-aids (keeping the legitimate per-sample / per-file inner guards in analysis and import). Cancel flag now lives in the runtime (set on Drop), and import/export drop their command-channel peeking for it. Throttle analysis Progress to cap the event channel under rayon fan-out. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: Max Johnson <me@maxj.phd> · 2026-06-22 00:03 UTC

Commit: f9817b77e01ddb0e9b7fcd3f47990062a43fdfb2

Parent: df95512

9 files changed, +870 insertions, -896 deletions

M crates/audiofiles-browser/src/backend/search_worker.rs +73 -140

			@@ -11,11 +11,10 @@
11	11		//! The indexes are cached across queries and dropped on [`SearchCommand::Invalidate`]
12	12		//! (sent when new analysis lands), so they rebuild from fresh data on the next query.
13	13
14		-	use std::path::{Path, PathBuf};
15		-	use std::sync::{mpsc, Mutex};
16		-	use std::thread;
	14	+	use std::path::PathBuf;
17	15
18	16		use audiofiles_core::db::Database;
	17	+	use audiofiles_core::worker_runtime::{spawn_worker, WorkerCtx, WorkerHandle};
19	18		use audiofiles_core::{fingerprint, similarity};
20	19
21	20		/// Command sent from the GUI thread to the search worker.
			@@ -26,8 +25,6 @@ pub enum SearchCommand {
26	25		FindNearDuplicates { hash: String, limit: usize },
27	26		/// Drop both cached indexes so the next query rebuilds from fresh data.
28	27		Invalidate,
29		-	/// Shut the worker thread down.
30		-	Shutdown,
31	28		}
32	29
33	30		/// Event sent from the search worker back to the GUI thread.
			@@ -40,151 +37,87 @@ pub enum SearchEvent {
40	37		Error { error: String },
41	38		}
42	39
43		-	/// Handle for communicating with the background search worker.
44		-	pub struct SearchWorkerHandle {
45		-	cmd_tx: mpsc::Sender<SearchCommand>,
46		-	event_rx: Mutex<mpsc::Receiver<SearchEvent>>,
47		-	thread: Option<thread::JoinHandle<()>>,
48		-	}
49		-
50		-	impl SearchWorkerHandle {
51		-	/// Poll for the next event without blocking.
52		-	pub fn try_recv(&self) -> Option<SearchEvent> {
53		-	self.event_rx.lock().ok()?.try_recv().ok()
54		-	}
55		-
56		-	/// Send a command to the worker.
57		-	pub fn send(&self, cmd: SearchCommand) {
58		-	let _ = self.cmd_tx.send(cmd);
59		-	}
60		-	}
	40	+	/// Handle for communicating with the background search worker. Search has no
	41	+	/// Cancel command, so the runtime handle is used directly.
	42	+	pub type SearchWorkerHandle = WorkerHandle<SearchCommand, SearchEvent>;
61	43
62		-	impl Drop for SearchWorkerHandle {
63		-	fn drop(&mut self) {
64		-	let _ = self.cmd_tx.send(SearchCommand::Shutdown);
65		-	if let Some(handle) = self.thread.take() {
66		-	let _ = handle.join();
67		-	}
68		-	}
	44	+	/// Persistent worker state: the read-only DB connection and the two cached
	45	+	/// VP-tree indexes, reused across queries and dropped on `Invalidate`.
	46	+	struct SearchState {
	47	+	db: Database,
	48	+	similarity_index: Option<similarity::SimilarityIndex>,
	49	+	fingerprint_index: Option<fingerprint::FingerprintIndex>,
69	50		}
70	51
71		-	/// Spawn the background search worker. It opens its own connection to the
72		-	/// database at `db_path` on first use.
	52	+	/// Spawn the background search worker. It opens its own read-only connection to
	53	+	/// the database at `db_path` on the worker thread.
73	54		pub fn spawn_search_worker(db_path: PathBuf) -> std::io::Result<SearchWorkerHandle> {
74		-	let (cmd_tx, cmd_rx) = mpsc::channel::<SearchCommand>();
75		-	let (event_tx, event_rx) = mpsc::channel::<SearchEvent>();
76		-
77		-	let thread = thread::Builder::new()
78		-	.name("search-worker".to_string())
79		-	.spawn(move \|\| search_worker_loop(db_path, cmd_rx, event_tx))?;
80		-
81		-	Ok(SearchWorkerHandle {
82		-	cmd_tx,
83		-	event_rx: Mutex::new(event_rx),
84		-	thread: Some(thread),
85		-	})
	55	+	spawn_worker(
	56	+	"search-worker",
	57	+	move \|\| Database::open(&db_path).map(\|db\| SearchState {
	58	+	db,
	59	+	similarity_index: None,
	60	+	fingerprint_index: None,
	61	+	}),
	62	+	\|e\| SearchEvent::Error { error: e.to_string() },
	63	+	\|_state\| SearchEvent::Error {
	64	+	error: "search worker panicked (internal error)".to_string(),
	65	+	},
	66	+	search_step,
	67	+	)
86	68		}
87	69
88		-	fn search_worker_loop(
89		-	db_path: PathBuf,
90		-	cmd_rx: mpsc::Receiver<SearchCommand>,
91		-	event_tx: mpsc::Sender<SearchEvent>,
92		-	) {
93		-	let mut similarity_index: Option<similarity::SimilarityIndex> = None;
94		-	let mut fingerprint_index: Option<fingerprint::FingerprintIndex> = None;
95		-	// Opened lazily on the first query and reused.
96		-	let mut db: Option<Database> = None;
97		-
98		-	while let Ok(cmd) = cmd_rx.recv() {
99		-	match cmd {
100		-	SearchCommand::Shutdown => break,
101		-	SearchCommand::Invalidate => {
102		-	similarity_index = None;
103		-	fingerprint_index = None;
104		-	}
105		-	SearchCommand::FindSimilar { hash, limit } => {
106		-	let database = match ensure_db(&mut db, &db_path) {
107		-	Ok(d) => d,
108		-	Err(e) => {
109		-	let _ = event_tx.send(SearchEvent::Error { error: e });
110		-	continue;
111		-	}
112		-	};
113		-	// Isolate panics: a malformed feature row must report an error,
114		-	// not unwind the worker and silently stop all future searches.
115		-	let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
116		-	if similarity_index.is_none() {
117		-	let data = similarity::SimilarityIndex::load_data(database)?;
118		-	similarity_index = Some(similarity::SimilarityIndex::build_from_data(data));
119		-	}
120		-	let built = similarity_index.as_ref().expect("set above");
121		-	let features = similarity::load_features(database, &hash)?;
122		-	let results = built.find_similar(&hash, &features, limit);
123		-	Ok::<_, audiofiles_core::error::CoreError>(
124		-	results.into_iter().map(\|r\| r.hash).collect::<Vec<String>>(),
125		-	)
126		-	}));
127		-	let event = match outcome {
128		-	Ok(Ok(hashes)) => SearchEvent::SimilarResults {
129		-	source: hash,
130		-	hashes,
131		-	},
132		-	Ok(Err(e)) => SearchEvent::Error {
133		-	error: e.to_string(),
134		-	},
135		-	Err(_) => SearchEvent::Error {
136		-	error: "similarity search panicked (internal error)".to_string(),
137		-	},
138		-	};
139		-	let _ = event_tx.send(event);
140		-	}
141		-	SearchCommand::FindNearDuplicates { hash, limit } => {
142		-	let database = match ensure_db(&mut db, &db_path) {
143		-	Ok(d) => d,
144		-	Err(e) => {
145		-	let _ = event_tx.send(SearchEvent::Error { error: e });
146		-	continue;
147		-	}
148		-	};
149		-	let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
150		-	if fingerprint_index.is_none() {
151		-	let entries = fingerprint::FingerprintIndex::load_data(database)?;
152		-	fingerprint_index =
153		-	Some(fingerprint::FingerprintIndex::build_from_data(entries));
154		-	}
155		-	let built = fingerprint_index.as_ref().expect("set above");
156		-	let reference = fingerprint::load_fingerprint(database, &hash)?;
157		-	let results = built.find_near_duplicates(&hash, &reference.envelope, limit);
158		-	Ok::<_, audiofiles_core::error::CoreError>(
159		-	results.into_iter().map(\|r\| r.hash).collect::<Vec<String>>(),
160		-	)
161		-	}));
162		-	let event = match outcome {
163		-	Ok(Ok(hashes)) => SearchEvent::NearDuplicateResults {
164		-	source: hash,
165		-	hashes,
166		-	},
167		-	Ok(Err(e)) => SearchEvent::Error {
168		-	error: e.to_string(),
169		-	},
170		-	Err(_) => SearchEvent::Error {
171		-	error: "duplicate search panicked (internal error)".to_string(),
172		-	},
173		-	};
174		-	let _ = event_tx.send(event);
175		-	}
	70	+	fn search_step(state: &mut SearchState, cmd: SearchCommand, ctx: &WorkerCtx<SearchEvent>) {
	71	+	match cmd {
	72	+	SearchCommand::Invalidate => {
	73	+	state.similarity_index = None;
	74	+	state.fingerprint_index = None;
	75	+	}
	76	+	SearchCommand::FindSimilar { hash, limit } => {
	77	+	// A malformed feature row that panics is caught by the runtime and
	78	+	// reported as Error; the worker survives for the next query.
	79	+	let result: Result<Vec<String>, audiofiles_core::error::CoreError> = (\|\| {
	80	+	if state.similarity_index.is_none() {
	81	+	let data = similarity::SimilarityIndex::load_data(&state.db)?;
	82	+	state.similarity_index =
	83	+	Some(similarity::SimilarityIndex::build_from_data(data));
	84	+	}
	85	+	let built = state.similarity_index.as_ref().expect("set above");
	86	+	let features = similarity::load_features(&state.db, &hash)?;
	87	+	Ok(built
	88	+	.find_similar(&hash, &features, limit)
	89	+	.into_iter()
	90	+	.map(\|r\| r.hash)
	91	+	.collect())
	92	+	})();
	93	+	ctx.emit(match result {
	94	+	Ok(hashes) => SearchEvent::SimilarResults { source: hash, hashes },
	95	+	Err(e) => SearchEvent::Error { error: e.to_string() },
	96	+	});
	97	+	}
	98	+	SearchCommand::FindNearDuplicates { hash, limit } => {
	99	+	let result: Result<Vec<String>, audiofiles_core::error::CoreError> = (\|\| {
	100	+	if state.fingerprint_index.is_none() {
	101	+	let entries = fingerprint::FingerprintIndex::load_data(&state.db)?;
	102	+	state.fingerprint_index =
	103	+	Some(fingerprint::FingerprintIndex::build_from_data(entries));
	104	+	}
	105	+	let built = state.fingerprint_index.as_ref().expect("set above");
	106	+	let reference = fingerprint::load_fingerprint(&state.db, &hash)?;
	107	+	Ok(built
	108	+	.find_near_duplicates(&hash, &reference.envelope, limit)
	109	+	.into_iter()
	110	+	.map(\|r\| r.hash)
	111	+	.collect())
	112	+	})();
	113	+	ctx.emit(match result {
	114	+	Ok(hashes) => SearchEvent::NearDuplicateResults { source: hash, hashes },
	115	+	Err(e) => SearchEvent::Error { error: e.to_string() },
	116	+	});
176	117		}
177	118		}
178	119		}
179	120
180		-	/// Open the worker's own database connection on first use, then reuse it.
181		-	fn ensure_db<'a>(db: &'a mut Option<Database>, db_path: &Path) -> Result<&'a Database, String> {
182		-	if db.is_none() {
183		-	*db = Some(Database::open(db_path).map_err(\|e\| e.to_string())?);
184		-	}
185		-	Ok(db.as_ref().expect("opened above"))
186		-	}
187		-
188	121		#[cfg(test)]
189	122		mod tests {
190	123		use super::*;

M crates/audiofiles-browser/src/export.rs +58 -107

			@@ -4,13 +4,12 @@
4	4		//! communicating via channels. The GUI thread polls events each frame.
5	5
6	6		use std::path::PathBuf;
7		-	use std::sync::{mpsc, Mutex};
8		-	use std::thread;
9	7
10		-	use tracing::{error, instrument};
	8	+	use tracing::instrument;
11	9
12	10		use audiofiles_core::export::{ExportConfig, ExportItem, ExportSummary};
13	11		use audiofiles_core::store::SampleStore;
	12	+	use audiofiles_core::worker_runtime::{spawn_worker, WorkerCtx, WorkerHandle};
14	13
15	14		/// Command sent from the GUI thread to the export worker.
16	15		pub enum ExportCommand {
			@@ -19,10 +18,8 @@ pub enum ExportCommand {
19	18		items: Vec<ExportItem>,
20	19		config: ExportConfig,
21	20		},
22		-	/// Cancel the current export.
	21	+	/// Cancel the current export (sets the worker's cancel flag synchronously).
23	22		Cancel,
24		-	/// Shut down the worker thread.
25		-	Shutdown,
26	23		}
27	24
28	25		/// Event sent from the export worker back to the GUI thread.
			@@ -42,32 +39,22 @@ pub enum ExportEvent {
42	39
43	40		/// Handle for communicating with the background export worker.
44	41		///
45		-	/// The receiver is wrapped in a `Mutex` so `BrowserState` remains `Sync` (required by nih-plug).
46		-	/// Only the GUI thread actually calls `try_recv`, so contention is zero.
47		-	pub struct ExportHandle {
48		-	cmd_tx: mpsc::Sender<ExportCommand>,
49		-	event_rx: Mutex<mpsc::Receiver<ExportEvent>>,
50		-	_thread: Option<thread::JoinHandle<()>>,
51		-	}
	42	+	/// Thin newtype over the generic [`WorkerHandle`]: `send` maps a `Cancel`
	43	+	/// command to a synchronous cancel-flag set, checked between files.
	44	+	pub struct ExportHandle(WorkerHandle<ExportCommand, ExportEvent>);
52	45
53	46		impl ExportHandle {
54	47		/// Poll for the next event without blocking.
55	48		pub fn try_recv(&self) -> Option<ExportEvent> {
56		-	self.event_rx.lock().ok()?.try_recv().ok()
	49	+	self.0.try_recv()
57	50		}
58	51
59	52		/// Send a command to the worker.
60	53		pub fn send(&self, cmd: ExportCommand) {
61		-	let _ = self.cmd_tx.send(cmd);
62		-	}
63		-	}
64		-
65		-	impl Drop for ExportHandle {
66		-	fn drop(&mut self) {
67		-	let _ = self.cmd_tx.send(ExportCommand::Shutdown);
68		-	if let Some(handle) = self._thread.take() {
69		-	let _ = handle.join();
	54	+	if matches!(cmd, ExportCommand::Cancel) {
	55	+	self.0.request_cancel();
70	56		}
	57	+	self.0.send(cmd);
71	58		}
72	59		}
73	60
			@@ -76,94 +63,59 @@ impl Drop for ExportHandle {
76	63		/// The worker opens its own `SampleStore` to avoid Mutex contention with the GUI thread.
77	64		#[instrument(skip_all)]
78	65		pub fn spawn_export_worker(store_root: PathBuf) -> std::io::Result<ExportHandle> {
79		-	let (cmd_tx, cmd_rx) = mpsc::channel::<ExportCommand>();
80		-	let (event_tx, event_rx) = mpsc::channel::<ExportEvent>();
81		-
82		-	let thread = thread::Builder::new()
83		-	.name("export-worker".to_string())
84		-	.spawn(move \|\| {
85		-	worker_loop(cmd_rx, event_tx, &store_root);
86		-	})?;
87		-
88		-	Ok(ExportHandle {
89		-	cmd_tx,
90		-	event_rx: Mutex::new(event_rx),
91		-	_thread: Some(thread),
92		-	})
	66	+	let handle = spawn_worker(
	67	+	"export-worker",
	68	+	move \|\| SampleStore::new(&store_root),
	69	+	\|e\| ExportEvent::Complete {
	70	+	total: 0,
	71	+	errors: vec![("init".to_string(), e.to_string())],
	72	+	},
	73	+	\|_state\| ExportEvent::Complete {
	74	+	total: 0,
	75	+	errors: vec![(
	76	+	"export".to_string(),
	77	+	"export panicked (internal error)".to_string(),
	78	+	)],
	79	+	},
	80	+	export_step,
	81	+	)?;
	82	+	Ok(ExportHandle(handle))
93	83		}
94	84
95		-	#[instrument(skip_all)]
96		-	fn worker_loop(
97		-	cmd_rx: mpsc::Receiver<ExportCommand>,
98		-	event_tx: mpsc::Sender<ExportEvent>,
99		-	store_root: &std::path::Path,
100		-	) {
101		-	let store = match SampleStore::new(store_root) {
102		-	Ok(s) => s,
103		-	Err(e) => {
104		-	let _ = event_tx.send(ExportEvent::Complete {
105		-	total: 0,
106		-	errors: vec![("init".to_string(), e.to_string())],
107		-	});
108		-	error!("Export worker failed to open store: {e}");
109		-	return;
110		-	}
	85	+	fn export_step(store: &mut SampleStore, cmd: ExportCommand, ctx: &WorkerCtx<ExportEvent>) {
	86	+	let ExportCommand::Export { items, config } = cmd else {
	87	+	// Cancel: the flag was already set synchronously by the handle.
	88	+	return;
111	89		};
112	90
113		-	while let Ok(cmd) = cmd_rx.recv() {
114		-	match cmd {
115		-	ExportCommand::Shutdown => break,
116		-	ExportCommand::Cancel => continue,
117		-	ExportCommand::Export { items, config } => {
118		-	let cancelled = std::sync::atomic::AtomicBool::new(false);
119		-
120		-	// Catch panics so a bad file reports an error instead of unwinding
121		-	// the worker thread — that would wedge the UI in the exporting
122		-	// state, since ExportEvent::Complete is the only exit.
123		-	let summary = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
124		-	audiofiles_core::export::run_export(
125		-	&items,
126		-	&config,
127		-	&store,
128		-	\|completed, total, current_name\| {
129		-	// Check for cancel between files
130		-	if let Ok(ExportCommand::Cancel) \| Ok(ExportCommand::Shutdown) =
131		-	cmd_rx.try_recv()
132		-	{
133		-	cancelled.store(true, std::sync::atomic::Ordering::Relaxed);
134		-	return false;
135		-	}
136		-
137		-	let _ = event_tx.send(ExportEvent::Progress {
138		-	completed,
139		-	total,
140		-	current_name: current_name.to_string(),
141		-	});
142		-
143		-	true
144		-	},
145		-	)
146		-	}));
147		-
148		-	match summary {
149		-	Ok(Ok(ExportSummary { total, errors })) => {
150		-	let _ = event_tx.send(ExportEvent::Complete { total, errors });
151		-	}
152		-	Ok(Err(e)) => {
153		-	let _ = event_tx.send(ExportEvent::Complete {
154		-	total: 0,
155		-	errors: vec![("export".to_string(), e.to_string())],
156		-	});
157		-	}
158		-	Err(_panic) => {
159		-	let _ = event_tx.send(ExportEvent::Complete {
160		-	total: 0,
161		-	errors: vec![("export".to_string(), "export panicked (internal error)".to_string())],
162		-	});
163		-	}
164		-	}
	91	+	ctx.reset_cancel();
	92	+	// A panic in run_export is caught by the runtime and reported as a
	93	+	// Complete-with-error, so the UI never wedges in the exporting state.
	94	+	let summary = audiofiles_core::export::run_export(
	95	+	&items,
	96	+	&config,
	97	+	store,
	98	+	\|completed, total, current_name\| {
	99	+	// Cooperative cancel between files (flag set by the handle on Cancel
	100	+	// or by the runtime on Drop).
	101	+	if ctx.is_cancelled() {
	102	+	return false;
165	103		}
166		-	}
	104	+	ctx.emit(ExportEvent::Progress {
	105	+	completed,
	106	+	total,
	107	+	current_name: current_name.to_string(),
	108	+	});
	109	+	true
	110	+	},
	111	+	);
	112	+
	113	+	match summary {
	114	+	Ok(ExportSummary { total, errors }) => ctx.emit(ExportEvent::Complete { total, errors }),
	115	+	Err(e) => ctx.emit(ExportEvent::Complete {
	116	+	total: 0,
	117	+	errors: vec![("export".to_string(), e.to_string())],
	118	+	}),
167	119		}
168	120		}
169	121
			@@ -191,7 +143,6 @@ mod tests {
191	143		},
192	144		};
193	145		let _cancel = ExportCommand::Cancel;
194		-	let _shutdown = ExportCommand::Shutdown;
195	146		}
196	147
197	148		#[test]

M crates/audiofiles-browser/src/import.rs +152 -185

			@@ -7,16 +7,16 @@
7	7		use std::collections::HashMap;
8	8		use std::fs;
9	9		use std::path::{Path, PathBuf};
10		-	use std::sync::{mpsc, Mutex};
11		-	use std::thread;
	10	+	use std::sync::atomic::{AtomicBool, Ordering};
12	11		use std::time::{Duration, Instant};
13	12
14		-	use tracing::{error, instrument, warn};
	13	+	use tracing::{instrument, warn};
15	14
16	15		use audiofiles_core::db::Database;
17	16		use audiofiles_core::error::CoreError;
18	17		use audiofiles_core::store::SampleStore;
19	18		use audiofiles_core::vfs::{self, NodeType};
	19	+	use audiofiles_core::worker_runtime::{spawn_worker, EventSink, WorkerCtx, WorkerHandle};
20	20		use audiofiles_core::{NodeId, VfsId};
21	21
22	22		/// Check whether a path has an audio file extension.
			@@ -68,10 +68,8 @@ pub enum ImportCommand {
68	68		source: PathBuf,
69	69		strategy: ImportStrategy,
70	70		},
71		-	/// Cancel the current import.
	71	+	/// Cancel the current import (sets the worker's cancel flag synchronously).
72	72		Cancel,
73		-	/// Shut down the worker thread.
74		-	Shutdown,
75	73		}
76	74
77	75		/// Event sent from the import worker back to the GUI thread.
			@@ -104,33 +102,30 @@ pub enum ImportEvent {
104	102
105	103		/// Handle for communicating with the background import worker.
106	104		///
107		-	/// The receiver is wrapped in a `Mutex` so `BrowserState` remains `Sync` (required by nih-plug).
108		-	/// Only the GUI thread actually calls `try_recv`, so contention is zero.
109		-	pub struct ImportHandle {
110		-	cmd_tx: mpsc::Sender<ImportCommand>,
111		-	event_rx: Mutex<mpsc::Receiver<ImportEvent>>,
112		-	_thread: Option<thread::JoinHandle<()>>,
113		-	}
	105	+	/// Thin newtype over the generic [`WorkerHandle`]: `send` maps a `Cancel`
	106	+	/// command to a synchronous cancel-flag set, checked between files and during
	107	+	/// the pre-walk.
	108	+	pub struct ImportHandle(WorkerHandle<ImportCommand, ImportEvent>);
114	109
115	110		impl ImportHandle {
116	111		/// Poll for the next event without blocking.
117	112		pub fn try_recv(&self) -> Option<ImportEvent> {
118		-	self.event_rx.lock().ok()?.try_recv().ok()
	113	+	self.0.try_recv()
119	114		}
120	115
121	116		/// Send a command to the worker.
122	117		pub fn send(&self, cmd: ImportCommand) {
123		-	let _ = self.cmd_tx.send(cmd);
	118	+	if matches!(cmd, ImportCommand::Cancel) {
	119	+	self.0.request_cancel();
	120	+	}
	121	+	self.0.send(cmd);
124	122		}
125	123		}
126	124
127		-	impl Drop for ImportHandle {
128		-	fn drop(&mut self) {
129		-	let _ = self.cmd_tx.send(ImportCommand::Shutdown);
130		-	if let Some(handle) = self._thread.take() {
131		-	let _ = handle.join();
132		-	}
133		-	}
	125	+	/// Per-worker state: the worker's own DB connection and store.
	126	+	struct ImportWorker {
	127	+	db: Database,
	128	+	store: SampleStore,
134	129		}
135	130
136	131		/// Spawn the background import worker thread.
			@@ -139,20 +134,33 @@ impl Drop for ImportHandle {
139	134		/// with the GUI thread. Returns a handle for sending commands and polling events.
140	135		#[instrument(skip_all)]
141	136		pub fn spawn_import_worker(db_path: PathBuf, store_root: PathBuf) -> std::io::Result<ImportHandle> {
142		-	let (cmd_tx, cmd_rx) = mpsc::channel::<ImportCommand>();
143		-	let (event_tx, event_rx) = mpsc::channel::<ImportEvent>();
144		-
145		-	let thread = thread::Builder::new()
146		-	.name("import-worker".to_string())
147		-	.spawn(move \|\| {
148		-	worker_loop(cmd_rx, event_tx, &db_path, &store_root);
149		-	})?;
150		-
151		-	Ok(ImportHandle {
152		-	cmd_tx,
153		-	event_rx: Mutex::new(event_rx),
154		-	_thread: Some(thread),
155		-	})
	137	+	let handle = spawn_worker(
	138	+	"import-worker",
	139	+	move \|\| -> Result<ImportWorker, CoreError> {
	140	+	let db = Database::open(&db_path)?;
	141	+	let store = SampleStore::new(&store_root)?;
	142	+	Ok(ImportWorker { db, store })
	143	+	},
	144	+	\|e\| {
	145	+	tracing::error!("Import worker failed to open DB/store: {e}");
	146	+	ImportEvent::Complete {
	147	+	imported: Vec::new(),
	148	+	total_files: 0,
	149	+	errors: 1,
	150	+	duplicates: 0,
	151	+	folders: Vec::new(),
	152	+	}
	153	+	},
	154	+	\|_state\| ImportEvent::Complete {
	155	+	imported: Vec::new(),
	156	+	total_files: 0,
	157	+	errors: 1,
	158	+	duplicates: 0,
	159	+	folders: Vec::new(),
	160	+	},
	161	+	import_step,
	162	+	)?;
	163	+	Ok(ImportHandle(handle))
156	164		}
157	165
158	166		/// Recursively count audio files and sum their sizes under `dir`.
			@@ -160,8 +168,8 @@ pub fn spawn_import_worker(db_path: PathBuf, store_root: PathBuf) -> std::io::Re
160	168		#[instrument(skip_all)]
161	169		fn count_audio_files(
162	170		dir: &Path,
163		-	cmd_rx: &mpsc::Receiver<ImportCommand>,
164		-	event_tx: &mpsc::Sender<ImportEvent>,
	171	+	cancel: &AtomicBool,
	172	+	sink: &EventSink<ImportEvent>,
165	173		) -> Option<(usize, u64)> {
166	174		let mut count = 0;
167	175		let mut total_bytes = 0u64;
			@@ -171,7 +179,7 @@ fn count_audio_files(
171	179
172	180		while let Some(current) = stack.pop() {
173	181		// Check for cancel
174		-	if let Ok(ImportCommand::Cancel) \| Ok(ImportCommand::Shutdown) = cmd_rx.try_recv() {
	182	+	if cancel.load(Ordering::Acquire) {
175	183		return None;
176	184		}
177	185
			@@ -198,7 +206,7 @@ fn count_audio_files(
198	206		total_bytes += meta.len();
199	207		}
200	208		if last_emit.elapsed() >= emit_interval {
201		-	let _ = event_tx.send(ImportEvent::WalkProgress { count, total_bytes });
	209	+	sink.emit(ImportEvent::WalkProgress { count, total_bytes });
202	210		last_emit = Instant::now();
203	211		}
204	212		}
			@@ -279,8 +287,8 @@ fn import_single_file(
279	287		struct ImportContext<'a> {
280	288		store: &'a SampleStore,
281	289		db: &'a Database,
282		-	event_tx: &'a mpsc::Sender<ImportEvent>,
283		-	cmd_rx: &'a mpsc::Receiver<ImportCommand>,
	290	+	sink: &'a EventSink<ImportEvent>,
	291	+	cancel: &'a AtomicBool,
284	292		completed: &'a mut usize,
285	293		total: usize,
286	294		errors: &'a mut usize,
			@@ -290,17 +298,15 @@ struct ImportContext<'a> {
290	298		}
291	299
292	300		impl ImportContext<'_> {
293		-	/// Check for a cancellation command without blocking.
	301	+	/// Whether cancellation has been requested (flag set by the handle on Cancel
	302	+	/// or by the runtime on Drop).
294	303		fn is_cancelled(&self) -> bool {
295		-	matches!(
296		-	self.cmd_rx.try_recv(),
297		-	Ok(ImportCommand::Cancel) \| Ok(ImportCommand::Shutdown)
298		-	)
	304	+	self.cancel.load(Ordering::Acquire)
299	305		}
300	306
301	307		/// Send a progress event for the current file.
302	308		fn send_progress(&self, name: String) {
303		-	let _ = self.event_tx.send(ImportEvent::Progress {
	309	+	self.sink.emit(ImportEvent::Progress {
304	310		completed: *self.completed,
305	311		total: self.total,
306	312		current_name: name,
			@@ -346,14 +352,17 @@ impl ImportContext<'_> {
346	352		}
347	353		Ok(Err(e)) => {
348	354		*self.errors += 1;
349		-	let _ = self.event_tx.send(ImportEvent::FileError {
	355	+	self.sink.emit(ImportEvent::FileError {
350	356		path: path.display().to_string(),
351	357		error: e.to_string(),
352	358		});
353	359		}
354	360		Err(_panic) => {
	361	+	// Per-file panic isolation: a single bad file must not abort the
	362	+	// whole import. (The runtime's outer guard backstops anything
	363	+	// outside this per-file loop.)
355	364		*self.errors += 1;
356		-	let _ = self.event_tx.send(ImportEvent::FileError {
	365	+	self.sink.emit(ImportEvent::FileError {
357	366		path: path.display().to_string(),
358	367		error: "import panicked (internal error)".to_string(),
359	368		});
			@@ -563,153 +572,112 @@ fn import_structured(
563	572		(false, folders)
564	573		}
565	574
566		-	fn worker_loop(
567		-	cmd_rx: mpsc::Receiver<ImportCommand>,
568		-	event_tx: mpsc::Sender<ImportEvent>,
569		-	db_path: &Path,
570		-	store_root: &Path,
571		-	) {
572		-	// Open our own DB connection and store
573		-	let db = match Database::open(db_path) {
574		-	Ok(db) => db,
575		-	Err(e) => {
576		-	let _ = event_tx.send(ImportEvent::Complete {
577		-	imported: Vec::new(),
578		-	total_files: 0,
579		-	errors: 1,
580		-	duplicates: 0,
581		-	folders: Vec::new(),
582		-	});
583		-	error!("Import worker failed to open DB: {e}");
584		-	return;
585		-	}
	575	+	fn import_step(worker: &mut ImportWorker, cmd: ImportCommand, ctx: &WorkerCtx<ImportEvent>) {
	576	+	let ImportCommand::ImportDirectory { source, strategy } = cmd else {
	577	+	// Cancel: the flag was already set synchronously by the handle.
	578	+	return;
586	579		};
587	580
588		-	let store = match SampleStore::new(store_root) {
589		-	Ok(s) => s,
590		-	Err(e) => {
591		-	let _ = event_tx.send(ImportEvent::Complete {
	581	+	ctx.reset_cancel();
	582	+	let db = &worker.db;
	583	+	let store = &worker.store;
	584	+	let sink = ctx.sink();
	585	+	let cancel = ctx.cancel_flag();
	586	+
	587	+	// Resolve strategy to concrete (vfs_id, parent_id, flat)
	588	+	let (vfs_id, parent_id, flat) = match strategy {
	589	+	ImportStrategy::Flat { vfs_id, parent_id } => (vfs_id, parent_id, true),
	590	+	ImportStrategy::NewVfs { vfs_name } => match vfs::create_vfs(db, &vfs_name) {
	591	+	Ok(id) => (id, None, false),
	592	+	Err(e) => {
	593	+	sink.emit(ImportEvent::Complete {
	594	+	imported: Vec::new(),
	595	+	total_files: 0,
	596	+	errors: 1,
	597	+	duplicates: 0,
	598	+	folders: Vec::new(),
	599	+	});
	600	+	tracing::error!("Failed to create VFS '{vfs_name}': {e}");
	601	+	return;
	602	+	}
	603	+	},
	604	+	ImportStrategy::MergeIntoVfs { vfs_id, parent_id } => (vfs_id, parent_id, false),
	605	+	};
	606	+
	607	+	// Phase 1: pre-walk to count audio files and sum sizes
	608	+	let (total, total_bytes) = match count_audio_files(&source, cancel, sink) {
	609	+	Some(result) => result,
	610	+	None => {
	611	+	sink.emit(ImportEvent::Complete {
592	612		imported: Vec::new(),
593	613		total_files: 0,
594		-	errors: 1,
	614	+	errors: 0,
595	615		duplicates: 0,
596	616		folders: Vec::new(),
597	617		});
598		-	error!("Import worker failed to open store: {e}");
599	618		return;
600	619		}
601	620		};
602	621
603		-	while let Ok(cmd) = cmd_rx.recv() {
604		-	match cmd {
605		-	ImportCommand::Shutdown => break,
606		-	ImportCommand::Cancel => continue,
607		-	ImportCommand::ImportDirectory { source, strategy } => {
608		-	// Resolve strategy to concrete (vfs_id, parent_id, flat)
609		-	let (vfs_id, parent_id, flat) = match strategy {
610		-	ImportStrategy::Flat { vfs_id, parent_id } => (vfs_id, parent_id, true),
611		-	ImportStrategy::NewVfs { vfs_name } => {
612		-	match vfs::create_vfs(&db, &vfs_name) {
613		-	Ok(id) => (id, None, false),
614		-	Err(e) => {
615		-	let _ = event_tx.send(ImportEvent::Complete {
616		-	imported: Vec::new(),
617		-	total_files: 0,
618		-	errors: 1,
619		-	duplicates: 0,
620		-	folders: Vec::new(),
621		-	});
622		-	error!("Failed to create VFS '{vfs_name}': {e}");
623		-	continue;
624		-	}
625		-	}
626		-	}
627		-	ImportStrategy::MergeIntoVfs { vfs_id, parent_id } => {
628		-	(vfs_id, parent_id, false)
629		-	}
630		-	};
	622	+	sink.emit(ImportEvent::WalkComplete { total, total_bytes });
631	623
632		-	// Phase 1: pre-walk to count audio files and sum sizes
633		-	let (total, total_bytes) = match count_audio_files(&source, &cmd_rx, &event_tx) {
634		-	Some(result) => result,
635		-	None => {
636		-	let _ = event_tx.send(ImportEvent::Complete {
637		-	imported: Vec::new(),
638		-	total_files: 0,
639		-	errors: 0,
640		-	duplicates: 0,
641		-	folders: Vec::new(),
642		-	});
643		-	continue;
644		-	}
645		-	};
646		-
647		-	let _ = event_tx.send(ImportEvent::WalkComplete { total, total_bytes });
648		-
649		-	// Check if loose-files mode is enabled for this vault. Distinguish
650		-	// "not configured" (legitimate default: off) from a real DB error,
651		-	// which must not silently flip import into copy mode unnoticed.
652		-	let loose_files = match db.conn().query_row(
653		-	"SELECT value FROM user_config WHERE key = 'loose_files'",
654		-	[],
655		-	\|row\| row.get::<_, String>(0),
656		-	) {
657		-	Ok(v) => v == "1",
658		-	Err(rusqlite::Error::QueryReturnedNoRows) => false,
659		-	Err(e) => {
660		-	tracing::warn!("Failed to read loose_files config, defaulting to off: {e}");
661		-	false
662		-	}
663		-	};
664		-
665		-	// Phase 2: import files with progress
666		-	let mut completed = 0usize;
667		-	let mut errors = 0usize;
668		-	let mut duplicates = 0usize;
669		-	let mut imported = Vec::new();
670		-
671		-	let mut ctx = ImportContext {
672		-	store: &store,
673		-	db: &db,
674		-	event_tx: &event_tx,
675		-	cmd_rx: &cmd_rx,
676		-	completed: &mut completed,
677		-	total,
678		-	errors: &mut errors,
679		-	duplicates: &mut duplicates,
680		-	imported: &mut imported,
681		-	loose_files,
682		-	};
	624	+	// Check if loose-files mode is enabled for this vault. Distinguish
	625	+	// "not configured" (legitimate default: off) from a real DB error,
	626	+	// which must not silently flip import into copy mode unnoticed.
	627	+	let loose_files = match db.conn().query_row(
	628	+	"SELECT value FROM user_config WHERE key = 'loose_files'",
	629	+	[],
	630	+	\|row\| row.get::<_, String>(0),
	631	+	) {
	632	+	Ok(v) => v == "1",
	633	+	Err(rusqlite::Error::QueryReturnedNoRows) => false,
	634	+	Err(e) => {
	635	+	tracing::warn!("Failed to read loose_files config, defaulting to off: {e}");
	636	+	false
	637	+	}
	638	+	};
683	639
684		-	let (cancelled, folders) = if flat {
685		-	let c = import_directory_flat(
686		-	&source, vfs_id, parent_id, &mut ctx,
687		-	);
688		-	(c, Vec::new())
689		-	} else {
690		-	import_structured(
691		-	&source, vfs_id, parent_id, &mut ctx,
692		-	)
693		-	};
	640	+	// Phase 2: import files with progress
	641	+	let mut completed = 0usize;
	642	+	let mut errors = 0usize;
	643	+	let mut duplicates = 0usize;
	644	+	let mut imported = Vec::new();
	645	+
	646	+	let mut import_ctx = ImportContext {
	647	+	store,
	648	+	db,
	649	+	sink,
	650	+	cancel,
	651	+	completed: &mut completed,
	652	+	total,
	653	+	errors: &mut errors,
	654	+	duplicates: &mut duplicates,
	655	+	imported: &mut imported,
	656	+	loose_files,
	657	+	};
694	658
695		-	let total_files = if cancelled { completed } else { total };
	659	+	let (cancelled, folders) = if flat {
	660	+	let c = import_directory_flat(&source, vfs_id, parent_id, &mut import_ctx);
	661	+	(c, Vec::new())
	662	+	} else {
	663	+	import_structured(&source, vfs_id, parent_id, &mut import_ctx)
	664	+	};
696	665
697		-	// Checkpoint WAL after large import to keep -shm file fresh
698		-	// and avoid stale memory-mapped state on macOS.
699		-	if let Err(e) = db.wal_checkpoint() {
700		-	warn!("WAL checkpoint after import failed: {e}");
701		-	}
	666	+	let total_files = if cancelled { completed } else { total };
702	667
703		-	let _ = event_tx.send(ImportEvent::Complete {
704		-	imported,
705		-	total_files,
706		-	errors,
707		-	duplicates,
708		-	folders,
709		-	});
710		-	}
711		-	}
	668	+	// Checkpoint WAL after large import to keep -shm file fresh
	669	+	// and avoid stale memory-mapped state on macOS.
	670	+	if let Err(e) = db.wal_checkpoint() {
	671	+	warn!("WAL checkpoint after import failed: {e}");
712	672		}
	673	+
	674	+	sink.emit(ImportEvent::Complete {
	675	+	imported,
	676	+	total_files,
	677	+	errors,
	678	+	duplicates,
	679	+	folders,
	680	+	});
713	681		}
714	682
715	683		#[cfg(test)]
			@@ -739,7 +707,6 @@ mod tests {
739	707		},
740	708		};
741	709		let _cancel = ImportCommand::Cancel;
742		-	let _shutdown = ImportCommand::Shutdown;
743	710		}
744	711
745	712		#[test]
			@@ -794,14 +761,14 @@ mod tests {
794	761		let store = SampleStore::new(tmp.path().join("store")).unwrap();
795	762		let vfs_id = vfs::create_vfs(&db, "T").unwrap();
796	763
797		-	let (event_tx, _event_rx) = mpsc::channel();
798		-	let (_cmd_tx, cmd_rx) = mpsc::channel();
	764	+	let (sink, _event_rx) = EventSink::channel();
	765	+	let cancel = AtomicBool::new(false);
799	766		let (mut completed, mut errors, mut dups, mut imported) = (0, 0, 0, Vec::new());
800	767		let mut ctx = ImportContext {
801	768		store: &store,
802	769		db: &db,
803		-	event_tx: &event_tx,
804		-	cmd_rx: &cmd_rx,
	770	+	sink: &sink,
	771	+	cancel: &cancel,
805	772		completed: &mut completed,
806	773		total: 3,
807	774		errors: &mut errors,

M crates/audiofiles-browser/src/loose_files_worker.rs +48 -93

			@@ -4,12 +4,11 @@
4	4		//! a dedicated thread with its own Database, polled each frame via channels.
5	5
6	6		use std::path::PathBuf;
7		-	use std::sync::{mpsc, Mutex};
8		-	use std::thread;
9	7
10		-	use tracing::{error, instrument};
	8	+	use tracing::instrument;
11	9
12	10		use audiofiles_core::db::Database;
	11	+	use audiofiles_core::worker_runtime::{spawn_worker, WorkerCtx, WorkerHandle};
13	12
14	13		/// Command from the GUI thread to the loose-files worker.
15	14		pub enum LooseFilesCommand {
			@@ -19,8 +18,6 @@ pub enum LooseFilesCommand {
19	18		Relocate(PathBuf),
20	19		/// Remove loose-files samples whose source file is missing.
21	20		Purge,
22		-	/// Shut the worker down.
23		-	Shutdown,
24	21		}
25	22
26	23		/// Event from the loose-files worker back to the GUI thread.
			@@ -37,103 +34,61 @@ pub enum LooseFilesEvent {
37	34
38	35		/// Handle for communicating with the loose-files worker.
39	36		///
40		-	/// The receiver is wrapped in a `Mutex` so `BrowserState` stays `Sync` (nih-plug);
41		-	/// only the GUI thread calls `try_recv`, so contention is zero.
42		-	pub struct LooseFilesHandle {
43		-	cmd_tx: mpsc::Sender<LooseFilesCommand>,
44		-	event_rx: Mutex<mpsc::Receiver<LooseFilesEvent>>,
45		-	_thread: Option<thread::JoinHandle<()>>,
46		-	}
47		-
48		-	impl LooseFilesHandle {
49		-	/// Poll for the next event without blocking.
50		-	pub fn try_recv(&self) -> Option<LooseFilesEvent> {
51		-	self.event_rx.lock().ok()?.try_recv().ok()
52		-	}
53		-
54		-	/// Send a command to the worker.
55		-	pub fn send(&self, cmd: LooseFilesCommand) {
56		-	let _ = self.cmd_tx.send(cmd);
57		-	}
58		-	}
59		-
60		-	impl Drop for LooseFilesHandle {
61		-	fn drop(&mut self) {
62		-	let _ = self.cmd_tx.send(LooseFilesCommand::Shutdown);
63		-	if let Some(handle) = self._thread.take() {
64		-	let _ = handle.join();
65		-	}
66		-	}
67		-	}
	37	+	/// Handle for the loose-files worker. No Cancel command, so the runtime handle
	38	+	/// is used directly.
	39	+	///
	40	+	/// Previously this worker hand-rolled its own recv loop and was the one worker
	41	+	/// that omitted `catch_unwind` — a panic in relocate/purge silently killed the
	42	+	/// thread and dropped all future commands. Routing through [`spawn_worker`]
	43	+	/// makes that drift unrepresentable: the runtime owns the loop and the guard.
	44	+	pub type LooseFilesHandle = WorkerHandle<LooseFilesCommand, LooseFilesEvent>;
68	45
69	46		/// Spawn the loose-files worker. It opens its own `Database` to avoid contending
70	47		/// with the GUI connection.
71	48		#[instrument(skip_all)]
72	49		pub fn spawn_loose_files_worker(db_path: PathBuf) -> std::io::Result<LooseFilesHandle> {
73		-	let (cmd_tx, cmd_rx) = mpsc::channel::<LooseFilesCommand>();
74		-	let (event_tx, event_rx) = mpsc::channel::<LooseFilesEvent>();
75		-
76		-	let thread = thread::Builder::new()
77		-	.name("loose-files-worker".to_string())
78		-	.spawn(move \|\| worker_loop(&cmd_rx, &event_tx, &db_path))?;
79		-
80		-	Ok(LooseFilesHandle {
81		-	cmd_tx,
82		-	event_rx: Mutex::new(event_rx),
83		-	_thread: Some(thread),
84		-	})
	50	+	spawn_worker(
	51	+	"loose-files-worker",
	52	+	move \|\| Database::open(&db_path),
	53	+	\|e\| LooseFilesEvent::Failed {
	54	+	message: format!("could not open the library database: {e}"),
	55	+	},
	56	+	\|_state\| LooseFilesEvent::Failed {
	57	+	message: "loose-files maintenance panicked (internal error)".to_string(),
	58	+	},
	59	+	loose_files_step,
	60	+	)
85	61		}
86	62
87		-	#[instrument(skip_all)]
88		-	fn worker_loop(
89		-	cmd_rx: &mpsc::Receiver<LooseFilesCommand>,
90		-	event_tx: &mpsc::Sender<LooseFilesEvent>,
91		-	db_path: &std::path::Path,
92		-	) {
93		-	let db = match Database::open(db_path) {
94		-	Ok(d) => d,
95		-	Err(e) => {
96		-	error!("Loose-files worker failed to open database: {e}");
97		-	let _ = event_tx.send(LooseFilesEvent::Failed {
98		-	message: "could not open the library database".to_string(),
99		-	});
100		-	return;
101		-	}
102		-	};
103		-
104		-	while let Ok(cmd) = cmd_rx.recv() {
105		-	let event = match cmd {
106		-	LooseFilesCommand::Shutdown => break,
107		-	LooseFilesCommand::CheckIntegrity => {
108		-	match audiofiles_core::store::check_loose_files_integrity(&db) {
109		-	Ok((_valid, missing)) => LooseFilesEvent::IntegrityResult { missing },
110		-	Err(e) => LooseFilesEvent::Failed {
111		-	message: e.to_string(),
112		-	},
113		-	}
	63	+	fn loose_files_step(db: &mut Database, cmd: LooseFilesCommand, ctx: &WorkerCtx<LooseFilesEvent>) {
	64	+	let event = match cmd {
	65	+	LooseFilesCommand::CheckIntegrity => {
	66	+	match audiofiles_core::store::check_loose_files_integrity(db) {
	67	+	Ok((_valid, missing)) => LooseFilesEvent::IntegrityResult { missing },
	68	+	Err(e) => LooseFilesEvent::Failed {
	69	+	message: e.to_string(),
	70	+	},
114	71		}
115		-	LooseFilesCommand::Relocate(root) => {
116		-	match audiofiles_core::store::relocate_missing_loose_files(&db, &root) {
117		-	Ok((relocated, still_missing)) => LooseFilesEvent::RelocateResult {
118		-	relocated,
119		-	still_missing,
120		-	},
121		-	Err(e) => LooseFilesEvent::Failed {
122		-	message: e.to_string(),
123		-	},
124		-	}
125		-	}
126		-	LooseFilesCommand::Purge => {
127		-	match audiofiles_core::store::purge_missing_loose_files(&db) {
128		-	Ok(purged) => LooseFilesEvent::PurgeResult { purged },
129		-	Err(e) => LooseFilesEvent::Failed {
130		-	message: e.to_string(),
131		-	},
132		-	}
	72	+	}
	73	+	LooseFilesCommand::Relocate(root) => {
	74	+	match audiofiles_core::store::relocate_missing_loose_files(db, &root) {
	75	+	Ok((relocated, still_missing)) => LooseFilesEvent::RelocateResult {
	76	+	relocated,
	77	+	still_missing,
	78	+	},
	79	+	Err(e) => LooseFilesEvent::Failed {
	80	+	message: e.to_string(),
	81	+	},
133	82		}
134		-	};
135		-	let _ = event_tx.send(event);
136		-	}
	83	+	}
	84	+	LooseFilesCommand::Purge => match audiofiles_core::store::purge_missing_loose_files(db) {
	85	+	Ok(purged) => LooseFilesEvent::PurgeResult { purged },
	86	+	Err(e) => LooseFilesEvent::Failed {
	87	+	message: e.to_string(),
	88	+	},
	89	+	},
	90	+	};
	91	+	ctx.emit(event);
137	92		}
138	93
139	94		#[cfg(test)]

M crates/audiofiles-core/src/analysis/worker.rs +89 -121

			@@ -4,16 +4,17 @@
4	4		//! and emitting [`WorkerEvent`]s back. Batches are processed in parallel using rayon.
5	5		//! An `AtomicBool` cancel flag allows interrupting in-flight parallel work.
6	6
	7	+	use std::convert::Infallible;
7	8		use std::path::PathBuf;
8		-	use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
9		-	use std::sync::{mpsc, Arc, Mutex};
10		-	use std::thread;
	9	+	use std::sync::atomic::{AtomicUsize, Ordering};
	10	+	use std::sync::Arc;
11	11
12	12		use rayon::prelude::*;
13	13
14	14		use super::config::AnalysisConfig;
15	15		use super::suggest::TagSuggestion;
16	16		use super::{analyze_sample, AnalysisResult};
	17	+	use crate::worker_runtime::{spawn_worker as spawn_runtime_worker, WorkerCtx, WorkerHandle as RuntimeHandle};
17	18		use tracing::instrument;
18	19
19	20		/// Command sent from the GUI thread to the analysis worker.
			@@ -25,10 +26,8 @@ pub enum WorkerCommand {
25	26		/// Which analysis stages to run.
26	27		config: AnalysisConfig,
27	28		},
28		-	/// Cancel the current batch.
	29	+	/// Cancel the current batch (sets the worker's cancel flag synchronously).
29	30		Cancel,
30		-	/// Shut down the worker thread.
31		-	Shutdown,
32	31		}
33	32
34	33		/// Event sent from the analysis worker back to the GUI thread.
			@@ -62,38 +61,23 @@ pub enum WorkerEvent {
62	61
63	62		/// Handle for communicating with the background worker.
64	63		///
65		-	/// The receiver is wrapped in a Mutex to satisfy `Sync` bounds required
66		-	/// by nih-plug's editor state (BrowserState must be Send + Sync).
67		-	/// Only the GUI thread actually calls try_recv, so contention is zero.
68		-	pub struct WorkerHandle {
69		-	cmd_tx: mpsc::Sender<WorkerCommand>,
70		-	event_rx: Mutex<mpsc::Receiver<WorkerEvent>>,
71		-	cancel_flag: Arc<AtomicBool>,
72		-	thread: Option<thread::JoinHandle<()>>,
73		-	}
	64	+	/// Thin newtype over the generic [`RuntimeHandle`]: `send` maps a `Cancel`
	65	+	/// command to a synchronous cancel-flag set so an in-flight batch stops at the
	66	+	/// next per-sample checkpoint.
	67	+	pub struct WorkerHandle(RuntimeHandle<WorkerCommand, WorkerEvent>);
74	68
75	69		impl WorkerHandle {
76	70		/// Poll for the next event without blocking. Returns `None` if no events pending.
77	71		pub fn try_recv(&self) -> Option<WorkerEvent> {
78		-	self.event_rx.lock().ok()?.try_recv().ok()
	72	+	self.0.try_recv()
79	73		}
80	74
81	75		/// Send a command to the worker.
82	76		pub fn send(&self, cmd: WorkerCommand) {
83	77		if matches!(cmd, WorkerCommand::Cancel) {
84		-	self.cancel_flag.store(true, Ordering::Release);
85		-	}
86		-	let _ = self.cmd_tx.send(cmd);
87		-	}
88		-	}
89		-
90		-	impl Drop for WorkerHandle {
91		-	fn drop(&mut self) {
92		-	self.cancel_flag.store(true, Ordering::Release);
93		-	let _ = self.cmd_tx.send(WorkerCommand::Shutdown);
94		-	if let Some(handle) = self.thread.take() {
95		-	let _ = handle.join();
	78	+	self.0.request_cancel();
96	79		}
	80	+	self.0.send(cmd);
97	81		}
98	82		}
99	83
			@@ -103,98 +87,90 @@ impl Drop for WorkerHandle {
103	87		/// should call `handle.try_recv()` each frame to poll for events.
104	88		#[instrument(skip_all)]
105	89		pub fn spawn_worker() -> std::io::Result<WorkerHandle> {
106		-	let (cmd_tx, cmd_rx) = mpsc::channel::<WorkerCommand>();
107		-	let (event_tx, event_rx) = mpsc::channel::<WorkerEvent>();
108		-	let cancel_flag = Arc::new(AtomicBool::new(false));
109		-	let cancel_clone = Arc::clone(&cancel_flag);
110		-
111		-	let thread = thread::Builder::new()
112		-	.name("analysis-worker".to_string())
113		-	.spawn(move \|\| {
114		-	worker_loop(cmd_rx, event_tx, cancel_clone);
115		-	})?;
116		-
117		-	Ok(WorkerHandle {
118		-	cmd_tx,
119		-	event_rx: Mutex::new(event_rx),
120		-	cancel_flag,
121		-	thread: Some(thread),
122		-	})
	90	+	let handle = spawn_worker_impl()?;
	91	+	Ok(WorkerHandle(handle))
123	92		}
124	93
125		-	fn worker_loop(
126		-	cmd_rx: mpsc::Receiver<WorkerCommand>,
127		-	event_tx: mpsc::Sender<WorkerEvent>,
128		-	cancel_flag: Arc<AtomicBool>,
129		-	) {
130		-	while let Ok(cmd) = cmd_rx.recv() {
131		-	match cmd {
132		-	WorkerCommand::Shutdown => break,
133		-	WorkerCommand::Cancel => {
134		-	// Cancel flag already set by WorkerHandle::send; nothing else to do
135		-	continue;
136		-	}
137		-	WorkerCommand::AnalyzeBatch { samples, config } => {
138		-	// Reset cancel flag for this batch
139		-	cancel_flag.store(false, Ordering::Release);
140		-
141		-	let total = samples.len();
142		-	let completed = Arc::new(AtomicUsize::new(0));
143		-
144		-	// Process samples in parallel using rayon
145		-	samples.par_iter().for_each(\|(hash, _ext, path)\| {
146		-	// Check cancel flag before starting each sample
147		-	if cancel_flag.load(Ordering::Acquire) {
148		-	return;
149		-	}
150		-
151		-	let name = crate::util::get_filename(path, "unknown");
152		-	let done = completed.load(Ordering::Acquire);
153		-
154		-	let _ = event_tx.send(WorkerEvent::Progress {
155		-	completed: done,
156		-	total,
157		-	current_name: name,
158		-	});
	94	+	fn spawn_worker_impl() -> std::io::Result<RuntimeHandle<WorkerCommand, WorkerEvent>> {
	95	+	spawn_runtime_worker(
	96	+	"analysis-worker",
	97	+	\|\| Ok::<(), Infallible>(()),
	98	+	\|e: Infallible\| match e {},
	99	+	// An outer panic (outside the per-sample guard below) ends the batch; emit
	100	+	// BatchComplete so the import/backfill flow does not wedge waiting for it.
	101	+	\|_state: &()\| WorkerEvent::BatchComplete,
	102	+	analysis_step,
	103	+	)
	104	+	}
159	105
160		-	// Catch panics so a single bad sample doesn't kill the worker
161		-	let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
162		-	analyze_sample(hash, path, &config)
163		-	}));
	106	+	fn analysis_step(_state: &mut (), cmd: WorkerCommand, ctx: &WorkerCtx<WorkerEvent>) {
	107	+	let WorkerCommand::AnalyzeBatch { samples, config } = cmd else {
	108	+	// Cancel: the flag was already set synchronously by the handle.
	109	+	return;
	110	+	};
	111	+
	112	+	ctx.reset_cancel();
	113	+	let total = samples.len();
	114	+	let completed = Arc::new(AtomicUsize::new(0));
	115	+
	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)\| {
	119	+	// Check cancel flag before starting each sample.
	120	+	if ctx.is_cancelled() {
	121	+	return;
	122	+	}
164	123
165		-	match result {
166		-	Ok(Ok(result)) => {
167		-	let suggestions = if config.auto_suggest_tags {
168		-	super::suggest::suggest_tags(&result)
169		-	} else {
170		-	Vec::new()
171		-	};
172		-	let _ = event_tx.send(WorkerEvent::SampleDone {
173		-	result: Box::new(result),
174		-	suggestions,
175		-	});
176		-	}
177		-	Ok(Err(e)) => {
178		-	let _ = event_tx.send(WorkerEvent::SampleError {
179		-	hash: hash.clone(),
180		-	error: e.to_string(),
181		-	});
182		-	}
183		-	Err(_panic) => {
184		-	let _ = event_tx.send(WorkerEvent::SampleError {
185		-	hash: hash.clone(),
186		-	error: "analysis panicked (internal error)".to_string(),
187		-	});
188		-	}
189		-	}
	124	+	let done = completed.load(Ordering::Acquire);
	125	+	// Throttle Progress: it is high-frequency and low-value, and at large
	126	+	// batch sizes one event per sample can flood the channel faster than the
	127	+	// GUI drains it. Emit on the first sample and every 8th thereafter; the
	128	+	// GUI shows a coarse-grained bar regardless.
	129	+	if done.is_multiple_of(8) {
	130	+	ctx.emit(WorkerEvent::Progress {
	131	+	completed: done,
	132	+	total,
	133	+	current_name: crate::util::get_filename(path, "unknown"),
	134	+	});
	135	+	}
190	136
191		-	completed.fetch_add(1, Ordering::Release);
	137	+	// Inner per-sample panic isolation: a single bad sample must not abort
	138	+	// the whole batch. (The runtime's outer guard only catches panics
	139	+	// outside this loop.)
	140	+	let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
	141	+	analyze_sample(hash, path, &config)
	142	+	}));
	143	+
	144	+	match result {
	145	+	Ok(Ok(result)) => {
	146	+	let suggestions = if config.auto_suggest_tags {
	147	+	super::suggest::suggest_tags(&result)
	148	+	} else {
	149	+	Vec::new()
	150	+	};
	151	+	ctx.emit(WorkerEvent::SampleDone {
	152	+	result: Box::new(result),
	153	+	suggestions,
	154	+	});
	155	+	}
	156	+	Ok(Err(e)) => {
	157	+	ctx.emit(WorkerEvent::SampleError {
	158	+	hash: hash.clone(),
	159	+	error: e.to_string(),
	160	+	});
	161	+	}
	162	+	Err(_panic) => {
	163	+	ctx.emit(WorkerEvent::SampleError {
	164	+	hash: hash.clone(),
	165	+	error: "analysis panicked (internal error)".to_string(),
192	166		});
193		-
194		-	let _ = event_tx.send(WorkerEvent::BatchComplete);
195	167		}
196	168		}
197		-	}
	169	+
	170	+	completed.fetch_add(1, Ordering::Release);
	171	+	});
	172	+
	173	+	ctx.emit(WorkerEvent::BatchComplete);
198	174		}
199	175
200	176		#[cfg(test)]
			@@ -209,12 +185,6 @@ mod tests {
209	185		}
210	186
211	187		#[test]
212		-	fn worker_command_shutdown_construction() {
213		-	let cmd = WorkerCommand::Shutdown;
214		-	assert!(matches!(cmd, WorkerCommand::Shutdown));
215		-	}
216		-
217		-	#[test]
218	188		fn worker_command_analyze_batch_construction() {
219	189		let config = AnalysisConfig::default();
220	190		let samples = vec![
			@@ -262,12 +232,10 @@ mod tests {
262	232		}
263	233
264	234		#[test]
265		-	fn cancel_flag_set_on_cancel() {
	235	+	fn cancel_then_drop_does_not_hang() {
266	236		let handle = spawn_worker().unwrap();
267	237		handle.send(WorkerCommand::Cancel);
268		-	// Give it a moment to process
269	238		std::thread::sleep(std::time::Duration::from_millis(10));
270		-	assert!(handle.cancel_flag.load(Ordering::Acquire));
271	239		drop(handle);
272	240		}
273	241		}

M crates/audiofiles-core/src/edit/worker.rs +69 -99

			@@ -3,10 +3,9 @@
3	3		//! Follows the same pattern as `analysis::worker`: a dedicated thread receiving
4	4		//! commands over a channel and emitting events back.
5	5
	6	+	use std::convert::Infallible;
6	7		use std::path::{Path, PathBuf};
7	8		use std::sync::atomic::{AtomicBool, Ordering};
8		-	use std::sync::{mpsc, Arc, Mutex};
9		-	use std::thread;
10	9
11	10		use tracing::instrument;
12	11
			@@ -14,6 +13,7 @@ use super::{apply_edit, EditOperation};
14	13		use crate::export::convert::ConvertedAudio;
15	14		use crate::export::decode::decode_multichannel;
16	15		use crate::export::encode::encode_wav;
	16	+	use crate::worker_runtime::{spawn_worker, WorkerCtx, WorkerHandle};
17	17
18	18		/// Command sent from the GUI thread to the edit worker.
19	19		pub enum EditCommand {
			@@ -24,10 +24,8 @@ pub enum EditCommand {
24	24		path: PathBuf,
25	25		operation: EditOperation,
26	26		},
27		-	/// Cancel the current operation.
	27	+	/// Cancel the current operation (sets the worker's cancel flag synchronously).
28	28		Cancel,
29		-	/// Shut down the worker thread.
30		-	Shutdown,
31	29		}
32	30
33	31		/// Event sent from the edit worker back to the GUI thread.
			@@ -45,58 +43,52 @@ pub enum EditEvent {
45	43		}
46	44
47	45		/// Handle for communicating with the background edit worker.
48		-	pub struct EditWorkerHandle {
49		-	cmd_tx: mpsc::Sender<EditCommand>,
50		-	event_rx: Mutex<mpsc::Receiver<EditEvent>>,
51		-	cancel_flag: Arc<AtomicBool>,
52		-	thread: Option<thread::JoinHandle<()>>,
53		-	}
	46	+	///
	47	+	/// Thin newtype over the generic [`WorkerHandle`]: `send` maps a `Cancel`
	48	+	/// command to a synchronous cancel-flag set so an in-flight edit aborts at its
	49	+	/// next checkpoint, before the queued command is even dequeued.
	50	+	pub struct EditWorkerHandle(WorkerHandle<EditCommand, EditEvent>);
54	51
55	52		impl EditWorkerHandle {
56	53		/// Poll for the next event without blocking.
57	54		pub fn try_recv(&self) -> Option<EditEvent> {
58		-	self.event_rx.lock().ok()?.try_recv().ok()
	55	+	self.0.try_recv()
59	56		}
60	57
61	58		/// Send a command to the worker.
62	59		pub fn send(&self, cmd: EditCommand) {
63	60		if matches!(cmd, EditCommand::Cancel) {
64		-	self.cancel_flag.store(true, Ordering::Release);
	61	+	self.0.request_cancel();
65	62		}
66		-	let _ = self.cmd_tx.send(cmd);
	63	+	self.0.send(cmd);
67	64		}
68	65		}
69	66
70		-	impl Drop for EditWorkerHandle {
71		-	fn drop(&mut self) {
72		-	self.cancel_flag.store(true, Ordering::Release);
73		-	let _ = self.cmd_tx.send(EditCommand::Shutdown);
74		-	if let Some(handle) = self.thread.take() {
75		-	let _ = handle.join();
76		-	}
77		-	}
	67	+	/// Per-worker state: the hash of the edit currently in flight, so a panic mid-
	68	+	/// edit still reports a terminal `Error` for the right sample (un-wedging the UI).
	69	+	#[derive(Default)]
	70	+	struct EditState {
	71	+	current_hash: Option<String>,
78	72		}
79	73
80	74		/// Spawn the background edit worker thread.
81	75		#[instrument(skip_all)]
82	76		pub fn spawn_edit_worker() -> std::io::Result<EditWorkerHandle> {
83		-	let (cmd_tx, cmd_rx) = mpsc::channel::<EditCommand>();
84		-	let (event_tx, event_rx) = mpsc::channel::<EditEvent>();
85		-	let cancel_flag = Arc::new(AtomicBool::new(false));
86		-	let cancel_clone = Arc::clone(&cancel_flag);
87		-
88		-	let thread = thread::Builder::new()
89		-	.name("edit-worker".to_string())
90		-	.spawn(move \|\| {
91		-	edit_worker_loop(cmd_rx, event_tx, cancel_clone);
92		-	})?;
93		-
94		-	Ok(EditWorkerHandle {
95		-	cmd_tx,
96		-	event_rx: Mutex::new(event_rx),
97		-	cancel_flag,
98		-	thread: Some(thread),
99		-	})
	77	+	let handle = spawn_worker(
	78	+	"edit-worker",
	79	+	\|\| {
	80	+	// Clean up stale temp files from prior sessions once on startup.
	81	+	cleanup_edit_temp_dir();
	82	+	Ok::<EditState, Infallible>(EditState::default())
	83	+	},
	84	+	\|e: Infallible\| match e {},
	85	+	\|state: &EditState\| EditEvent::Error {
	86	+	hash: state.current_hash.clone().unwrap_or_default(),
	87	+	error: "edit panicked (internal error)".to_string(),
	88	+	},
	89	+	edit_step,
	90	+	)?;
	91	+	Ok(EditWorkerHandle(handle))
100	92		}
101	93
102	94		/// Clean up leftover temp files from previous edit sessions.
			@@ -112,64 +104,41 @@ fn cleanup_edit_temp_dir() {
112	104		}
113	105		}
114	106
115		-	fn edit_worker_loop(
116		-	cmd_rx: mpsc::Receiver<EditCommand>,
117		-	event_tx: mpsc::Sender<EditEvent>,
118		-	cancel_flag: Arc<AtomicBool>,
119		-	) {
120		-	// Clean up stale temp files from prior sessions
121		-	cleanup_edit_temp_dir();
122		-
123		-	while let Ok(cmd) = cmd_rx.recv() {
124		-	match cmd {
125		-	EditCommand::Shutdown => break,
126		-	EditCommand::Cancel => continue,
127		-	EditCommand::Edit {
128		-	hash,
129		-	ext: _,
130		-	path,
131		-	operation,
132		-	} => {
133		-	cancel_flag.store(false, Ordering::Release);
134		-
135		-	let _ = event_tx.send(EditEvent::Started {
136		-	hash: hash.clone(),
137		-	});
138		-
139		-	if cancel_flag.load(Ordering::Acquire) {
140		-	continue;
141		-	}
142		-
143		-	// Catch panics so a malformed file (bad decode/encode) reports an
144		-	// error instead of unwinding and killing the worker thread —
145		-	// which would silently stop all future edits with no UI signal.
146		-	let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
147		-	process_edit(&path, &operation, &cancel_flag)
148		-	}));
149		-	match outcome {
150		-	Ok(Ok(result_path)) => {
151		-	let _ = event_tx.send(EditEvent::Complete {
152		-	source_hash: hash,
153		-	result_path,
154		-	operation,
155		-	});
156		-	}
157		-	Ok(Err(e)) => {
158		-	let _ = event_tx.send(EditEvent::Error {
159		-	hash,
160		-	error: e.to_string(),
161		-	});
162		-	}
163		-	Err(_panic) => {
164		-	let _ = event_tx.send(EditEvent::Error {
165		-	hash,
166		-	error: "edit panicked (internal error)".to_string(),
167		-	});
168		-	}
169		-	}
170		-	}
171		-	}
	107	+	fn edit_step(state: &mut EditState, cmd: EditCommand, ctx: &WorkerCtx<EditEvent>) {
	108	+	let EditCommand::Edit {
	109	+	hash,
	110	+	ext: _,
	111	+	path,
	112	+	operation,
	113	+	} = cmd
	114	+	else {
	115	+	// Cancel: the flag was already set synchronously by the handle.
	116	+	return;
	117	+	};
	118	+
	119	+	ctx.reset_cancel();
	120	+	state.current_hash = Some(hash.clone());
	121	+	ctx.emit(EditEvent::Started { hash: hash.clone() });
	122	+
	123	+	if ctx.is_cancelled() {
	124	+	state.current_hash = None;
	125	+	return;
	126	+	}
	127	+
	128	+	// A panic in process_edit (bad decode/encode) is caught by the runtime and
	129	+	// reported as Error for `current_hash`, keeping the worker alive.
	130	+	match process_edit(&path, &operation, ctx.cancel_flag()) {
	131	+	Ok(result_path) => ctx.emit(EditEvent::Complete {
	132	+	source_hash: hash,
	133	+	result_path,
	134	+	operation,
	135	+	}),
	136	+	Err(e) => ctx.emit(EditEvent::Error {
	137	+	hash,
	138	+	error: e.to_string(),
	139	+	}),
172	140		}
	141	+	state.current_hash = None;
173	142		}
174	143
175	144		/// Decode → edit → encode to a temporary WAV file.
			@@ -233,7 +202,6 @@ mod tests {
233	202		#[test]
234	203		fn edit_command_variants() {
235	204		let _cancel = EditCommand::Cancel;
236		-	let _shutdown = EditCommand::Shutdown;
237	205		let _edit = EditCommand::Edit {
238	206		hash: "abc".to_string(),
239	207		ext: "wav".to_string(),
			@@ -266,11 +234,13 @@ mod tests {
266	234		}
267	235
268	236		#[test]
269		-	fn cancel_flag_set_on_cancel() {
	237	+	fn cancel_then_edit_does_not_hang() {
	238	+	// Cancel sets the flag synchronously; a subsequent edit still resets the
	239	+	// flag and runs. (The runtime's own tests cover panic isolation + cancel
	240	+	// semantics directly.)
270	241		let handle = spawn_edit_worker().unwrap();
271	242		handle.send(EditCommand::Cancel);
272	243		std::thread::sleep(std::time::Duration::from_millis(10));
273		-	assert!(handle.cancel_flag.load(Ordering::Relaxed));
274	244		drop(handle);
275	245		}
276	246

M crates/audiofiles-core/src/forge/worker.rs +90 -151

			@@ -6,14 +6,13 @@
6	6		//! event arrives. This keeps the slow CPU work — and the DB lock — off the
7	7		//! render thread, so the forge no longer freezes the UI on long files.
8	8
	9	+	use std::convert::Infallible;
9	10		use std::path::PathBuf;
10		-	use std::sync::atomic::{AtomicBool, Ordering};
11		-	use std::sync::{mpsc, Arc, Mutex};
12		-	use std::thread;
13	11
14	12		use tracing::instrument;
15	13
16	14		use crate::error::CoreError;
	15	+	use crate::worker_runtime::{spawn_worker, WorkerCtx, WorkerHandle};
17	16		use super::chop::ChopMethod;
18	17		use super::conform::ConformTarget;
19	18		use super::runner::{chop_to_temp, compute_preview_marks, conform_to_temp, ChopStaging, ConformStaging};
			@@ -39,10 +38,8 @@ pub enum ForgeCommand {
39	38		source_path: PathBuf,
40	39		method: ChopMethod,
41	40		},
42		-	/// Cancel the current operation.
	41	+	/// Cancel the current operation (sets the worker's cancel flag synchronously).
43	42		Cancel,
44		-	/// Shut down the worker thread.
45		-	Shutdown,
46	43		}
47	44
48	45		/// Event sent from the forge worker back to the GUI thread.
			@@ -60,173 +57,117 @@ pub enum ForgeEvent {
60	57		}
61	58
62	59		/// Handle for communicating with the background forge worker.
63		-	pub struct ForgeWorkerHandle {
64		-	cmd_tx: mpsc::Sender<ForgeCommand>,
65		-	event_rx: Mutex<mpsc::Receiver<ForgeEvent>>,
66		-	cancel_flag: Arc<AtomicBool>,
67		-	thread: Option<thread::JoinHandle<()>>,
68		-	}
	60	+	///
	61	+	/// Thin newtype over the generic [`WorkerHandle`]: `send` maps a `Cancel`
	62	+	/// command to a synchronous cancel-flag set so an in-flight chop/conform aborts
	63	+	/// at its next checkpoint.
	64	+	pub struct ForgeWorkerHandle(WorkerHandle<ForgeCommand, ForgeEvent>);
69	65
70	66		impl ForgeWorkerHandle {
71	67		/// Poll for the next event without blocking.
72	68		pub fn try_recv(&self) -> Option<ForgeEvent> {
73		-	self.event_rx.lock().ok()?.try_recv().ok()
	69	+	self.0.try_recv()
74	70		}
75	71
76	72		/// Send a command to the worker.
77	73		pub fn send(&self, cmd: ForgeCommand) {
78	74		if matches!(cmd, ForgeCommand::Cancel) {
79		-	self.cancel_flag.store(true, Ordering::Release);
80		-	}
81		-	let _ = self.cmd_tx.send(cmd);
82		-	}
83		-	}
84		-
85		-	impl Drop for ForgeWorkerHandle {
86		-	fn drop(&mut self) {
87		-	self.cancel_flag.store(true, Ordering::Release);
88		-	let _ = self.cmd_tx.send(ForgeCommand::Shutdown);
89		-	if let Some(handle) = self.thread.take() {
90		-	let _ = handle.join();
	75	+	self.0.request_cancel();
91	76		}
	77	+	self.0.send(cmd);
92	78		}
93	79		}
94	80
95	81		/// Spawn the background forge worker thread.
96	82		#[instrument(skip_all)]
97	83		pub fn spawn_forge_worker() -> std::io::Result<ForgeWorkerHandle> {
98		-	let (cmd_tx, cmd_rx) = mpsc::channel::<ForgeCommand>();
99		-	let (event_tx, event_rx) = mpsc::channel::<ForgeEvent>();
100		-	let cancel_flag = Arc::new(AtomicBool::new(false));
101		-	let cancel_clone = Arc::clone(&cancel_flag);
102		-
103		-	let thread = thread::Builder::new()
104		-	.name("forge-worker".to_string())
105		-	.spawn(move \|\| {
106		-	forge_worker_loop(cmd_rx, event_tx, cancel_clone);
107		-	})?;
108		-
109		-	Ok(ForgeWorkerHandle {
110		-	cmd_tx,
111		-	event_rx: Mutex::new(event_rx),
112		-	cancel_flag,
113		-	thread: Some(thread),
114		-	})
	84	+	let handle = spawn_worker(
	85	+	"forge-worker",
	86	+	\|\| Ok::<(), Infallible>(()),
	87	+	\|e: Infallible\| match e {},
	88	+	\|_state: &()\| ForgeEvent::Error {
	89	+	error: "forge panicked (internal error)".to_string(),
	90	+	},
	91	+	forge_step,
	92	+	)?;
	93	+	Ok(ForgeWorkerHandle(handle))
115	94		}
116	95
117		-	fn forge_worker_loop(
118		-	cmd_rx: mpsc::Receiver<ForgeCommand>,
119		-	event_tx: mpsc::Sender<ForgeEvent>,
120		-	cancel_flag: Arc<AtomicBool>,
121		-	) {
122		-	while let Ok(cmd) = cmd_rx.recv() {
123		-	match cmd {
124		-	ForgeCommand::Shutdown => break,
125		-	ForgeCommand::Cancel => continue,
126		-	ForgeCommand::Chop {
127		-	source_path,
128		-	base_name,
129		-	method,
130		-	} => {
131		-	cancel_flag.store(false, Ordering::Release);
132		-	let _ = event_tx.send(ForgeEvent::Started);
133		-	if cancel_flag.load(Ordering::Acquire) {
134		-	continue;
135		-	}
136		-	// Isolate panics: a bad file must report an error, not unwind
137		-	// the worker thread and silently stop all future forge jobs.
138		-	let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
139		-	chop_to_temp(&source_path, &base_name, &method, &cancel_flag)
140		-	}));
141		-	match outcome {
142		-	Ok(Ok(staging)) => {
143		-	// If cancelled mid-flight, drop the staged temp files
144		-	// rather than handing them to the GUI.
145		-	if cancel_flag.load(Ordering::Acquire) {
146		-	for (_, p) in &staging.slices {
147		-	let _ = std::fs::remove_file(p);
148		-	}
149		-	continue;
150		-	}
151		-	let _ = event_tx.send(ForgeEvent::ChopComplete { staging });
152		-	}
153		-	// Cooperative cancel mid-chop: clean stop, no error surfaced.
154		-	Ok(Err(CoreError::Cancelled)) => continue,
155		-	Ok(Err(e)) => {
156		-	let _ = event_tx.send(ForgeEvent::Error {
157		-	error: e.to_string(),
158		-	});
159		-	}
160		-	Err(_panic) => {
161		-	let _ = event_tx.send(ForgeEvent::Error {
162		-	error: "chop panicked (internal error)".to_string(),
163		-	});
164		-	}
165		-	}
	96	+	fn forge_step(_state: &mut (), cmd: ForgeCommand, ctx: &WorkerCtx<ForgeEvent>) {
	97	+	match cmd {
	98	+	// Cancel: the flag was already set synchronously by the handle.
	99	+	ForgeCommand::Cancel => {}
	100	+	ForgeCommand::Chop {
	101	+	source_path,
	102	+	base_name,
	103	+	method,
	104	+	} => {
	105	+	ctx.reset_cancel();
	106	+	ctx.emit(ForgeEvent::Started);
	107	+	if ctx.is_cancelled() {
	108	+	return;
166	109		}
167		-	ForgeCommand::Conform {
168		-	source_path,
169		-	base_name,
170		-	target,
171		-	auto_trim,
172		-	} => {
173		-	cancel_flag.store(false, Ordering::Release);
174		-	let _ = event_tx.send(ForgeEvent::Started);
175		-	if cancel_flag.load(Ordering::Acquire) {
176		-	continue;
177		-	}
178		-	let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
179		-	conform_to_temp(&source_path, &base_name, &target, auto_trim, &cancel_flag)
180		-	}));
181		-	match outcome {
182		-	Ok(Ok(staging)) => {
183		-	if cancel_flag.load(Ordering::Acquire) {
184		-	let _ = std::fs::remove_file(&staging.temp_path);
185		-	continue;
	110	+	// A panic here is caught by the runtime and reported as Error.
	111	+	match chop_to_temp(&source_path, &base_name, &method, ctx.cancel_flag()) {
	112	+	Ok(staging) => {
	113	+	// If cancelled mid-flight, drop the staged temp files rather
	114	+	// than handing them to the GUI.
	115	+	if ctx.is_cancelled() {
	116	+	for (_, p) in &staging.slices {
	117	+	let _ = std::fs::remove_file(p);
186	118		}
187		-	let _ = event_tx.send(ForgeEvent::ConformComplete { staging });
188		-	}
189		-	// Cooperative cancel mid-conform: clean stop, no error surfaced.
190		-	Ok(Err(CoreError::Cancelled)) => continue,
191		-	Ok(Err(e)) => {
192		-	let _ = event_tx.send(ForgeEvent::Error {
193		-	error: e.to_string(),
194		-	});
195		-	}
196		-	Err(_panic) => {
197		-	let _ = event_tx.send(ForgeEvent::Error {
198		-	error: "conform panicked (internal error)".to_string(),
199		-	});
	119	+	return;
200	120		}
	121	+	ctx.emit(ForgeEvent::ChopComplete { staging });
201	122		}
	123	+	// Cooperative cancel mid-chop: clean stop, no error surfaced.
	124	+	Err(CoreError::Cancelled) => {}
	125	+	Err(e) => ctx.emit(ForgeEvent::Error {
	126	+	error: e.to_string(),
	127	+	}),
202	128		}
203		-	ForgeCommand::Preview {
204		-	source_path,
205		-	method,
206		-	} => {
207		-	cancel_flag.store(false, Ordering::Release);
208		-	let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
209		-	compute_preview_marks(&source_path, &method)
210		-	}));
211		-	// A newer preview/op supersedes this one — drop the stale marks.
212		-	if cancel_flag.load(Ordering::Acquire) {
213		-	continue;
214		-	}
215		-	match outcome {
216		-	Ok(Ok(marks)) => {
217		-	let _ = event_tx.send(ForgeEvent::PreviewComplete { marks });
218		-	}
219		-	Ok(Err(e)) => {
220		-	let _ = event_tx.send(ForgeEvent::Error {
221		-	error: e.to_string(),
222		-	});
223		-	}
224		-	Err(_panic) => {
225		-	let _ = event_tx.send(ForgeEvent::Error {
226		-	error: "chop preview panicked (internal error)".to_string(),
227		-	});
	129	+	}
	130	+	ForgeCommand::Conform {
	131	+	source_path,
	132	+	base_name,
	133	+	target,
	134	+	auto_trim,
	135	+	} => {
	136	+	ctx.reset_cancel();
	137	+	ctx.emit(ForgeEvent::Started);
	138	+	if ctx.is_cancelled() {
	139	+	return;
	140	+	}
	141	+	match conform_to_temp(&source_path, &base_name, &target, auto_trim, ctx.cancel_flag()) {
	142	+	Ok(staging) => {
	143	+	if ctx.is_cancelled() {
	144	+	let _ = std::fs::remove_file(&staging.temp_path);
	145	+	return;
228	146		}
	147	+	ctx.emit(ForgeEvent::ConformComplete { staging });
229	148		}
	149	+	// Cooperative cancel mid-conform: clean stop, no error surfaced.
	150	+	Err(CoreError::Cancelled) => {}
	151	+	Err(e) => ctx.emit(ForgeEvent::Error {
	152	+	error: e.to_string(),
	153	+	}),
	154	+	}
	155	+	}
	156	+	ForgeCommand::Preview {
	157	+	source_path,
	158	+	method,
	159	+	} => {
	160	+	ctx.reset_cancel();
	161	+	let outcome = compute_preview_marks(&source_path, &method);
	162	+	// A newer preview/op supersedes this one — drop the stale marks.
	163	+	if ctx.is_cancelled() {
	164	+	return;
	165	+	}
	166	+	match outcome {
	167	+	Ok(marks) => ctx.emit(ForgeEvent::PreviewComplete { marks }),
	168	+	Err(e) => ctx.emit(ForgeEvent::Error {
	169	+	error: e.to_string(),
	170	+	}),
230	171		}
231	172		}
232	173		}
			@@ -239,7 +180,6 @@ mod tests {
239	180		#[test]
240	181		fn forge_command_variants() {
241	182		let _cancel = ForgeCommand::Cancel;
242		-	let _shutdown = ForgeCommand::Shutdown;
243	183		let _chop = ForgeCommand::Chop {
244	184		source_path: PathBuf::from("/test.wav"),
245	185		base_name: "test.wav".to_string(),
			@@ -255,11 +195,10 @@ mod tests {
255	195		}
256	196
257	197		#[test]
258		-	fn cancel_flag_set_on_cancel() {
	198	+	fn cancel_then_drop_does_not_hang() {
259	199		let handle = spawn_forge_worker().unwrap();
260	200		handle.send(ForgeCommand::Cancel);
261	201		std::thread::sleep(std::time::Duration::from_millis(10));
262		-	assert!(handle.cancel_flag.load(Ordering::Relaxed));
263	202		drop(handle);
264	203		}
265	204		}

M crates/audiofiles-core/src/lib.rs +2

			@@ -29,6 +29,7 @@
29	29		//! - [`id_types`] — Strongly-typed entity ID newtypes ([`VfsId`], [`NodeId`], [`SampleHash`], etc.)
30	30		//!
31	31		//! ### Infrastructure
	32	+	//! - [`worker_runtime`] — Generic background-worker factory: owns the recv loop and `catch_unwind`s every command so a panic can never silently kill a worker thread
32	33		//! - [`db`] — SQLite database wrapper with versioned inline migrations (sync-only, no async)
33	34		//! - [`error`] — Unified error type ([`error::CoreError`]) and shared utilities
34	35		//! - [`util`] — Path/file helpers and audio extension whitelist
			@@ -55,6 +56,7 @@ pub mod vault;
55	56		pub mod vfs;
56	57		pub mod vfs_mirror;
57	58		pub mod vp_tree;
	59	+	pub mod worker_runtime;
58	60
59	61		pub use id_types::{CollectionId, NodeId, SampleHash, VfsId};
60	62

A crates/audiofiles-core/src/worker_runtime.rs +289

		@@ -0,0 +1,289 @@
1	+	//! Generic background-worker runtime.
2	+	//!
3	+	//! Every recv-loop worker in the app (analysis, edit, forge, import, export,
4	+	//! search, loose-files) routes its command loop through [`spawn_worker`]. The
5	+	//! factory owns the command channel's `Receiver` and the `recv` loop, and wraps
6	+	//! each command's handler in `catch_unwind` so a panic becomes a caller-defined
7	+	//! event instead of silently killing the thread — which would stop every future
8	+	//! command of that kind with no UI signal.
9	+	//!
10	+	//! Because the `Receiver` is created inside [`spawn_worker`] and never handed
11	+	//! back, no worker module can write its own unguarded `rx.recv()` loop. The
12	+	//! historically-recurring "added a worker, forgot the `catch_unwind`" drift (the
13	+	//! `loose_files_worker` regression) is structurally unwritable: a new worker has
14	+	//! no receiver to loop over except the one the runtime already guards.
15	+	//!
16	+	//! ## Cancellation
17	+	//!
18	+	//! Every worker gets a shared cancel flag for free (in [`WorkerCtx`]). It is set
19	+	//! to `true` automatically when the [`WorkerHandle`] is dropped, so an in-flight
20	+	//! long operation aborts promptly on shutdown instead of blocking the join.
21	+	//! Workers that support an explicit Cancel command wrap the handle in a thin
22	+	//! newtype whose `send` calls [`WorkerHandle::request_cancel`] before enqueuing,
23	+	//! so the flag is set synchronously by the GUI thread (the queued command itself
24	+	//! becomes a no-op). Long operations reset the flag at their start via
25	+	//! [`WorkerCtx::reset_cancel`] so a stale cancel does not abort the next command.
26	+
27	+	use std::sync::atomic::{AtomicBool, Ordering};
28	+	use std::sync::{mpsc, Arc, Mutex};
29	+	use std::thread;
30	+
31	+	/// Sink handed to a worker step so it can emit zero or more events per command.
32	+	///
33	+	/// Cloneable and `Sync` (it wraps an `mpsc::Sender`), so a step may share it
34	+	/// across rayon worker threads (e.g. the analysis batch fan-out).
35	+	pub struct EventSink<Ev> {
36	+	tx: mpsc::Sender<Ev>,
37	+	}
38	+
39	+	impl<Ev> Clone for EventSink<Ev> {
40	+	fn clone(&self) -> Self {
41	+	Self { tx: self.tx.clone() }
42	+	}
43	+	}
44	+
45	+	impl<Ev> EventSink<Ev> {
46	+	/// Emit one event back to the handle. Silently dropped if the handle is gone.
47	+	pub fn emit(&self, ev: Ev) {
48	+	let _ = self.tx.send(ev);
49	+	}
50	+
51	+	/// Create a sink paired with a receiver. For tests that exercise functions
52	+	/// taking an `&EventSink` directly, outside a spawned worker.
53	+	pub fn channel() -> (Self, mpsc::Receiver<Ev>) {
54	+	let (tx, rx) = mpsc::channel();
55	+	(Self { tx }, rx)
56	+	}
57	+	}
58	+
59	+	/// Context passed to each worker step: an [`EventSink`] plus the shared cancel
60	+	/// flag. Both are `Sync`, so `&WorkerCtx` can be shared into rayon closures.
61	+	pub struct WorkerCtx<Ev> {
62	+	sink: EventSink<Ev>,
63	+	cancel: Arc<AtomicBool>,
64	+	}
65	+
66	+	impl<Ev> WorkerCtx<Ev> {
67	+	/// Emit one event.
68	+	pub fn emit(&self, ev: Ev) {
69	+	self.sink.emit(ev);
70	+	}
71	+
72	+	/// The event sink, for handing to code that emits many events.
73	+	pub fn sink(&self) -> &EventSink<Ev> {
74	+	&self.sink
75	+	}
76	+
77	+	/// Whether cancellation has been requested for the current operation.
78	+	pub fn is_cancelled(&self) -> bool {
79	+	self.cancel.load(Ordering::Acquire)
80	+	}
81	+
82	+	/// The raw cancel flag, for passing to processing functions that take
83	+	/// `&AtomicBool` (e.g. `chop_to_temp`, `process_edit`).
84	+	pub fn cancel_flag(&self) -> &AtomicBool {
85	+	&self.cancel
86	+	}
87	+
88	+	/// Clear the cancel flag at the start of a fresh operation so a stale cancel
89	+	/// from a previous command does not abort this one.
90	+	pub fn reset_cancel(&self) {
91	+	self.cancel.store(false, Ordering::Release);
92	+	}
93	+	}
94	+
95	+	/// Handle to a background worker. Sends commands, polls events; `Drop` requests
96	+	/// cancellation, closes the command channel (ending the worker's recv loop), and
97	+	/// joins the thread.
98	+	///
99	+	/// Workers expose a thin newtype around this rather than the raw handle, so
100	+	/// worker-specific `send` semantics (mapping a Cancel command to
101	+	/// [`request_cancel`](Self::request_cancel)) live at the edge, not in the loop.
102	+	pub struct WorkerHandle<Cmd, Ev> {
103	+	cmd_tx: Option<mpsc::Sender<Cmd>>,
104	+	event_rx: Mutex<mpsc::Receiver<Ev>>,
105	+	cancel: Arc<AtomicBool>,
106	+	thread: Option<thread::JoinHandle<()>>,
107	+	}
108	+
109	+	impl<Cmd, Ev> WorkerHandle<Cmd, Ev> {
110	+	/// Poll for the next event without blocking.
111	+	pub fn try_recv(&self) -> Option<Ev> {
112	+	self.event_rx.lock().ok()?.try_recv().ok()
113	+	}
114	+
115	+	/// Enqueue a command for the worker.
116	+	pub fn send(&self, cmd: Cmd) {
117	+	if let Some(tx) = &self.cmd_tx {
118	+	let _ = tx.send(cmd);
119	+	}
120	+	}
121	+
122	+	/// Request cancellation of the in-flight operation synchronously (on the
123	+	/// caller's thread). Newtype handles call this from `send` for their Cancel
124	+	/// command so the flag is set before the running operation next checks it.
125	+	pub fn request_cancel(&self) {
126	+	self.cancel.store(true, Ordering::Release);
127	+	}
128	+	}
129	+
130	+	impl<Cmd, Ev> Drop for WorkerHandle<Cmd, Ev> {
131	+	fn drop(&mut self) {
132	+	// Abort any in-flight long operation, then close the channel so the
133	+	// recv loop ends, then join.
134	+	self.cancel.store(true, Ordering::Release);
135	+	self.cmd_tx = None;
136	+	if let Some(handle) = self.thread.take() {
137	+	let _ = handle.join();
138	+	}
139	+	}
140	+	}
141	+
142	+	/// Spawn a background worker.
143	+	///
144	+	/// - `init` runs once on the worker thread to build per-worker state (open a DB
145	+	/// or store, etc.). If it returns `Err`, `on_init_err` maps the error to an
146	+	/// event, that event is emitted, and the thread exits.
147	+	/// - `step` handles one command, using the [`WorkerCtx`] to emit any number of
148	+	/// events. Every `step` call is wrapped in `catch_unwind`; a panic is mapped
149	+	/// to a terminal event via `on_panic(&state)` (so the UI un-wedges for the
150	+	/// operation that was in flight) and the loop continues (the worker survives).
151	+	/// `on_panic` receives `&State` so it can carry context a step recorded before
152	+	/// panicking (e.g. the hash of the edit being processed).
153	+	///
154	+	/// For workers with no persistent state, use `State = ()` and an infallible
155	+	/// `init` returning `Ok::<(), std::convert::Infallible>(())` with
156	+	/// `on_init_err = \|e\| match e {}`.
157	+	pub fn spawn_worker<Cmd, Ev, State, InitErr>(
158	+	name: &str,
159	+	init: impl FnOnce() -> Result<State, InitErr> + Send + 'static,
160	+	on_init_err: impl FnOnce(InitErr) -> Ev + Send + 'static,
161	+	on_panic: impl Fn(&State) -> Ev + Send + 'static,
162	+	mut step: impl FnMut(&mut State, Cmd, &WorkerCtx<Ev>) + Send + 'static,
163	+	) -> std::io::Result<WorkerHandle<Cmd, Ev>>
164	+	where
165	+	Cmd: Send + 'static,
166	+	Ev: Send + 'static,
167	+	State: 'static,
168	+	InitErr: 'static,
169	+	{
170	+	let (cmd_tx, cmd_rx) = mpsc::channel::<Cmd>();
171	+	let (event_tx, event_rx) = mpsc::channel::<Ev>();
172	+	let cancel = Arc::new(AtomicBool::new(false));
173	+	let cancel_worker = Arc::clone(&cancel);
174	+
175	+	let thread = thread::Builder::new()
176	+	.name(name.to_string())
177	+	.spawn(move \|\| {
178	+	let ctx = WorkerCtx {
179	+	sink: EventSink { tx: event_tx },
180	+	cancel: cancel_worker,
181	+	};
182	+	let mut state = match init() {
183	+	Ok(s) => s,
184	+	Err(e) => {
185	+	ctx.emit(on_init_err(e));
186	+	return;
187	+	}
188	+	};
189	+	// Owned, guarded recv loop — the only consumer of `cmd_rx`, which
190	+	// never leaves this function. A panic in `step` becomes an event,
191	+	// not a dead thread.
192	+	while let Ok(cmd) = cmd_rx.recv() {
193	+	let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(\|\| {
194	+	step(&mut state, cmd, &ctx);
195	+	}));
196	+	if result.is_err() {
197	+	ctx.emit(on_panic(&state));
198	+	}
199	+	}
200	+	})?;
201	+
202	+	Ok(WorkerHandle {
203	+	cmd_tx: Some(cmd_tx),
204	+	event_rx: Mutex::new(event_rx),
205	+	cancel,
206	+	thread: Some(thread),
207	+	})
208	+	}
209	+
210	+	#[cfg(test)]
211	+	mod tests {
212	+	use super::*;
213	+	use std::convert::Infallible;
214	+
215	+	fn recv_one<Cmd, Ev>(handle: &WorkerHandle<Cmd, Ev>) -> Ev {
216	+	let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
217	+	while std::time::Instant::now() < deadline {
218	+	if let Some(ev) = handle.try_recv() {
219	+	return ev;
220	+	}
221	+	std::thread::sleep(std::time::Duration::from_millis(1));
222	+	}
223	+	panic!("worker produced no event");
224	+	}
225	+
226	+	#[test]
227	+	fn step_panic_becomes_event_and_worker_survives() {
228	+	// A command that panics must not kill the worker: the next command still
229	+	// gets processed. This is the invariant the loose_files regression broke.
230	+	let handle = spawn_worker::<u32, &'static str, (), Infallible>(
231	+	"test-panic",
232	+	\|\| Ok(()),
233	+	\|e\| match e {},
234	+	\|_state\| "panicked",
235	+	\|_state, cmd: u32, ctx\| {
236	+	if cmd == 0 {
237	+	panic!("boom");
238	+	}
239	+	ctx.emit("ok");
240	+	},
241	+	)
242	+	.unwrap();
243	+
244	+	handle.send(0); // panics
245	+	assert_eq!(recv_one(&handle), "panicked");
246	+	handle.send(1); // worker must still be alive
247	+	assert_eq!(recv_one(&handle), "ok");
248	+	drop(handle);
249	+	}
250	+
251	+	#[test]
252	+	fn init_failure_emits_event_and_exits() {
253	+	let handle = spawn_worker::<u32, &'static str, (), &'static str>(
254	+	"test-init-fail",
255	+	\|\| Err("no db"),
256	+	\|e\| e,
257	+	\|_state\| "panicked",
258	+	\|_state, _cmd, _ctx\| {},
259	+	)
260	+	.unwrap();
261	+	assert_eq!(recv_one(&handle), "no db");
262	+	drop(handle);
263	+	}
264	+
265	+	#[test]
266	+	fn drop_sets_cancel_flag() {
267	+	// Drop must request cancellation so an in-flight op aborts promptly.
268	+	let observed = Arc::new(AtomicBool::new(false));
269	+	let observed_clone = Arc::clone(&observed);
270	+	let handle = spawn_worker::<u32, (), (), Infallible>(
271	+	"test-cancel",
272	+	\|\| Ok(()),
273	+	\|e\| match e {},
274	+	\|_state\| (),
275	+	move \|_state, _cmd: u32, ctx\| {
276	+	// Spin until cancelled, then record that we saw it.
277	+	while !ctx.is_cancelled() {
278	+	std::thread::sleep(std::time::Duration::from_millis(1));
279	+	}
280	+	observed_clone.store(true, Ordering::Release);
281	+	},
282	+	)
283	+	.unwrap();
284	+	handle.send(1);
285	+	std::thread::sleep(std::time::Duration::from_millis(10));
286	+	drop(handle); // sets cancel, the step exits, join completes
287	+	assert!(observed.load(Ordering::Acquire));
288	+	}
289	+	}