0.18 round 2 (Halt threading): DiscStream accepts Halt at construction

Adds a Halt field to DiscStream, populated via the new
`with_halt(halt)` builder. The internal recovery / fill_extents
loops check `halt.is_cancelled()` directly. The existing
`set_halt(Arc<AtomicBool>)` method stays through the deprecation
window for callers (autorip mux) that haven't migrated; marked
#[deprecated] with a pointer to the constructor-time path.

Both signals are unified inside DiscStream: either Halt or the
legacy Arc<AtomicBool> triggers cancellation, so callers can mix
during the deprecation window without breaking stop behaviour.

See freemkv-private/memory/0_18_redesign.md and
0_18_round3_migration_audit.md.

Single contributor: MattJackson.

src/halt.rs

@@ -34,6 +34,26 @@ impl Halt {
         Self(Arc::new(AtomicBool::new(false)))
     }
 
+    /// Wrap an existing `Arc<AtomicBool>` as a `Halt`. Useful as a
+    /// bridge during the 0.18 deprecation window: callers that already
+    /// hold an `Arc<AtomicBool>` (e.g. `Drive::halt_flag()`, the
+    /// deprecated `DiscStream::set_halt`) can adopt the new token API
+    /// without changing the underlying flag.
+    ///
+    /// Cancelling either side flips the same bit — the wrapping `Halt`
+    /// and the original `Arc` are two views over one shared flag.
+    pub fn from_arc(flag: Arc<AtomicBool>) -> Self {
+        Self(flag)
+    }
+
+    /// Borrow the underlying `Arc<AtomicBool>`. Used at boundaries with
+    /// pre-`Halt` APIs that still take an `Arc<AtomicBool>` directly
+    /// (`CopyOptions::halt`, the deprecated `DiscStream::set_halt`).
+    /// Round 3 deletes those boundaries and this accessor with them.
+    pub fn as_arc(&self) -> &Arc<AtomicBool> {
+        &self.0
+    }
+
     /// Flip the shared flag to cancelled. Idempotent.
     pub fn cancel(&self) {
         self.0.store(true, Ordering::Relaxed);
@@ -110,4 +130,37 @@ mod tests {
         handle.join().unwrap();
         assert!(h.is_cancelled());
     }
+
+    #[test]
+    fn from_arc_shares_state() {
+        // The 0.18 deprecation-window bridge: a Halt built from an
+        // existing Arc<AtomicBool> must be a *view* over the same bit,
+        // not a fresh copy. Cancelling either side flips both.
+        let arc = Arc::new(AtomicBool::new(false));
+        let halt = Halt::from_arc(arc.clone());
+        assert!(!halt.is_cancelled());
+        assert!(!arc.load(Ordering::Relaxed));
+
+        // Cancel via the wrapping Halt; the original Arc observes it.
+        halt.cancel();
+        assert!(arc.load(Ordering::Relaxed));
+
+        // Conversely: flip the Arc directly; the Halt view observes it.
+        let arc2 = Arc::new(AtomicBool::new(false));
+        let halt2 = Halt::from_arc(arc2.clone());
+        arc2.store(true, Ordering::Relaxed);
+        assert!(halt2.is_cancelled());
+    }
+
+    #[test]
+    fn as_arc_returns_backing_flag() {
+        // `as_arc()` must hand back the *same* Arc, not a clone of a
+        // different bit. Verified by writing through the borrowed Arc
+        // and observing through the Halt.
+        let halt = Halt::new();
+        let arc = halt.as_arc().clone();
+        assert!(!halt.is_cancelled());
+        arc.store(true, Ordering::Relaxed);
+        assert!(halt.is_cancelled());
+    }
 }

src/mux/disc.rs

@@ -8,10 +8,11 @@
 use crate::disc::{Disc, DiscTitle, Extent};
 use crate::drive::extract_scsi_context;
 use crate::event::{BatchSizeReason, Event, EventKind};
+use crate::halt::Halt;
 use crate::sector::SectorReader;
 use std::io;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::AtomicBool;
 
 /// Ramp back up to the preferred batch size after this many sectors
 /// of clean reading at the current (reduced) size. 100 MiB = 51,200 sectors.
@@ -120,11 +121,13 @@ pub struct DiscStream {
     adaptive: AdaptiveBatch,
     pub errors: u64,
     pub skip_errors: bool,
-    /// When set and the flag is raised, fill_extents returns Err(Halted) at the
+    /// When set and the token is cancelled, fill_extents returns Err(Halted)
-    /// next retry boundary. Unlike skip_errors, this propagates the error up so
+    /// at the next retry boundary. Unlike skip_errors, this propagates the
-    /// the rip terminates cleanly. Share the Arc with Drive::halt_flag() to get
+    /// error up so the rip terminates cleanly. Construct with
-    /// unified Stop behavior across drive reads and sector processing.
+    /// [`DiscStream::with_halt`] (preferred) or set post-hoc via the
-    halt: Option<Arc<AtomicBool>>,
+    /// deprecated [`DiscStream::set_halt`] bridge — both populate this same
+    /// field and either entry point yields one source of truth.
+    halt: Option<Halt>,
     event_fn: Option<Box<dyn Fn(Event) + Send>>,
     eof: bool,
 
@@ -219,19 +222,39 @@ impl DiscStream {
         self.event_fn = Some(Box::new(f));
     }
 
-    /// Share a halt flag — typically from `Drive::halt_flag()`. When raised,
+    /// Constructor-time builder: attach a [`Halt`] token so that when
-    /// the next read-retry boundary inside fill_extents returns Err(Halted)
+    /// any clone is cancelled, the next read-retry boundary inside
-    /// instead of continuing. Required for Stop to work during dense bad-sector
+    /// `fill_extents` returns `Err(Halted)`. Required for Stop to work
-    /// regions (where read() loops internally waiting for enough clean data to
+    /// during dense bad-sector regions (where the outer PES read() loop
-    /// emit a frame and would otherwise never check an external stop signal).
+    /// can spend minutes inside fill_extents before emitting a frame).
+    ///
+    /// Preferred over the post-hoc [`DiscStream::set_halt`] bridge —
+    /// pass the same `Halt` clone you hand to sweep / patch / mux so
+    /// every phase observes a single Stop signal.
+    pub fn with_halt(mut self, halt: Halt) -> Self {
+        self.halt = Some(halt);
+        self
+    }
+
+    /// Bridge for callers that haven't migrated to the
+    /// [`DiscStream::with_halt`] constructor-time path yet. Wraps the
+    /// supplied `Arc<AtomicBool>` as a [`Halt`] (`Halt::from_arc`) and
+    /// stores it in the same internal slot, so a halt installed via
+    /// either entry point goes through one halt-check inside
+    /// `fill_extents`. Calling `set_halt` after `with_halt` (or vice
+    /// versa) replaces the previous token with the new one.
+    #[deprecated(
+        since = "0.18.0",
+        note = "use `DiscStream::with_halt(Halt)` at construction instead"
+    )]
     pub fn set_halt(&mut self, flag: Arc<AtomicBool>) {
-        self.halt = Some(flag);
+        self.halt = Some(Halt::from_arc(flag));
     }
 
     fn is_halted(&self) -> bool {
         self.halt
             .as_ref()
-            .map(|h| h.load(Ordering::Relaxed))
+            .map(|h| h.is_cancelled())
             .unwrap_or(false)
     }
 
@@ -629,4 +652,47 @@ mod tests {
         }
         assert_eq!(frames, 0);
     }
+
+    /// `is_halted()` must observe a cancellation signal regardless of
+    /// which entry point installed the token. The deprecated
+    /// `set_halt(Arc<AtomicBool>)` and the new `with_halt(Halt)` are
+    /// two views over one slot — flipping either bit must cause the
+    /// next `fill_extents` retry boundary to bail.
+    #[test]
+    fn halt_via_with_halt_observed_by_is_halted() {
+        let halt = Halt::new();
+        let stream = DiscStream::new(
+            Box::new(ZeroReader { capacity: 8 }),
+            synthetic_title(8),
+            crate::decrypt::DecryptKeys::None,
+            8,
+            crate::disc::ContentFormat::BdTs,
+        )
+        .with_halt(halt.clone());
+        assert!(!stream.is_halted());
+        halt.cancel();
+        assert!(
+            stream.is_halted(),
+            "with_halt token cancellation must be observed by is_halted()"
+        );
+    }
+
+    #[test]
+    fn halt_via_set_halt_bridge_observed_by_is_halted() {
+        let arc = Arc::new(AtomicBool::new(false));
+        let mut stream = DiscStream::new(
+            Box::new(ZeroReader { capacity: 8 }),
+            synthetic_title(8),
+            crate::decrypt::DecryptKeys::None,
+            8,
+            crate::disc::ContentFormat::BdTs,
+        );
+        stream.set_halt(arc.clone());
+        assert!(!stream.is_halted());
+        arc.store(true, std::sync::atomic::Ordering::Relaxed);
+        assert!(
+            stream.is_halted(),
+            "set_halt(Arc<AtomicBool>) bridge must observe Arc-side flips"
+        );
+    }
 }