From c3c5259f845a7ce0a7e7a6a65c199c29a3cdcad3 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Mon, 22 Jun 2026 23:22:57 -0700 Subject: [PATCH] pipeline: leaked consumer must not finalise an abandoned output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When finish_with_halt's grace period expires it detaches from the consumer thread and returns an error to the caller, but the leaked consumer kept running to completion: once its wedged write syscall returned it would fall through to sink.close(). For the mux writer close() finalises the MKV (Cues block + segment-header patch), so a leaked consumer could finalise — and keep writing to — an output file the caller had already reported as failed, racing a fresh rip for the same device over the same path. Add a shared abandonment flag the consumer polls in its drain loop and again before close(). finish_with_grace sets it before dropping the JoinHandle, so the moment the wedged syscall returns the consumer skips any further apply and skips close() entirely, then exits. This does not interrupt the in-flight syscall (only its return or process exit can), but it bounds the damage to the write already in flight instead of a full finalise of an abandoned file. Regression tests cover both sides: a consumer leaked past the grace period skips close(), while one that finishes inside the grace window still calls close() and finalises normally. --- src/io/pipeline.rs | 228 ++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 217 insertions(+), 11 deletions(-) diff --git a/src/io/pipeline.rs b/src/io/pipeline.rs index 20fad9f..c1697ba 100644 --- a/src/io/pipeline.rs +++ b/src/io/pipeline.rs @@ -32,6 +32,8 @@ //! logging throughout the pipeline (channel sends/receives, backpressure, //! consumer lag detection). This is critical for diagnosing stalls. +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; @@ -121,11 +123,16 @@ fn consumer_panicked(payload: Box) -> Error { /// returned and is about to drop its output file) into a clean join, /// releasing the file handle without waiting the full grace period. /// -/// If the consumer is still running when the grace expires, dropping the -/// `JoinHandle` detaches from the thread — the consumer keeps running -/// until its kernel call returns or the process exits. +/// If the consumer is still running when the grace expires, the +/// `abandoned` flag is set and the `JoinHandle` is dropped, detaching +/// from the thread. The leaked consumer keeps running until its current +/// kernel call returns, then — observing `abandoned` — exits WITHOUT +/// calling `close()`, so it does not finalise an output the caller has +/// already reported as failed. It does not unblock the in-flight +/// syscall itself; that still returns on its own (or at process exit). fn finish_with_grace( handle: thread::JoinHandle>, + abandoned: &Arc, leak_err: Error, ) -> Result { let grace = Instant::now() + Duration::from_secs(FINISH_GRACE_SECS); @@ -138,17 +145,24 @@ fn finish_with_grace( } thread::sleep(POLL_INTERVAL); } - // Grace expired. Log and leak. + // Grace expired. Signal abandonment, then log and leak. Setting the + // flag BEFORE dropping the handle guarantees the leaked consumer + // observes it the moment its wedged syscall returns: it then skips + // any further `apply` and skips `close()`, rather than running on to + // finalise the abandoned output file. + abandoned.store(true, Ordering::Relaxed); tracing::warn!( target: "freemkv::pipeline", phase = "finish_with_halt_grace_expired", - "pipeline consumer did not finish within {}s grace period; leaking thread", + "pipeline consumer did not finish within {}s grace period; abandoning thread \ + (output will not be finalised)", FINISH_GRACE_SECS ); // Dropping `handle` without joining detaches from the thread — the // consumer keeps running until its kernel call returns or the process // exits. This is the intentional "leak" documented in - // `finish_with_halt`'s contract. + // `finish_with_halt`'s contract; the `abandoned` flag bounds what the + // leaked thread is allowed to do to the output before it exits. drop(handle); Err(leak_err) } @@ -218,6 +232,18 @@ pub trait Sink: Send + 'static { pub struct Pipeline { tx: Sender, handle: JoinHandle>, + /// Set by [`finish_with_grace`] when the grace period expires and the + /// consumer thread is about to be leaked. The consumer holds a clone + /// and polls it in its drain loop and before `close()`: once set, it + /// stops applying further items and — crucially — does NOT call + /// `close()`. For sinks that finalise an output file in `close()` + /// (e.g. the mux MKV writer's Cues/segment-header patch), this + /// prevents a leaked consumer from finalising an output the caller + /// has already abandoned and reported as failed. It cannot interrupt + /// a syscall the consumer is currently wedged in, but it does bound + /// the damage to "whatever write is already in flight" once that + /// syscall returns, instead of running on to a clean finalise. + abandoned: Arc, } impl Pipeline { @@ -251,6 +277,8 @@ impl Pipeline { sink: S, ) -> Result { let (tx, rx) = bounded::(depth); + let abandoned = Arc::new(AtomicBool::new(false)); + let abandoned_consumer = abandoned.clone(); let handle = thread::Builder::new() .name(name.into()) .spawn(move || -> Result { @@ -264,6 +292,17 @@ impl Pipeline { tracing::debug!("Pipeline receive: item={}", std::any::type_name::()); } + // If the caller has abandoned us (grace period expired + // after a halt/timeout and the JoinHandle was dropped), + // stop applying items. We keep draining so the producer + // — if it is somehow still alive — never blocks on a + // dead receiver, but we touch the output no further. The + // final post-loop abandonment check returns the error + // and skips `close()`. + if abandoned_consumer.load(Ordering::Relaxed) { + continue; + } + if first_err.is_some() || stopped { // Drain remaining items so the producer never // blocks on a dead receiver. `apply` is not @@ -310,6 +349,20 @@ impl Pipeline { } } + // Final abandonment check: the common leak case is a + // consumer wedged inside `apply` (a blocking write + // syscall). When that syscall finally returns, the + // producer has long since dropped `tx`, so `recv` + // yields `Err` and we fall through to here. If the + // caller abandoned us in the meantime, skip `close()` + // entirely — finalising the output (e.g. writing the + // MKV Cues + patching the segment header) on a file the + // caller already reported as failed is exactly the + // write race we must not run. + if abandoned_consumer.load(Ordering::Relaxed) { + return Err(Error::Halted); + } + match first_err { Some(e) => Err(e), None => sink.close(), @@ -317,7 +370,11 @@ impl Pipeline { }) .map_err(|e| Error::IoError { source: e })?; - Ok(Pipeline { tx, handle }) + Ok(Pipeline { + tx, + handle, + abandoned, + }) } /// Push one item. Blocks if the channel is full — that's the @@ -459,7 +516,11 @@ impl Pipeline { /// logged at the join site (the library carries no English in its /// error values), so callers discriminate on the variant. pub fn finish(self) -> Result { - let Pipeline { tx, handle } = self; + let Pipeline { + tx, + handle, + abandoned: _, + } = self; // Explicit drop, although the destructure already drops `tx` // at end-of-scope. Being explicit keeps the intent obvious. drop(tx); @@ -495,7 +556,11 @@ impl Pipeline { /// halt-token plumbed through; that path still blocks indefinitely /// on `join()`, matching pre-0.20.8 behaviour. pub fn finish_with_halt(self, halt: Option<&Halt>) -> Result { - let Pipeline { tx, handle } = self; + let Pipeline { + tx, + handle, + abandoned, + } = self; drop(tx); let deadline = Instant::now() + Duration::from_secs(JOIN_TIMEOUT_SECS); loop { @@ -507,11 +572,11 @@ impl Pipeline { } if let Some(h) = halt { if h.is_cancelled() { - return finish_with_grace(handle, Error::Halted); + return finish_with_grace(handle, &abandoned, Error::Halted); } } if Instant::now() >= deadline { - return finish_with_grace(handle, Error::PipelineJoinTimeout); + return finish_with_grace(handle, &abandoned, Error::PipelineJoinTimeout); } thread::sleep(POLL_INTERVAL); } @@ -1326,6 +1391,147 @@ mod tests { ); } + /// Regression for the "leaked consumer finalises an abandoned output" + /// bug. When the grace period expires and the consumer is leaked, the + /// consumer — once its wedged `apply` syscall returns — must observe + /// the abandonment flag and exit WITHOUT calling `close()`. For the + /// mux writer, `close()` is where the MKV is finalised (Cues + + /// segment-header patch); running it on a file the caller already + /// reported as failed is the write race this fix prevents. + /// + /// Setup: a sink whose `apply` blocks until released (simulating a + /// wedged write syscall) and records whether `close()` ran. We fire + /// the halt, let `finish_with_halt` spin through the full grace period + /// and leak the thread, THEN release the wedged `apply`. The consumer + /// drains to EOF (tx already dropped) and must skip `close()`. + /// + /// Without the fix, the leaked consumer would fall through to + /// `sink.close()` and `closed` would flip to true. + #[test] + fn leaked_consumer_skips_close_after_abandonment() { + struct WedgeThenRecord { + release: Arc, + started: Arc, + closed: Arc, + } + impl Sink for WedgeThenRecord { + type Output = (); + fn apply(&mut self, _item: u64) -> Result { + self.started.store(true, Ordering::SeqCst); + while !self.release.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(20)); + } + Ok(Flow::Continue) + } + fn close(self) -> Result<(), Error> { + self.closed.store(true, Ordering::SeqCst); + Ok(()) + } + } + + let release = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let started = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let closed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let pipe = Pipeline::spawn( + DEFAULT_PIPELINE_DEPTH, + WedgeThenRecord { + release: release.clone(), + started: started.clone(), + closed: closed.clone(), + }, + ) + .expect("spawn"); + pipe.send(0u64) + .expect("seed the item the consumer wedges on"); + wait_for_started(&started, Duration::from_secs(2)); + + // Halt is already cancelled when finish_with_halt is called, so + // it enters the grace spin immediately; the consumer is wedged in + // apply for the whole grace window, so the thread is leaked. + let halt = crate::halt::Halt::new(); + halt.cancel(); + let res = pipe.finish_with_halt(Some(&halt)); + assert!( + matches!(res, Err(Error::Halted)), + "expected Err(Halted) after grace-expiry leak, got {res:?}" + ); + + // The thread is now leaked but still parked in apply. Release it + // and give it time to drain to EOF and reach the close() decision. + release.store(true, Ordering::SeqCst); + let deadline = Instant::now() + Duration::from_secs(2); + while Instant::now() < deadline && !closed.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(20)); + } + + assert!( + !closed.load(Ordering::SeqCst), + "abandoned consumer called close() — it must skip finalisation \ + of an output the caller already reported as failed" + ); + } + + /// Companion to the above: the abandonment guard must NOT fire on the + /// normal halt path where the consumer finishes inside the grace + /// window. There, `close()` runs and the output is finalised as + /// usual. (Covered for the value-return case by + /// `finish_with_halt_joins_cleanly_when_consumer_finishes_in_grace`; + /// this one asserts `close()` specifically ran.) + #[test] + fn consumer_finishing_in_grace_still_calls_close() { + struct ReleasableClose { + release: Arc, + started: Arc, + closed: Arc, + } + impl Sink for ReleasableClose { + type Output = (); + fn apply(&mut self, _item: u64) -> Result { + self.started.store(true, Ordering::SeqCst); + while !self.release.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(20)); + } + Ok(Flow::Continue) + } + fn close(self) -> Result<(), Error> { + self.closed.store(true, Ordering::SeqCst); + Ok(()) + } + } + + let release = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let started = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let closed = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let pipe = Pipeline::spawn( + DEFAULT_PIPELINE_DEPTH, + ReleasableClose { + release: release.clone(), + started: started.clone(), + closed: closed.clone(), + }, + ) + .expect("spawn"); + pipe.send(0u64).expect("seed"); + wait_for_started(&started, Duration::from_secs(2)); + + // Release the consumer almost immediately — well inside the grace + // window — so finish_with_halt joins cleanly and close() runs. + let release2 = release.clone(); + std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(100)); + release2.store(true, Ordering::SeqCst); + }); + + let halt = crate::halt::Halt::new(); + halt.cancel(); + let res = pipe.finish_with_halt(Some(&halt)); + assert!(res.is_ok(), "expected clean Ok join in grace, got {res:?}"); + assert!( + closed.load(Ordering::SeqCst), + "consumer that finished inside grace must have called close()" + ); + } + /// Regression: `finish_with_halt` with no halt token and a consumer /// that completes normally must still return `Ok` (the None-halt /// polling path is unchanged by the grace-period fix). This is the