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 (internal)/memory/0_18_redesign.md and
0_18_round3_migration_audit.md.

Single contributor: MattJackson.

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
-    /// next retry boundary. Unlike skip_errors, this propagates the error up so
-    /// the rip terminates cleanly. Share the Arc with Drive::halt_flag() to get
-    /// unified Stop behavior across drive reads and sector processing.
-    halt: Option<Arc<AtomicBool>>,
+    /// When set and the token is cancelled, fill_extents returns Err(Halted)
+    /// at the next retry boundary. Unlike skip_errors, this propagates the
+    /// error up so the rip terminates cleanly. Construct with
+    /// [`DiscStream::with_halt`] (preferred) or set post-hoc via the
+    /// 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,
-    /// the next read-retry boundary inside fill_extents returns Err(Halted)
-    /// instead of continuing. Required for Stop to work during dense bad-sector
-    /// regions (where read() loops internally waiting for enough clean data to
-    /// emit a frame and would otherwise never check an external stop signal).
+    /// Constructor-time builder: attach a [`Halt`] token so that when
+    /// any clone is cancelled, the next read-retry boundary inside
+    /// `fill_extents` returns `Err(Halted)`. Required for Stop to work
+    /// during dense bad-sector regions (where the outer PES read() loop
+    /// 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"
+        );
+    }
 }