Skip to main content

max / audiofiles

Fuzz run 5 follow-up: finish CHRONIC #1 seal, fix dead-worker repaint loop Run-5 confirmation audit found two gaps in the run-4 remediation: - CHRONIC #1 was incomplete: cleanup.rs was an 8th recv-loop worker, never migrated, still hand-rolling its loop with no catch_unwind (the exact loose_files failure mode). Migrate it to worker_runtime. Correct the runtime docstring's overclaim — the seal is a convention (every recv-loop worker must route through spawn_worker), not compiler-enforced; one-shot job threads (classifier, preview decode) carry their own guard and are a different shape. - Regression introduced by the run-4 repaint guard: a dead cached worker handle (if its init/DB-open failed inside the thread) left a busy flag set with no terminal event coming -> permanent full-rate repaint loop / forge busy-guard deadlock. WorkerHandle::send now returns whether the command reached a live worker (#[must_use]); the forge/search/loose-files backend start_* methods drop a dead handle and return Err so the caller never sets the busy flag. Also close cheap correctness items the audit surfaced: - remove_tags_by_source: wrap its two dependent deletes in one transaction. - wav_size_check: saturating_add to avoid a theoretical wrap. - import flush: surface save_analysis_batch errors to the status line. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 03:10 UTC
Commit: 445c39bdb37dee029cafad850955b9aafdb101f0
Parent: 01305c0
11 files changed, +157 insertions, -142 deletions
M crates/audiofiles-browser/src/backend/direct.rs +50 -12
@@ -143,7 +143,16 @@ impl DirectBackend {
143 143 )?;
144 144 *guard = Some(handle);
145 145 }
146 - guard.as_ref().expect("loose-files worker present").send(cmd);
146 + // If the cached worker died (e.g. its DB open failed inside the thread),
147 + // send returns false. Drop the dead handle so the next call respawns, and
148 + // report Err so the caller never sets a busy flag for a command that no
149 + // worker will ever answer (which would wedge the repaint-while-busy guard).
150 + if !guard.as_ref().expect("loose-files worker present").send(cmd) {
151 + *guard = None;
152 + return Err(BackendError::Other(
153 + "loose-files worker is not running".to_string(),
154 + ));
155 + }
147 156 Ok(())
148 157 }
149 158
@@ -638,7 +647,9 @@ impl Backend for DirectBackend {
638 647 // analysis data changes normalization ranges and may add fingerprints, so
639 648 // both indexes must rebuild on the next query.
640 649 if let Some(w) = self.search_worker.lock().as_ref() {
641 - w.send(SearchCommand::Invalidate);
650 + // Fire-and-forget cache drop; if the worker is gone it simply rebuilds
651 + // on next spawn, so a dropped Invalidate is harmless (no busy flag).
652 + let _ = w.send(SearchCommand::Invalidate);
642 653 }
643 654 Ok(())
644 655 }
@@ -874,22 +885,35 @@ impl Backend for DirectBackend {
874 885
875 886 fn start_find_similar(&self, hash: &str, limit: usize) -> BackendResult<()> {
876 887 self.ensure_search_worker()?;
877 - if let Some(w) = self.search_worker.lock().as_ref() {
888 + let mut guard = self.search_worker.lock();
889 + let delivered = guard.as_ref().is_some_and(|w| {
878 890 w.send(SearchCommand::FindSimilar {
879 891 hash: hash.to_string(),
880 892 limit,
881 - });
893 + })
894 + });
895 + // A dead cached worker would never answer; drop it (next call respawns)
896 + // and report Err so the caller doesn't set latest_similarity_request and
897 + // spin the repaint-while-busy guard forever.
898 + if !delivered {
899 + *guard = None;
900 + return Err(BackendError::Other("search worker is not running".to_string()));
882 901 }
883 902 Ok(())
884 903 }
885 904
886 905 fn start_find_near_duplicates(&self, hash: &str, limit: usize) -> BackendResult<()> {
887 906 self.ensure_search_worker()?;
888 - if let Some(w) = self.search_worker.lock().as_ref() {
907 + let mut guard = self.search_worker.lock();
908 + let delivered = guard.as_ref().is_some_and(|w| {
889 909 w.send(SearchCommand::FindNearDuplicates {
890 910 hash: hash.to_string(),
891 911 limit,
892 - });
912 + })
913 + });
914 + if !delivered {
915 + *guard = None;
916 + return Err(BackendError::Other("search worker is not running".to_string()));
893 917 }
894 918 Ok(())
895 919 }
@@ -1065,11 +1089,17 @@ impl Backend for DirectBackend {
1065 1089 .map_err(|e| BackendError::Other(format!("failed to spawn forge worker: {e}")))?;
1066 1090 *guard = Some(handle);
1067 1091 }
1068 - if let Some(w) = guard.as_ref() {
1092 + let delivered = guard.as_ref().is_some_and(|w| {
1069 1093 w.send(ForgeCommand::Preview {
1070 1094 source_path: path,
1071 1095 method: method.clone(),
1072 - });
1096 + })
1097 + });
1098 + // Dead cached worker → drop it (next call respawns) and report Err so the
1099 + // caller doesn't set forge.busy and wedge the repaint guard / busy-guard.
1100 + if !delivered {
1101 + *guard = None;
1102 + return Err(BackendError::Other("forge worker is not running".to_string()));
1073 1103 }
1074 1104 Ok(())
1075 1105 }
@@ -1165,12 +1195,16 @@ impl Backend for DirectBackend {
1165 1195 );
1166 1196 }
1167 1197 let handle = guard.as_ref().expect("forge worker present");
1168 - handle.send(ForgeCommand::Cancel);
1169 - handle.send(ForgeCommand::Chop {
1198 + let _ = handle.send(ForgeCommand::Cancel);
1199 + let delivered = handle.send(ForgeCommand::Chop {
1170 1200 source_path: path,
1171 1201 base_name: name.to_string(),
1172 1202 method: method.clone(),
1173 1203 });
1204 + if !delivered {
1205 + *guard = None;
1206 + return Err(BackendError::Other("forge worker is not running".to_string()));
1207 + }
1174 1208 }
1175 1209 *self.forge_pending.lock() = Some(ForgePending::Chop { vfs_id, parent_id });
1176 1210 Ok(())
@@ -1213,13 +1247,17 @@ impl Backend for DirectBackend {
1213 1247 );
1214 1248 }
1215 1249 let handle = guard.as_ref().expect("forge worker present");
1216 - handle.send(ForgeCommand::Cancel);
1217 - handle.send(ForgeCommand::Conform {
1250 + let _ = handle.send(ForgeCommand::Cancel);
1251 + let delivered = handle.send(ForgeCommand::Conform {
1218 1252 source_path: path,
1219 1253 base_name: name.to_string(),
1220 1254 target: target.clone(),
1221 1255 auto_trim,
1222 1256 });
1257 + if !delivered {
1258 + *guard = None;
1259 + return Err(BackendError::Other("forge worker is not running".to_string()));
1260 + }
1223 1261 }
1224 1262 *self.forge_pending.lock() = Some(ForgePending::Conform {
1225 1263 vfs_id,
M crates/audiofiles-browser/src/cleanup.rs +45 -93
@@ -4,22 +4,19 @@
4 4 //! SampleStore, 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 8 use tracing::{error, instrument};
11 9
12 10 use audiofiles_core::db::Database;
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 cleanup worker.
16 15 pub enum CleanupCommand {
17 16 /// Start removing orphaned samples.
18 17 RemoveOrphans,
19 - /// Cancel the current cleanup.
18 + /// Cancel the current cleanup (sets the worker's cancel flag synchronously).
20 19 Cancel,
21 - /// Shut down the worker thread.
22 - Shutdown,
23 20 }
24 21
25 22 /// Event sent from the cleanup worker back to the GUI thread.
@@ -37,35 +34,32 @@ pub enum CleanupEvent {
37 34 },
38 35 }
39 36
40 - /// Handle for communicating with the background cleanup worker.
41 - ///
42 - /// The receiver is wrapped in a `Mutex` so `BrowserState` remains `Sync` (required by nih-plug).
43 - /// Only the GUI thread actually calls `try_recv`, so contention is zero.
44 - pub struct CleanupHandle {
45 - cmd_tx: mpsc::Sender<CleanupCommand>,
46 - event_rx: Mutex<mpsc::Receiver<CleanupEvent>>,
47 - _thread: Option<thread::JoinHandle<()>>,
48 - }
37 + /// Handle for communicating with the background cleanup worker. Thin newtype over
38 + /// the generic [`WorkerHandle`] so the recv loop + `catch_unwind` live in the
39 + /// runtime (this worker previously hand-rolled its loop with no panic guard).
40 + pub struct CleanupHandle(WorkerHandle<CleanupCommand, CleanupEvent>);
49 41
50 42 impl CleanupHandle {
51 43 /// Poll for the next event without blocking.
52 44 pub fn try_recv(&self) -> Option<CleanupEvent> {
53 - self.event_rx.lock().ok()?.try_recv().ok()
45 + self.0.try_recv()
54 46 }
55 47
56 - /// Send a command to the worker.
57 - pub fn send(&self, cmd: CleanupCommand) {
58 - let _ = self.cmd_tx.send(cmd);
48 + /// Send a command to the worker. Returns false if the worker is no longer
49 + /// alive (its thread ended, e.g. a failed init), so callers don't treat a
50 + /// dropped command as accepted.
51 + pub fn send(&self, cmd: CleanupCommand) -> bool {
52 + if matches!(cmd, CleanupCommand::Cancel) {
53 + self.0.request_cancel();
54 + }
55 + self.0.send(cmd)
59 56 }
60 57 }
61 58
62 - impl Drop for CleanupHandle {
63 - fn drop(&mut self) {
64 - let _ = self.cmd_tx.send(CleanupCommand::Shutdown);
65 - if let Some(handle) = self._thread.take() {
66 - let _ = handle.join();
67 - }
68 - }
59 + /// Per-worker state: its own DB connection + store.
60 + struct CleanupWorker {
61 + db: Database,
62 + store: SampleStore,
69 63 }
70 64
71 65 /// Spawn the background cleanup worker thread.
@@ -77,76 +71,35 @@ pub fn spawn_cleanup_worker(
77 71 db_path: PathBuf,
78 72 store_root: PathBuf,
79 73 ) -> std::io::Result<CleanupHandle> {
80 - let (cmd_tx, cmd_rx) = mpsc::channel::<CleanupCommand>();
81 - let (event_tx, event_rx) = mpsc::channel::<CleanupEvent>();
82 -
83 - let thread = thread::Builder::new()
84 - .name("cleanup-worker".to_string())
85 - .spawn(move || {
86 - worker_loop(cmd_rx, event_tx, &db_path, &store_root);
87 - })?;
88 -
89 - Ok(CleanupHandle {
90 - cmd_tx,
91 - event_rx: Mutex::new(event_rx),
92 - _thread: Some(thread),
93 - })
74 + let handle = spawn_worker(
75 + "cleanup-worker",
76 + move || -> Result<CleanupWorker, audiofiles_core::error::CoreError> {
77 + let db = Database::open(&db_path)?;
78 + let store = SampleStore::new(&store_root)?;
79 + Ok(CleanupWorker { db, store })
80 + },
81 + |e| {
82 + error!("Cleanup worker failed to open DB/store: {e}");
83 + CleanupEvent::Complete { removed: 0, errors: 1 }
84 + },
85 + |_state| CleanupEvent::Complete { removed: 0, errors: 1 },
86 + cleanup_step,
87 + )?;
88 + Ok(CleanupHandle(handle))
94 89 }
95 90
96 - #[instrument(skip_all)]
97 - fn worker_loop(
98 - cmd_rx: mpsc::Receiver<CleanupCommand>,
99 - event_tx: mpsc::Sender<CleanupEvent>,
100 - db_path: &std::path::Path,
101 - store_root: &std::path::Path,
102 - ) {
103 - let db = match Database::open(db_path) {
104 - Ok(d) => d,
105 - Err(e) => {
106 - let _ = event_tx.send(CleanupEvent::Complete {
107 - removed: 0,
108 - errors: 1,
109 - });
110 - error!("Cleanup worker failed to open database: {e}");
111 - return;
112 - }
113 - };
114 -
115 - let store = match SampleStore::new(store_root) {
116 - Ok(s) => s,
117 - Err(e) => {
118 - let _ = event_tx.send(CleanupEvent::Complete {
119 - removed: 0,
120 - errors: 1,
121 - });
122 - error!("Cleanup worker failed to open store: {e}");
123 - return;
124 - }
125 - };
126 -
127 - while let Ok(cmd) = cmd_rx.recv() {
128 - match cmd {
129 - CleanupCommand::Shutdown => break,
130 - CleanupCommand::Cancel => continue,
131 - CleanupCommand::RemoveOrphans => {
132 - remove_orphans(&cmd_rx, &event_tx, &db, &store);
133 - }
134 - }
91 + fn cleanup_step(worker: &mut CleanupWorker, cmd: CleanupCommand, ctx: &WorkerCtx<CleanupEvent>) {
92 + // Cancel: flag already set by the handle; remove_orphaned_samples is a single
93 + // atomic core op, so there is nothing finer to interrupt.
94 + if matches!(cmd, CleanupCommand::Cancel) {
95 + return;
135 96 }
136 - }
137 -
138 - fn remove_orphans(
139 - _cmd_rx: &mpsc::Receiver<CleanupCommand>,
140 - event_tx: &mpsc::Sender<CleanupEvent>,
141 - db: &Database,
142 - store: &SampleStore,
143 - ) {
144 97 // Delegate to the single sealed core operation: it suppresses sync triggers
145 98 // internally (so local GC never pushes a destructive `samples` DELETE across
146 99 // devices), re-checks orphan status atomically, and runs the row deletes in
147 - // one transaction. Re-implementing the loop here is what let this worker
148 - // drift from the safe path — keep the logic in exactly one place.
149 - let (removed, errors) = match store.remove_orphaned_samples(db) {
100 + // one transaction. A panic here is now caught by the runtime and reported as
101 + // Complete{errors:1} instead of silently killing the worker.
102 + let (removed, errors) = match worker.store.remove_orphaned_samples(&worker.db) {
150 103 Ok(removed) => (removed, 0),
151 104 Err(e) => {
152 105 error!("Cleanup worker failed to remove orphans: {e}");
@@ -155,9 +108,9 @@ fn remove_orphans(
155 108 };
156 109
157 110 // WAL checkpoint for clean state.
158 - let _ = db.conn().execute_batch("PRAGMA wal_checkpoint(TRUNCATE)");
111 + let _ = worker.db.conn().execute_batch("PRAGMA wal_checkpoint(TRUNCATE)");
159 112
160 - let _ = event_tx.send(CleanupEvent::Complete { removed, errors });
113 + ctx.emit(CleanupEvent::Complete { removed, errors });
161 114 }
162 115
163 116 #[cfg(test)]
@@ -168,7 +121,6 @@ mod tests {
168 121 fn cleanup_command_variants_constructible() {
169 122 let _remove = CleanupCommand::RemoveOrphans;
170 123 let _cancel = CleanupCommand::Cancel;
171 - let _shutdown = CleanupCommand::Shutdown;
172 124 }
173 125
174 126 #[test]
@@ -209,7 +161,7 @@ mod tests {
209 161 let _db = Database::open(&db_path).unwrap();
210 162
211 163 let handle = spawn_cleanup_worker(db_path, store_root).unwrap();
212 - handle.send(CleanupCommand::RemoveOrphans);
164 + assert!(handle.send(CleanupCommand::RemoveOrphans));
213 165
214 166 // Give the worker a moment to process
215 167 std::thread::sleep(std::time::Duration::from_millis(100));
M crates/audiofiles-browser/src/export.rs +3 -3
@@ -49,12 +49,12 @@ impl ExportHandle {
49 49 self.0.try_recv()
50 50 }
51 51
52 - /// Send a command to the worker.
53 - pub fn send(&self, cmd: ExportCommand) {
52 + /// Send a command to the worker. Returns false if the worker is no longer alive.
53 + pub fn send(&self, cmd: ExportCommand) -> bool {
54 54 if matches!(cmd, ExportCommand::Cancel) {
55 55 self.0.request_cancel();
56 56 }
57 - self.0.send(cmd);
57 + self.0.send(cmd)
58 58 }
59 59 }
60 60
M crates/audiofiles-browser/src/import.rs +3 -3
@@ -113,12 +113,12 @@ impl ImportHandle {
113 113 self.0.try_recv()
114 114 }
115 115
116 - /// Send a command to the worker.
117 - pub fn send(&self, cmd: ImportCommand) {
116 + /// Send a command to the worker. Returns false if the worker is no longer alive.
117 + pub fn send(&self, cmd: ImportCommand) -> bool {
118 118 if matches!(cmd, ImportCommand::Cancel) {
119 119 self.0.request_cancel();
120 120 }
121 - self.0.send(cmd);
121 + self.0.send(cmd)
122 122 }
123 123 }
124 124
M crates/audiofiles-browser/src/state/import_workflow.rs +7 -1
@@ -66,7 +66,13 @@ impl BrowserState {
66 66 }
67 67 let results = std::mem::take(&mut self.analysis_save_buffer);
68 68 let hashes: Vec<String> = results.iter().map(|r| r.hash.clone()).collect();
69 - let _ = self.backend.save_analysis_batch(&results);
69 + let n = results.len();
70 + if let Err(e) = self.backend.save_analysis_batch(&results) {
71 + // The transaction is atomic (no partial write), but the user should
72 + // know the analysis didn't persist rather than see silent loss.
73 + self.status = format!("Failed to save analysis for {n} samples: {e}");
74 + return;
75 + }
70 76 let _ = self.backend.apply_rules_to_samples(&hashes);
71 77 }
72 78
M crates/audiofiles-core/src/analysis/worker.rs +3 -3
@@ -72,12 +72,12 @@ impl WorkerHandle {
72 72 self.0.try_recv()
73 73 }
74 74
75 - /// Send a command to the worker.
76 - pub fn send(&self, cmd: WorkerCommand) {
75 + /// Send a command to the worker. Returns false if the worker is no longer alive.
76 + pub fn send(&self, cmd: WorkerCommand) -> bool {
77 77 if matches!(cmd, WorkerCommand::Cancel) {
78 78 self.0.request_cancel();
79 79 }
80 - self.0.send(cmd);
80 + self.0.send(cmd)
81 81 }
82 82 }
83 83
M crates/audiofiles-core/src/edit/worker.rs +3 -3
@@ -55,12 +55,12 @@ impl EditWorkerHandle {
55 55 self.0.try_recv()
56 56 }
57 57
58 - /// Send a command to the worker.
59 - pub fn send(&self, cmd: EditCommand) {
58 + /// Send a command to the worker. Returns false if the worker is no longer alive.
59 + pub fn send(&self, cmd: EditCommand) -> bool {
60 60 if matches!(cmd, EditCommand::Cancel) {
61 61 self.0.request_cancel();
62 62 }
63 - self.0.send(cmd);
63 + self.0.send(cmd)
64 64 }
65 65 }
66 66
M crates/audiofiles-core/src/export/encode.rs +1 -1
@@ -82,7 +82,7 @@ pub(crate) fn wav_size_check(total_samples: u64, bit_depth: u16) -> Result<(), C
82 82 let bytes_per_sample = (bit_depth / 8) as u64;
83 83 let data_size = total_samples.saturating_mul(bytes_per_sample);
84 84 // The RIFF chunk-size field (u32) covers the 36-byte header remainder + data.
85 - if data_size + 36 > u32::MAX as u64 {
85 + if data_size.saturating_add(36) > u32::MAX as u64 {
86 86 return Err(CoreError::Export(format!(
87 87 "WAV: file too large ({total_samples} samples, {bit_depth}-bit = {data_size} bytes, exceeds 4 GB RIFF limit)"
88 88 )));
M crates/audiofiles-core/src/forge/worker.rs +3 -3
@@ -69,12 +69,12 @@ impl ForgeWorkerHandle {
69 69 self.0.try_recv()
70 70 }
71 71
72 - /// Send a command to the worker.
73 - pub fn send(&self, cmd: ForgeCommand) {
72 + /// Send a command to the worker. Returns false if the worker is no longer alive.
73 + pub fn send(&self, cmd: ForgeCommand) -> bool {
74 74 if matches!(cmd, ForgeCommand::Cancel) {
75 75 self.0.request_cancel();
76 76 }
77 - self.0.send(cmd);
77 + self.0.send(cmd)
78 78 }
79 79 }
80 80
M crates/audiofiles-core/src/rules.rs +15 -8
@@ -784,14 +784,21 @@ pub fn apply_tag_sourced(db: &Database, _tx: &Tx, hash: &str, tag: &str, source:
784 784 /// affects tags whose provenance matches `source`; manual tags are untouched.
785 785 #[instrument(skip_all)]
786 786 pub fn remove_tags_by_source(db: &Database, source: &str) -> Result<usize> {
787 - let conn = db.conn();
788 - let removed = conn.execute(
789 - "DELETE FROM tags WHERE (sample_hash, tag) IN
790 - (SELECT sample_hash, tag FROM tag_provenance WHERE source = ?1)",
791 - rusqlite::params![source],
792 - )?;
793 - conn.execute("DELETE FROM tag_provenance WHERE source = ?1", rusqlite::params![source])?;
794 - Ok(removed)
787 + // Two dependent deletes (tags, then their provenance) in one transaction, so a
788 + // crash between them can't orphan provenance rows from their tags.
789 + db.transaction_core(|_tx| {
790 + let conn = db.conn();
791 + let removed = conn.execute(
792 + "DELETE FROM tags WHERE (sample_hash, tag) IN
793 + (SELECT sample_hash, tag FROM tag_provenance WHERE source = ?1)",
794 + rusqlite::params![source],
795 + )?;
796 + conn.execute(
797 + "DELETE FROM tag_provenance WHERE source = ?1",
798 + rusqlite::params![source],
799 + )?;
800 + Ok(removed)
801 + })
795 802 }
796 803
797 804 /// Provenance of each tag on a sample: `(tag, source, rule_id)`. Tags absent from
M crates/audiofiles-core/src/worker_runtime.rs +24 -12
@@ -7,11 +7,16 @@
7 7 //! event instead of silently killing the thread — which would stop every future
8 8 //! command of that kind with no UI signal.
9 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.
10 + //! Every recv-loop worker routes through here, so the `catch_unwind` lives in one
11 + //! place instead of being copy-pasted (and forgotten) per worker — the
12 + //! historically-recurring "added a worker, forgot the guard" drift (the
13 + //! `loose_files_worker` regression). This is enforced by convention, not by the
14 + //! compiler: nothing *prevents* a new worker from hand-rolling its own
15 + //! `mpsc::channel()` + `rx.recv()` loop. The rule is: any worker that loops over
16 + //! commands MUST be spawned via [`spawn_worker`]; one-shot job threads (which run
17 + //! a single task and exit, e.g. the classifier job and the preview decode) carry
18 + //! their own `catch_unwind` and are a different shape. A new bare recv loop is a
19 + //! review smell, not a build error.
15 20 //!
16 21 //! ## Cancellation
17 22 //!
@@ -112,10 +117,17 @@ impl<Cmd, Ev> WorkerHandle<Cmd, Ev> {
112 117 self.event_rx.lock().ok()?.try_recv().ok()
113 118 }
114 119
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);
120 + /// Enqueue a command for the worker. Returns `false` if the worker is no
121 + /// longer alive — its receive loop has ended (e.g. `init` failed, or the
122 + /// thread panicked out past the guard), so the command was dropped. Callers
123 + /// that set a "busy" flag on dispatch must check this: setting busy for a
124 + /// command that never reached a worker means no terminal event will ever
125 + /// clear the flag.
126 + #[must_use]
127 + pub fn send(&self, cmd: Cmd) -> bool {
128 + match &self.cmd_tx {
129 + Some(tx) => tx.send(cmd).is_ok(),
130 + None => false,
119 131 }
120 132 }
121 133
@@ -241,9 +253,9 @@ mod tests {
241 253 )
242 254 .unwrap();
243 255
244 - handle.send(0); // panics
256 + assert!(handle.send(0)); // panics (delivered, then caught)
245 257 assert_eq!(recv_one(&handle), "panicked");
246 - handle.send(1); // worker must still be alive
258 + assert!(handle.send(1)); // worker must still be alive
247 259 assert_eq!(recv_one(&handle), "ok");
248 260 drop(handle);
249 261 }
@@ -281,7 +293,7 @@ mod tests {
281 293 },
282 294 )
283 295 .unwrap();
284 - handle.send(1);
296 + let _ = handle.send(1);
285 297 std::thread::sleep(std::time::Duration::from_millis(10));
286 298 drop(handle); // sets cancel, the step exits, join completes
287 299 assert!(observed.load(Ordering::Acquire));