algeon-query-runtime 0.1.0

//! Deferred service-stream cleanup supervisor and its fail-closed accounting.
//!
//! A leased runtime cannot return to the pool until `cudaStreamSynchronize`
//! has confirmed that every operation enqueued on its stream has completed;
//! otherwise the next lease holder could overwrite buffers the previous
//! holder's kernels still read. That synchronize blocks the host, and leases
//! are dropped on host-runtime tasks, GPU pool workers, and inside other
//! `Drop` impls, none of which may block on the device. Each pool therefore
//! owns one thread, `algeon-stream-cleanup-d{device}-p{pool}`, that receives
//! `StreamCleanupJob`s over a channel, synchronizes, and either idles or
//! destroys the runtime. The release row of the admission table in
//! `website/docs/design/admission.mdx#from-capability-validation-to-admission`
//! names this thread as the place where `cudaStreamSynchronize` runs for each
//! leased stream.
//!
//! Fail-closed means every path that cannot prove a stream was synchronized
//! sets the pool's `closed` flag and destroys the idle set: a failed
//! synchronize, a cleanup job dropped unrun because the thread panicked, and
//! an enqueue that finds the thread gone. A closed pool rejects new leases
//! and destroys each returning runtime, so an unsynchronized stream is never
//! reused.
//!
//! The thread's outcome is not inferred from pool counters. An owner requests
//! shutdown, one caller is elected to join the thread, and that joiner
//! commits the terminal state that every later caller observes. The
//! process-exit hook joins the thread in `ProcessExitPhase::StreamPool`, the
//! last phase, because supervised workers and pool jobs still hold leases
//! during the earlier phases.

use std::{
    any::Any,
    panic::{AssertUnwindSafe, catch_unwind},
    sync::{
        Arc, Condvar, Mutex, OnceLock, PoisonError, Weak,
        atomic::{AtomicBool, AtomicU64, Ordering},
        mpsc,
    },
    thread,
    time::Instant,
};
#[cfg(test)]
use std::{cell::RefCell, sync::Barrier};

use query_planning::error::{QueryError, QueryErrorCode, Result};

use super::{
    context::QueryServiceResourceEpoch,
    process_exit::{
        ProcessExitPhase, ProcessExitRegistration, ProcessExitShutdown, register_cuda_process_exit_shutdown,
    },
    service_stream::{ServiceStreamPool, ServiceStreamPoolInner, StreamPermitClass},
};

pub(super) const STREAM_CLEANUP_SUPERVISOR_BUILD_OPERATION: &str = "query_engine.service_stream_pool.build";

type StreamCleanupCallback = Box<dyn FnOnce(StreamCleanupOutcome) + Send>;

/// Fault-injection hooks for cleanup-supervisor unit tests. Outside
/// `cfg(test)` the slot carries no state and every hook is a constant no-op.
#[derive(Clone, Default)]
pub(super) struct ServiceStreamTestProbeSlot {
    #[cfg(test)]
    probe: Arc<Mutex<Option<Arc<TestServiceStreamProbe>>>>,
}

impl std::fmt::Debug for ServiceStreamTestProbeSlot {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.debug_struct("ServiceStreamTestProbeSlot").finish_non_exhaustive()
    }
}

#[cfg(not(test))]
impl ServiceStreamTestProbeSlot {
    fn fail_next_supervisor_spawn(&self) -> bool {
        false
    }
    fn force_next_cleanup_enqueue_fallback(&self) -> bool {
        false
    }
    fn force_next_synchronization_failure(&self) -> bool {
        false
    }
    fn pause_before_worker_outcome(&self) {}
    fn panic_after_worker_outcome(&self) -> bool {
        false
    }
    fn notify_shutdown_send_failure(&self) {}
    fn notify_join_waiter(&self) {}
}

#[cfg(test)]
impl ServiceStreamTestProbeSlot {
    fn current(&self) -> Option<Arc<TestServiceStreamProbe>> {
        self.probe.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone()
    }

    fn fail_next_supervisor_spawn(&self) -> bool {
        self.current().is_some_and(|probe| probe.fail_next_supervisor_spawn())
    }
    fn force_next_cleanup_enqueue_fallback(&self) -> bool {
        self.current().is_some_and(|probe| probe.force_next_cleanup_enqueue_fallback())
    }
    fn force_next_synchronization_failure(&self) -> bool {
        self.current().is_some_and(|probe| probe.force_next_synchronization_failure())
    }
    fn pause_before_worker_outcome(&self) {
        if let Some(probe) = self.current() {
            probe.pause_before_worker_outcome();
        }
    }
    fn panic_after_worker_outcome(&self) -> bool {
        self.current().is_some_and(|probe| probe.panic_after_worker_outcome())
    }
    fn notify_shutdown_send_failure(&self) {
        if let Some(probe) = self.current() {
            probe.notify_shutdown_send_failure();
        }
    }
    fn notify_join_waiter(&self) {
        if let Some(probe) = self.current() {
            probe.notify_join_waiter();
        }
    }

    pub(super) fn with_test_probe<T>(&self, action: impl FnOnce(&TestServiceStreamProbe) -> T) -> T {
        let probe =
            Arc::clone(self.probe.lock().unwrap_or_else(std::sync::PoisonError::into_inner).get_or_insert_default());
        action(&probe)
    }

    pub(super) fn install_for_next_pool(probe: Arc<TestServiceStreamProbe>) {
        NEXT_POOL_TEST_PROBE.with(|slot| *slot.borrow_mut() = Some(probe));
    }

    pub(super) fn install_next_pool_probe(&self) {
        let probe = NEXT_POOL_TEST_PROBE.with(|slot| slot.borrow_mut().take());
        if probe.is_some() {
            *self.probe.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = probe;
        }
    }
}

#[cfg(test)]
thread_local! {
    static NEXT_POOL_TEST_PROBE: RefCell<Option<Arc<TestServiceStreamProbe>>> = const { RefCell::new(None) };
}

#[cfg(test)]
#[derive(Default)]
pub(super) struct TestServiceStreamProbe {
    pub(super) fail_next_supervisor_spawn: AtomicBool,
    pub(super) force_next_cleanup_enqueue_fallback: AtomicBool,
    pub(super) force_next_synchronization_failure: AtomicBool,
    pub(super) panic_after_worker_outcome: AtomicBool,
    pub(super) worker_outcome_pause: Mutex<Option<StreamCleanupWorkerOutcomePause>>,
    pub(super) shutdown_send_failed: Mutex<Option<mpsc::Sender<()>>>,
    pub(super) join_waiter: Mutex<Option<mpsc::Sender<()>>>,
}

#[cfg(test)]
impl TestServiceStreamProbe {
    fn fail_next_supervisor_spawn(&self) -> bool {
        self.fail_next_supervisor_spawn.swap(false, Ordering::AcqRel)
    }
    fn force_next_cleanup_enqueue_fallback(&self) -> bool {
        self.force_next_cleanup_enqueue_fallback.swap(false, Ordering::AcqRel)
    }
    fn force_next_synchronization_failure(&self) -> bool {
        self.force_next_synchronization_failure.swap(false, Ordering::AcqRel)
    }
    fn pause_before_worker_outcome(&self) {
        if let Some(pause) = self.worker_outcome_pause.lock().unwrap().clone() {
            pause.arrived.wait();
            pause.release.wait();
        }
    }
    fn panic_after_worker_outcome(&self) -> bool {
        self.panic_after_worker_outcome.load(Ordering::Acquire)
    }
    fn notify_shutdown_send_failure(&self) {
        if let Some(sender) = self.shutdown_send_failed.lock().unwrap().take() {
            let _ = sender.send(());
        }
    }
    fn notify_join_waiter(&self) {
        if let Some(sender) = self.join_waiter.lock().unwrap().take() {
            let _ = sender.send(());
        }
    }
}
type StreamCleanupTailCallback = Box<dyn FnOnce() + Send>;

/// Typed evidence handed to the pre-quiescence cleanup callback.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum StreamCleanupOutcome {
    /// The stream synchronized; the runtime was idled or, on a closed pool,
    /// destroyed.
    Completed,
    /// The stream did not synchronize. The runtime was destroyed and the pool
    /// closed; the attempt's release commit records a cleanup failure.
    StreamSynchronizationFailed,
    /// The callback runs from a `Drop`, because the cleanup job was never
    /// executed or the completion was rejected. Whether the stream
    /// synchronized is unknown, and the pool is closed.
    CallbackAbandoned,
}

impl StreamCleanupOutcome {
    fn from_synchronized(synchronized: bool) -> Self {
        if synchronized { Self::Completed } else { Self::StreamSynchronizationFailed }
    }
}

/// The two callbacks a lease owner installs for the end of its cleanup.
///
/// Each runs at most once. `before_quiescence` runs while the cleanup is still
/// counted in `active_cleanups`; `after_quiescence` runs after the counters
/// are released. Dropping an instance with either callback still present runs
/// them under the abandonment protocol, so a payload that never reached the
/// cleanup thread still completes its handoff.
#[derive(Default)]
pub(super) struct StreamCleanupCompletion {
    before_quiescence: Option<StreamCleanupCallback>,
    after_quiescence: Option<StreamCleanupTailCallback>,
}

impl StreamCleanupCompletion {
    pub(super) fn new(
        before_quiescence: impl FnOnce(StreamCleanupOutcome) + Send + 'static,
        after_quiescence: impl FnOnce() + Send + 'static,
    ) -> Self {
        Self {
            before_quiescence: Some(Box::new(before_quiescence)),
            after_quiescence: Some(Box::new(after_quiescence)),
        }
    }

    fn run_before_quiescence(&mut self, outcome: StreamCleanupOutcome) {
        if let Some(completion) = self.before_quiescence.take() {
            completion(outcome);
        }
    }

    fn run_after_quiescence(&mut self) {
        if let Some(completion) = self.after_quiescence.take() {
            completion();
        }
    }

    /// The receiver can discard queued cleanup jobs when a prior job panics.
    /// Complete every remaining handoff while unwinding so that the typed
    /// resource and observation permits cannot be stranded. Callback panics
    /// are intentionally suppressed here. The originating supervisor failure
    /// remains the fail-closed result, and a second unwind from `Drop` would
    /// abort the process.
    fn run_remaining_after_abandonment(&mut self) {
        if let Some(completion) = self.before_quiescence.take() {
            let _ = catch_unwind(AssertUnwindSafe(|| completion(StreamCleanupOutcome::CallbackAbandoned)));
        }
        if let Some(completion) = self.after_quiescence.take() {
            let _ = catch_unwind(AssertUnwindSafe(completion));
        }
    }
}

impl Drop for StreamCleanupCompletion {
    fn drop(&mut self) {
        self.run_remaining_after_abandonment();
    }
}

/// The pool-owned receiver thread which serializes CUDA stream cleanup.
///
/// This is deliberately separate from external completion delivery. Its only
/// responsibilities are synchronization, CUDA runtime destruction, and pool
/// accounting. A clean shutdown requires the owner to request shutdown first,
/// observe this thread's terminal state, and then reclaim its join handle.
pub(super) struct StreamCleanupSupervisor {
    pub(super) device: cudf::DeviceId,
    pub(super) pool_id: u64,
    pub(super) probe: ServiceStreamTestProbeSlot,
    /// The receiver is owned by the worker thread and is dropped there; that
    /// drop runs the abandonment protocol on queued jobs.
    pub(super) sender: mpsc::Sender<StreamCleanupJob>,
    /// Serializes every send with the shutdown request and with the
    /// process-exit flag check, so a cleanup job is either queued ahead of
    /// the `Shutdown` message or returned to the caller, never queued behind
    /// a shutdown the worker has already consumed.
    pub(super) submission_gate: Mutex<()>,
    pub(super) control: Arc<StreamCleanupSupervisorControl>,
    /// Consumed by the elected joiner or by the process-exit hook. `None`
    /// afterwards, and also while a self-join was refused.
    pub(super) join: Mutex<Option<thread::JoinHandle<StreamCleanupWorkerOutcome>>>,
    /// One of the `STREAM_CLEANUP_SUPERVISOR_*` constants; recorded because a
    /// `JoinHandle` can be joined only once.
    pub(super) join_disposition: AtomicU64,
    /// Present from `attach_pool` until the worker thread finishes or a join
    /// succeeds. The worker removes it itself on exit, so a thread that ends
    /// without any joiner does not stay in the process-exit snapshot.
    pub(super) process_exit_registration: Mutex<Option<ProcessExitRegistration>>,
    /// The owning pool, whose close and quiescence waiters wake when this
    /// supervisor commits its terminal state.
    pool: OnceLock<Weak<ServiceStreamPoolInner>>,
}

/// Why a cleanup job could not be queued; the job comes back so the caller
/// decides its fate. `ProcessExit` jobs are leaked, `SupervisorStopped` jobs
/// run synchronously on the caller's thread.
pub(super) enum StreamCleanupEnqueueError {
    ProcessExit(StreamCleanupJob),
    SupervisorStopped(StreamCleanupJob),
}

/// A process-exit owner deliberately retains only the cleanup-thread control
/// plane. It keeps no CUDA runtime or pool state alive, but it can upgrade a
/// still-live pool to close it before joining the worker. If the final pool
/// reference is released by the worker itself, the supervisor remains
/// registered long enough for the process hook to join that last thread.
struct ProcessExitStreamPool {
    inner: Weak<ServiceStreamPoolInner>,
    supervisor: Arc<StreamCleanupSupervisor>,
}

pub(super) const STREAM_CLEANUP_SUPERVISOR_NOT_JOINED: u64 = 0;
pub(super) const STREAM_CLEANUP_SUPERVISOR_JOINED: u64 = 1;
pub(super) const STREAM_CLEANUP_SUPERVISOR_JOIN_FAILED: u64 = 2;

/// Shutdown handshake between pool owners and the cleanup thread.
///
/// Lock order: a pool waiter holds the pool mutex and then takes `state`
/// (through `ensure_not_failed`); the joiner therefore releases `state`
/// before it takes the pool mutex in `notify_pool_waiters`.
#[derive(Debug, Default)]
pub(super) struct StreamCleanupSupervisorControl {
    pub(super) state: Mutex<StreamCleanupSupervisorState>,
    pub(super) changed: Condvar,
}

#[derive(Debug, Default)]
pub(super) struct StreamCleanupSupervisorState {
    /// Set once. A worker that exits before this is set is reported as
    /// `Unexpected`, because only shutdown is allowed to end the thread.
    pub(super) shutdown_requested: bool,
    /// Election flag; one caller at a time joins the thread. A joiner that
    /// times out before the join clears it so another caller can retry.
    pub(super) joiner_active: bool,
    /// Set once by the elected joiner. Every later wait or health check
    /// replays it.
    pub(super) terminal: Option<StreamCleanupSupervisorTerminal>,
}

/// What the worker thread itself returns. `Clean` means the executor saw the
/// `Shutdown` message; `Unexpected` means its receiver disconnected first.
#[derive(Debug, Clone)]
pub(super) enum StreamCleanupWorkerOutcome {
    Clean,
    Panicked { panic_message: String },
    Unexpected { detail: &'static str },
}

/// The committed terminal state. It differs from the worker outcome in two
/// ways: a `Clean` worker exit without a prior shutdown request becomes
/// `Unexpected`, and a panic carries the `stage` it happened in,
/// `cleanup_executor` for a callback or cleanup panic inside the loop and
/// `thread_join` for a panic after the worker produced its outcome.
#[derive(Debug, Clone)]
pub(super) enum StreamCleanupSupervisorTerminal {
    Clean,
    Panicked { stage: &'static str, panic_message: String },
    Unexpected { detail: &'static str },
}

#[cfg(test)]
#[derive(Debug, Clone)]
pub(super) struct StreamCleanupWorkerOutcomePause {
    pub(super) arrived: Arc<Barrier>,
    pub(super) release: Arc<Barrier>,
}

pub(super) enum StreamCleanupExecutorExit {
    Shutdown,
    ReceiverDisconnected,
    // Only test control jobs request an unexpected exit; see the `#[cfg(test)]`
    // `StreamCleanupControlJob` implementations below.
    #[cfg(test)]
    Unexpected(&'static str),
}

impl std::fmt::Debug for StreamCleanupSupervisor {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("StreamCleanupSupervisor")
            .field("device_ordinal", &self.device.ordinal())
            .field("pool_id", &self.pool_id)
            .field("join_disposition", &self.join_disposition.load(Ordering::Acquire))
            .finish_non_exhaustive()
    }
}

impl StreamCleanupSupervisor {
    /// Spawns the cleanup thread. The returned supervisor is not yet bound to
    /// a pool or registered for process exit; `attach_pool` does both once
    /// the pool's `Arc` exists. A spawn failure is an `Internal` error with
    /// reason `stream_cleanup_supervisor_spawn_failed` and no thread to join.
    pub(super) fn start(device: cudf::DeviceId, pool_id: u64, probe: ServiceStreamTestProbeSlot) -> Result<Arc<Self>> {
        let (sender, receiver) = mpsc::channel();
        let control = Arc::new(StreamCleanupSupervisorControl::default());
        let supervisor = Arc::new(Self {
            device,
            pool_id,
            probe: probe.clone(),
            sender,
            submission_gate: Mutex::new(()),
            control,
            join: Mutex::new(None),
            join_disposition: AtomicU64::new(STREAM_CLEANUP_SUPERVISOR_NOT_JOINED),
            process_exit_registration: Mutex::new(None),
            pool: OnceLock::new(),
        });
        let thread_control = Arc::clone(&supervisor.control);
        let thread_probe = probe.clone();
        let worker_supervisor = Arc::clone(&supervisor);
        let join = if probe.fail_next_supervisor_spawn() {
            Err(std::io::Error::other("injected service stream cleanup supervisor spawn failure"))
        } else {
            thread::Builder::new().name(format!("algeon-stream-cleanup-d{}-p{pool_id}", device.ordinal())).spawn(
                move || {
                    let outcome = run_stream_cleanup_worker(receiver, &thread_control, &thread_probe);
                    worker_supervisor.unregister_process_exit();
                    outcome
                },
            )
        }
        .map_err(|error| {
            QueryError::new(
                QueryErrorCode::Internal,
                STREAM_CLEANUP_SUPERVISOR_BUILD_OPERATION,
                "failed to start the service stream cleanup supervisor",
            )
            .with_fact("device_ordinal", device.ordinal())
            .with_fact("stream_pool_id", pool_id)
            .with_fact("reason", "stream_cleanup_supervisor_spawn_failed")
            .with_source(error)
        })?;

        *supervisor.join.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = Some(join);
        Ok(supervisor)
    }

    /// Binds the supervisor to its pool: the pool's waiters wake on terminal
    /// commitment, and the process-exit hook can close the pool before
    /// joining this worker.
    pub(super) fn attach_pool(self: &Arc<Self>, inner: Weak<ServiceStreamPoolInner>) {
        let _ = self.pool.set(Weak::clone(&inner));
        let shutdown: Arc<dyn ProcessExitShutdown> =
            Arc::new(ProcessExitStreamPool { inner, supervisor: Arc::clone(self) });
        let registration = register_cuda_process_exit_shutdown(shutdown);
        let mut slot = self.process_exit_registration.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        *slot = registration;
    }

    fn unregister_process_exit(&self) {
        let registration =
            self.process_exit_registration.lock().unwrap_or_else(std::sync::PoisonError::into_inner).take();
        if let Some(registration) = registration {
            registration.unregister();
        }
    }

    #[cfg(test)]
    pub(super) fn enqueue(&self, job: StreamCleanupJob) -> std::result::Result<(), mpsc::SendError<StreamCleanupJob>> {
        let _submission = self.submission_gate.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        let state = self.control.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if state.shutdown_requested || state.terminal.is_some() {
            return Err(mpsc::SendError(job));
        }
        // A sender failure only says that the receiver has gone away. The
        // worker's joined outcome remains the sole source of terminal state.
        self.sender.send(job)
    }

    /// Queues a cleanup job unless process exit or shutdown has begun.
    ///
    /// Both flags are read under `submission_gate`, the same lock the
    /// process-exit hook and `request_shutdown` hold while they set them, so
    /// a job is either queued before the worker can see `Shutdown` or
    /// returned in the error for the caller to leak or run inline.
    pub(super) fn enqueue_before_process_exit(
        &self,
        job: StreamCleanupJob,
        process_exit_started: &AtomicBool,
    ) -> std::result::Result<(), StreamCleanupEnqueueError> {
        let _submission = self.submission_gate.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if process_exit_started.load(Ordering::Acquire) {
            return Err(StreamCleanupEnqueueError::ProcessExit(job));
        }
        let state = self.control.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if state.shutdown_requested || state.terminal.is_some() {
            return Err(StreamCleanupEnqueueError::SupervisorStopped(job));
        }
        self.sender.send(job).map_err(|error| StreamCleanupEnqueueError::SupervisorStopped(error.0))
    }

    /// Queues the `Shutdown` message behind every cleanup already enqueued,
    /// so the worker drains them before it exits. Idempotent. A failed send
    /// means the receiver is already gone; that is not a terminal state by
    /// itself, the join classifies it.
    pub(super) fn request_shutdown(&self, deadline: Instant, operation: &'static str) -> Result<()> {
        let _submission = self.lock_submission_until(deadline, operation)?;
        let mut state = self.lock_control_until(deadline, operation)?;
        if state.shutdown_requested || state.terminal.is_some() {
            return Ok(());
        }
        state.shutdown_requested = true;
        let send_failed = self.sender.send(StreamCleanupJob::Shutdown).is_err();
        // The test probe may rendezvous with another thread that needs `state`.
        drop(state);
        if send_failed {
            self.probe.notify_shutdown_send_failure();
        }
        Ok(())
    }

    fn request_shutdown_from_drop(&self) {
        let _submission = self.submission_gate.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut state = self.control.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if state.shutdown_requested || state.terminal.is_some() {
            return;
        }
        state.shutdown_requested = true;
        let _ = self.sender.send(StreamCleanupJob::Shutdown);
    }

    pub(super) fn ensure_not_failed(&self, operation: &'static str) -> Result<()> {
        let state = self
            .control
            .state
            .lock()
            .map_err(|_| stream_cleanup_supervisor_state_poisoned(operation, self.device, self.pool_id))?;
        match state.terminal.as_ref() {
            None => Ok(()),
            Some(terminal) => self.terminal_result(operation, terminal),
        }
    }

    /// Commit an already-finished worker before an idle/quiescence caller
    /// reports success. A healthy worker remains untouched. Only a finished
    /// handle or a join already in progress can enter the terminal path.
    #[cfg(test)]
    pub(super) fn finalize_if_finished_until(&self, deadline: Instant, operation: &'static str) -> Result<()> {
        let mut state = self.lock_control_until(deadline, operation)?;
        loop {
            if let Some(terminal) = state.terminal.as_ref() {
                return self.terminal_result(operation, terminal);
            }
            if state.joiner_active {
                let remaining = deadline.saturating_duration_since(Instant::now());
                if remaining.is_zero() {
                    return Err(stream_cleanup_supervisor_join_timeout(operation, self.device, self.pool_id));
                }
                let (next, timeout) = self
                    .control
                    .changed
                    .wait_timeout(state, remaining)
                    .map_err(|_| stream_cleanup_supervisor_state_poisoned(operation, self.device, self.pool_id))?;
                state = next;
                if timeout.timed_out() && state.terminal.is_none() {
                    return Err(stream_cleanup_supervisor_join_timeout(operation, self.device, self.pool_id));
                }
                continue;
            }

            // Holding control state while the join handle is inspected prevents a
            // competing caller from electing itself between the finished
            // observation and this caller's election. The joiner never holds
            // the join mutex while acquiring control state.
            let finished = {
                let join = self.lock_join_until(deadline, operation)?;
                join.as_ref().map(thread::JoinHandle::is_finished).ok_or_else(|| {
                    QueryError::new(
                        QueryErrorCode::Internal,
                        operation,
                        "service stream cleanup supervisor join handle was unavailable before terminal commitment",
                    )
                    .with_fact("device_ordinal", self.device.ordinal())
                    .with_fact("stream_pool_id", self.pool_id)
                    .with_fact("reason", "stream_cleanup_supervisor_join_unavailable")
                })?
            };
            if !finished {
                return Ok(());
            }

            state.joiner_active = true;
            drop(state);
            return self.join_and_commit_terminal_until(deadline, operation);
        }
    }

    /// Elect one caller to join the worker and commit its terminal state. A
    /// worker outcome is not observable until the receiver and every queued
    /// cleanup job have been dropped and the elected caller has joined it.
    pub(super) fn wait_for_exit_until(&self, deadline: Instant, operation: &'static str) -> Result<()> {
        let mut state = self.lock_control_until(deadline, operation)?;
        loop {
            if let Some(terminal) = state.terminal.as_ref() {
                return self.terminal_result(operation, terminal);
            }
            if !state.joiner_active {
                state.joiner_active = true;
                // The join path relocks `state` to commit; the worker thread
                // never takes it, so the join itself needs nothing held.
                drop(state);
                return self.join_and_commit_terminal_until(deadline, operation);
            }

            self.probe.notify_join_waiter();
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(stream_cleanup_supervisor_join_timeout(operation, self.device, self.pool_id));
            }
            let (next, timeout) = self
                .control
                .changed
                .wait_timeout(state, remaining)
                .map_err(|_| stream_cleanup_supervisor_state_poisoned(operation, self.device, self.pool_id))?;
            state = next;
            if timeout.timed_out() && state.terminal.is_none() {
                return Err(stream_cleanup_supervisor_join_timeout(operation, self.device, self.pool_id));
            }
        }
    }

    fn join_and_commit_terminal_until(&self, deadline: Instant, operation: &'static str) -> Result<()> {
        let joined = self.join_worker_until(deadline, operation);
        let joined = match joined {
            Ok(joined) => joined,
            Err(error) => {
                self.relinquish_joiner();
                return Err(error);
            },
        };

        // Joining establishes the happens-before edge from receiver teardown,
        // in-flight recovery, and queued-job recovery to terminal commitment.
        let mut state = self.control.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        debug_assert!(state.terminal.is_none(), "only the elected joiner may commit a cleanup terminal state");
        let terminal = match joined {
            Ok(outcome) => Self::terminal_from_worker_outcome(outcome, state.shutdown_requested),
            Err(panic) => StreamCleanupSupervisorTerminal::Panicked {
                stage: "thread_join",
                panic_message: stream_cleanup_supervisor_panic_message(panic.as_ref()),
            },
        };
        state.terminal = Some(terminal.clone());
        state.joiner_active = false;
        self.control.changed.notify_all();
        // Lock order: pool waiters hold the pool mutex while they take
        // `state`, and `notify_pool_waiters` takes the pool mutex.
        drop(state);
        self.notify_pool_waiters();
        self.terminal_result(operation, &terminal)
    }

    /// Wakes the pool's close and quiescence waiters, which check the terminal
    /// state while holding the pool mutex; taking that mutex first keeps the
    /// wakeup from landing between such a check and its wait.
    fn notify_pool_waiters(&self) {
        if let Some(pool) = self.pool.get().and_then(Weak::upgrade) {
            let _state = pool.state.lock().unwrap_or_else(PoisonError::into_inner);
            pool.state_changed.notify_all();
        }
    }

    fn join_worker_until(
        &self,
        deadline: Instant,
        operation: &'static str,
    ) -> Result<std::thread::Result<StreamCleanupWorkerOutcome>> {
        let mut join = self.lock_join_until(deadline, operation)?;
        // `JoinHandle::join` has no timeout, so wait for `is_finished` under
        // the caller's deadline first and only then join, which then cannot
        // block for longer than the thread's final unwind.
        if let Some(handle) = join.as_ref() {
            while !handle.is_finished() {
                if Instant::now() >= deadline {
                    return Err(stream_cleanup_supervisor_join_timeout(operation, self.device, self.pool_id));
                }
                thread::yield_now();
            }
        }
        let handle = join.take().ok_or_else(|| {
            QueryError::new(
                QueryErrorCode::Internal,
                operation,
                "service stream cleanup supervisor join handle was unavailable before terminal commitment",
            )
            .with_fact("device_ordinal", self.device.ordinal())
            .with_fact("stream_pool_id", self.pool_id)
            .with_fact("reason", "stream_cleanup_supervisor_join_unavailable")
        })?;
        let joined = handle.join();
        self.join_disposition.store(
            if joined.is_ok() { STREAM_CLEANUP_SUPERVISOR_JOINED } else { STREAM_CLEANUP_SUPERVISOR_JOIN_FAILED },
            Ordering::Release,
        );
        self.unregister_process_exit();
        Ok(joined)
    }

    pub(super) fn join_for_process_exit(&self) {
        self.request_shutdown_from_drop();
        let mut join = self.join.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        let Some(handle) = join.take() else {
            return;
        };
        if handle.thread().id() == thread::current().id() {
            // The worker will finish its own current cleanup and then drop
            // this handle while unwinding out of its entry point. Joining
            // oneself would deadlock and cannot add a happens-before edge.
            *join = Some(handle);
            return;
        }
        let joined = handle.join();
        self.join_disposition.store(
            if joined.is_ok() { STREAM_CLEANUP_SUPERVISOR_JOINED } else { STREAM_CLEANUP_SUPERVISOR_JOIN_FAILED },
            Ordering::Release,
        );
        // `unregister_process_exit` takes the registry lock the exit hook
        // holds while iterating registrations; do not hold the join lock
        // across it.
        drop(join);
        self.unregister_process_exit();
    }

    fn relinquish_joiner(&self) {
        let mut state = self.control.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
        if state.terminal.is_none() {
            state.joiner_active = false;
            self.control.changed.notify_all();
        }
    }

    /// A `Clean` worker exit is only clean if an owner asked for it. The
    /// receiver disconnecting on its own means every sender was dropped
    /// while the supervisor state still expected a running thread.
    fn terminal_from_worker_outcome(
        outcome: StreamCleanupWorkerOutcome,
        shutdown_requested: bool,
    ) -> StreamCleanupSupervisorTerminal {
        match outcome {
            StreamCleanupWorkerOutcome::Clean if shutdown_requested => StreamCleanupSupervisorTerminal::Clean,
            StreamCleanupWorkerOutcome::Clean => StreamCleanupSupervisorTerminal::Unexpected {
                detail: "cleanup supervisor exited before explicit shutdown",
            },
            StreamCleanupWorkerOutcome::Panicked { panic_message } => {
                StreamCleanupSupervisorTerminal::Panicked { stage: "cleanup_executor", panic_message }
            },
            StreamCleanupWorkerOutcome::Unexpected { detail } => StreamCleanupSupervisorTerminal::Unexpected { detail },
        }
    }

    fn terminal_result(&self, operation: &'static str, terminal: &StreamCleanupSupervisorTerminal) -> Result<()> {
        match terminal {
            StreamCleanupSupervisorTerminal::Clean => Ok(()),
            StreamCleanupSupervisorTerminal::Panicked { stage, panic_message } => {
                Err(stream_cleanup_supervisor_panic_error(operation, self.device, self.pool_id, stage, panic_message))
            },
            StreamCleanupSupervisorTerminal::Unexpected { detail } => {
                Err(stream_cleanup_supervisor_unexpected_exit_error(operation, self.device, self.pool_id, detail))
            },
        }
    }

    fn lock_submission_until(
        &self,
        deadline: Instant,
        operation: &'static str,
    ) -> Result<std::sync::MutexGuard<'_, ()>> {
        loop {
            match self.submission_gate.try_lock() {
                Ok(gate) => return Ok(gate),
                Err(std::sync::TryLockError::Poisoned(_)) => {
                    return Err(stream_cleanup_supervisor_state_poisoned(operation, self.device, self.pool_id));
                },
                Err(std::sync::TryLockError::WouldBlock) if Instant::now() >= deadline => {
                    return Err(stream_cleanup_supervisor_exit_timeout(operation, self.device, self.pool_id));
                },
                Err(std::sync::TryLockError::WouldBlock) => thread::yield_now(),
            }
        }
    }

    fn lock_control_until(
        &self,
        deadline: Instant,
        operation: &'static str,
    ) -> Result<std::sync::MutexGuard<'_, StreamCleanupSupervisorState>> {
        loop {
            match self.control.state.try_lock() {
                Ok(state) => return Ok(state),
                Err(std::sync::TryLockError::Poisoned(_)) => {
                    return Err(stream_cleanup_supervisor_state_poisoned(operation, self.device, self.pool_id));
                },
                Err(std::sync::TryLockError::WouldBlock) if Instant::now() >= deadline => {
                    return Err(stream_cleanup_supervisor_exit_timeout(operation, self.device, self.pool_id));
                },
                Err(std::sync::TryLockError::WouldBlock) => thread::yield_now(),
            }
        }
    }

    fn lock_join_until(
        &self,
        deadline: Instant,
        operation: &'static str,
    ) -> Result<std::sync::MutexGuard<'_, Option<thread::JoinHandle<StreamCleanupWorkerOutcome>>>> {
        loop {
            match self.join.try_lock() {
                Ok(join) => return Ok(join),
                Err(std::sync::TryLockError::Poisoned(_)) => {
                    return Err(stream_cleanup_supervisor_state_poisoned(operation, self.device, self.pool_id));
                },
                Err(std::sync::TryLockError::WouldBlock) if Instant::now() >= deadline => {
                    return Err(stream_cleanup_supervisor_join_timeout(operation, self.device, self.pool_id));
                },
                Err(std::sync::TryLockError::WouldBlock) => thread::yield_now(),
            }
        }
    }
}
impl Drop for ServiceStreamPoolInner {
    /// Runs when the last pool clone is gone. Leases hold pool clones, so no
    /// lease exists and no further cleanup can be enqueued; the `Shutdown`
    /// message queues behind the cleanups already queued, and the thread
    /// unregisters itself from the process-exit hook when it exits.
    fn drop(&mut self) {
        self.cleanup_supervisor.request_shutdown_from_drop();
    }
}

impl ProcessExitShutdown for ProcessExitStreamPool {
    fn phase(&self) -> ProcessExitPhase {
        ProcessExitPhase::StreamPool
    }

    fn shutdown_for_process_exit(&self) {
        if let Some(inner) = self.inner.upgrade() {
            ServiceStreamPool { inner }.close_and_join_for_process_exit();
        } else {
            self.supervisor.join_for_process_exit();
        }
    }
}

impl ServiceStreamPool {
    /// Passes a returning runtime to the cleanup thread. Called from
    /// `StreamLease::drop` and `StreamSlot::drop`, on whatever thread drops
    /// them; the only CUDA calls here are on the fallback paths, which run
    /// when the pool mutex is poisoned or the cleanup thread is gone.
    pub(super) fn enqueue_cleanup(
        &self,
        runtime: Arc<cudf::Runtime>,
        resource_epoch: Option<Arc<QueryServiceResourceEpoch>>,
        mut completion: Option<StreamCleanupCompletion>,
        permit_class: StreamPermitClass,
    ) {
        if self.inner.process_exit_started.load(Ordering::Acquire) {
            // The process-exit owner has already joined (or is joining) the
            // only thread allowed to destroy service runtimes. Leaking this
            // terminal payload is intentional. Process exit will reclaim the
            // address space, whereas dropping it could race libcudart's later
            // atexit handler.
            std::mem::forget(runtime);
            std::mem::forget(resource_epoch);
            std::mem::forget(completion);
            return;
        }
        let registered = match self.inner.state.lock() {
            Ok(mut state) => {
                state.pending_cleanups = state.pending_cleanups.saturating_add(1);
                self.inner.state_changed.notify_all();
                true
            },
            Err(_error) => {
                eprintln!("[query-engine] warning: failed to account deferred stream cleanup");
                false
            },
        };
        if !registered {
            let synchronized = runtime.synchronize().is_ok();
            // The runtime is destroyed before the epoch releases the service
            // generation its memory manager belongs to.
            drop(runtime);
            // Release the epoch before the completion callbacks report quiescence.
            drop(resource_epoch);
            if let Some(completion) = completion.as_mut() {
                completion.run_before_quiescence(StreamCleanupOutcome::from_synchronized(synchronized));
                completion.run_after_quiescence();
            }
            return;
        }
        let job = StreamCleanupJob::Runtime(DeferredStreamCleanup::new(
            self.clone(),
            runtime,
            resource_epoch,
            completion,
            permit_class,
        ));
        let force_enqueue_fallback = self.inner.cleanup_probe.force_next_cleanup_enqueue_fallback();
        if force_enqueue_fallback {
            match job {
                StreamCleanupJob::Runtime(mut cleanup) => cleanup.run(),
                StreamCleanupJob::Shutdown => unreachable!("runtime cleanup cannot enqueue supervisor shutdown"),
                #[cfg(test)]
                StreamCleanupJob::Control(_) => {
                    unreachable!("new cleanup job is always a runtime cleanup")
                },
            }
            return;
        }
        match self.inner.cleanup_supervisor.enqueue_before_process_exit(job, &self.inner.process_exit_started) {
            Ok(()) => {},
            Err(StreamCleanupEnqueueError::ProcessExit(job)) => {
                // The exit flag is checked while holding the same submission
                // gate used to queue supervisor shutdown. A cleanup therefore
                // either precedes shutdown or is retained across the terminal
                // boundary; it can never fall back to CUDA work after join.
                std::mem::forget(job);
            },
            Err(StreamCleanupEnqueueError::SupervisorStopped(job)) => {
                // The only production send failure outside process exit is a
                // terminal cleanup supervisor. The synchronous fallback still
                // releases the job's permits, but the runtime cannot be
                // returned to a pool whose owner has disappeared.
                self.inner.closed.store(true, Ordering::Release);
                match job {
                    StreamCleanupJob::Runtime(mut cleanup) => cleanup.run(),
                    StreamCleanupJob::Shutdown => unreachable!("runtime cleanup cannot enqueue supervisor shutdown"),
                    #[cfg(test)]
                    StreamCleanupJob::Control(_) => {},
                }
            },
        }
    }
}

/// Owns one queued stream cleanup until both resource accounting and the
/// external completion handoff have finished. If a prior cleanup panics, the
/// receiver drops queued jobs. This guard turns that drop into fail-closed
/// resource cleanup that releases the permits and completes the handoff.
pub(super) struct DeferredStreamCleanup {
    pool: ServiceStreamPool,
    runtime: Option<Arc<cudf::Runtime>>,
    resource_epoch: Option<Arc<QueryServiceResourceEpoch>>,
    completion: Option<StreamCleanupCompletion>,
    phase: DeferredStreamCleanupPhase,
    permit_class: StreamPermitClass,
}

/// Which pool counter currently accounts for this job, so the abandonment
/// recovery in `Drop` can release that one.
#[derive(Clone, Copy, PartialEq, Eq)]
enum DeferredStreamCleanupPhase {
    /// Counted in `pending_cleanups`.
    Pending,
    /// Counted in `active_cleanups`; the runtime may already be destroyed.
    Active,
    /// Counters released; only the `after_quiescence` callback remains.
    AfterAccounting,
    /// Nothing left for `Drop` to do.
    Complete,
}

impl DeferredStreamCleanup {
    fn new(
        pool: ServiceStreamPool,
        runtime: Arc<cudf::Runtime>,
        resource_epoch: Option<Arc<QueryServiceResourceEpoch>>,
        completion: Option<StreamCleanupCompletion>,
        permit_class: StreamPermitClass,
    ) -> Self {
        Self {
            pool,
            runtime: Some(runtime),
            resource_epoch,
            completion,
            phase: DeferredStreamCleanupPhase::Pending,
            permit_class,
        }
    }

    fn run(&mut self) {
        if !self.begin_accounting() {
            self.discard_after_accounting_failure();
            return;
        }

        let synchronized = self.synchronize_runtime();
        self.finish_runtime_accounting(synchronized);
        if let Some(completion) = self.completion.as_mut() {
            // This callback releases GPU-side/query admission ownership while
            // cleanup remains active. A panic is caught by the supervisor and
            // this guard's Drop path completes the remaining handoff.
            completion.run_before_quiescence(StreamCleanupOutcome::from_synchronized(synchronized));
        }
        self.finish_active_accounting();
        if let Some(completion) = self.completion.as_mut() {
            // The typed callback only enqueues to ObservationCompletionExecutor
            // and therefore cannot reenter backend close from this supervisor.
            completion.run_after_quiescence();
        }
        self.phase = DeferredStreamCleanupPhase::Complete;
    }

    fn begin_accounting(&mut self) -> bool {
        match self.pool.inner.state.lock() {
            Ok(mut state) => {
                state.pending_cleanups = state.pending_cleanups.saturating_sub(1);
                state.active_cleanups = state.active_cleanups.saturating_add(1);
                self.phase = DeferredStreamCleanupPhase::Active;
                self.pool.inner.state_changed.notify_all();
                true
            },
            Err(_error) => {
                eprintln!("[query-engine] warning: failed to start accounted stream cleanup");
                false
            },
        }
    }

    fn synchronize_runtime(&self) -> bool {
        let force_failure = self.pool.inner.cleanup_probe.force_next_synchronization_failure();
        match self.runtime.as_ref().expect("deferred cleanup runtime is present").synchronize() {
            Ok(()) if force_failure => {
                eprintln!(
                    "[query-engine] warning: injecting device {} stream synchronization failure for test",
                    self.pool.device().ordinal()
                );
                false
            },
            Ok(()) => true,
            Err(error) => {
                eprintln!(
                    "[query-engine] warning: failed to synchronize device {} service stream; discarding runtime: {error}",
                    self.pool.device().ordinal()
                );
                false
            },
        }
    }

    /// Releases the class permit and either idles or destroys the runtime.
    ///
    /// The class permit is released only here, after synchronization, so a
    /// new lease of the same class cannot start while the returned stream
    /// still has work queued. Waiters are not woken yet; that happens in
    /// `finish_active_accounting`, after the `before_quiescence` callback has
    /// released the attempt's capacity.
    fn finish_runtime_accounting(&mut self, synchronized: bool) {
        let mut idle_to_drop = Vec::new();
        match self.pool.inner.state.lock() {
            Ok(mut state) => {
                match self.permit_class {
                    StreamPermitClass::Root => state.root_outstanding = state.root_outstanding.saturating_sub(1),
                    StreamPermitClass::Auxiliary => state.aux_outstanding = state.aux_outstanding.saturating_sub(1),
                }
                if synchronized && !self.pool.inner.closed.load(Ordering::Acquire) {
                    state.idle.push(self.runtime.take().expect("deferred cleanup runtime is present"));
                } else {
                    state.created = state.created.saturating_sub(1);
                    if !synchronized {
                        // A stream that will not synchronize leaves the device
                        // state unknown. Close the pool and destroy every idle
                        // runtime as well; none of them can be proven safe.
                        self.pool.inner.closed.store(true, Ordering::Release);
                        state.synchronization_failures = state.synchronization_failures.saturating_add(1);
                        idle_to_drop = std::mem::take(&mut state.idle);
                        state.created = state.created.saturating_sub(idle_to_drop.len());
                    }
                }
            },
            Err(_error) => {
                eprintln!("[query-engine] warning: failed to finish accounted stream cleanup");
            },
        }
        // Taken now so the abandonment recovery in `Drop` finds no runtime to
        // synchronize or count a second time. Runtimes are destroyed outside
        // the pool mutex and before the epoch releases the service generation.
        drop(self.runtime.take());
        drop(idle_to_drop);
        // Release the epoch before `run` invokes the completion callbacks.
        drop(self.resource_epoch.take());
    }

    fn finish_active_accounting(&mut self) {
        match self.pool.inner.state.lock() {
            Ok(mut state) => {
                state.active_cleanups = state.active_cleanups.saturating_sub(1);
            },
            Err(_error) => {
                eprintln!("[query-engine] warning: failed to complete stream cleanup accounting");
            },
        }
        self.pool.inner.available.notify_waiters();
        self.pool.inner.state_changed.notify_all();
        self.phase = DeferredStreamCleanupPhase::AfterAccounting;
    }

    /// Path for a poisoned pool mutex: the counters cannot be trusted, so the
    /// runtime is synchronized on a best-effort basis and destroyed, and the
    /// completion is told the synchronization failed.
    fn discard_after_accounting_failure(&mut self) {
        if let Some(runtime) = self.runtime.take() {
            let _ = catch_unwind(AssertUnwindSafe(|| runtime.synchronize()));
            // Destroyed before the epoch releases the service generation.
            drop(runtime);
        }
        // Release the epoch before the completion callbacks below run.
        drop(self.resource_epoch.take());
        if let Some(completion) = self.completion.as_mut() {
            completion.run_before_quiescence(StreamCleanupOutcome::StreamSynchronizationFailed);
            completion.run_after_quiescence();
        }
        self.phase = DeferredStreamCleanupPhase::Complete;
    }

    /// Runs from `Drop` when `run` did not reach `Complete`: the receiver was
    /// dropped with this job still queued, or a callback panicked mid-run.
    /// It runs on the cleanup thread during that thread's unwind. The
    /// synchronize is wrapped in `catch_unwind` because a second panic during
    /// an unwind aborts the process.
    fn recover_after_abandonment(&mut self) {
        let had_runtime = self.runtime.is_some();
        if let Some(runtime) = self.runtime.take() {
            let _ = catch_unwind(AssertUnwindSafe(|| runtime.synchronize()));
            // Destroyed before the epoch releases the service generation.
            drop(runtime);
        }
        // Release the epoch before the completion callbacks below run.
        drop(self.resource_epoch.take());

        let mut idle_to_drop = Vec::new();
        match self.pool.inner.state.lock() {
            Ok(mut state) => {
                match self.phase {
                    DeferredStreamCleanupPhase::Pending => {
                        state.pending_cleanups = state.pending_cleanups.saturating_sub(1);
                        if had_runtime {
                            state.created = state.created.saturating_sub(1);
                        }
                    },
                    DeferredStreamCleanupPhase::Active => {
                        state.active_cleanups = state.active_cleanups.saturating_sub(1);
                        if had_runtime {
                            state.created = state.created.saturating_sub(1);
                        }
                    },
                    DeferredStreamCleanupPhase::AfterAccounting | DeferredStreamCleanupPhase::Complete => {},
                }
                // An abandoned job means the owning supervisor is terminally
                // unhealthy. Never reuse a runtime while its cleanup protocol
                // has been interrupted.
                self.pool.inner.closed.store(true, Ordering::Release);
                idle_to_drop = std::mem::take(&mut state.idle);
                state.created = state.created.saturating_sub(idle_to_drop.len());
            },
            Err(_error) => {
                eprintln!("[query-engine] warning: failed to recover abandoned stream cleanup accounting");
            },
        }
        // Stream destruction runs after the match arm released the pool mutex.
        drop(idle_to_drop);
        self.pool.inner.available.notify_waiters();
        self.pool.inner.state_changed.notify_all();
        if let Some(mut completion) = self.completion.take() {
            completion.run_remaining_after_abandonment();
        }
        self.phase = DeferredStreamCleanupPhase::Complete;
    }
}

impl Drop for DeferredStreamCleanup {
    fn drop(&mut self) {
        if self.phase != DeferredStreamCleanupPhase::Complete {
            self.recover_after_abandonment();
        }
    }
}
fn stream_cleanup_supervisor_exit_timeout(operation: &'static str, device: cudf::DeviceId, pool_id: u64) -> QueryError {
    QueryError::new(
        QueryErrorCode::ServiceOverloaded,
        operation,
        "timed out waiting for the service stream cleanup supervisor to exit",
    )
    .with_fact("device_ordinal", device.ordinal())
    .with_fact("stream_pool_id", pool_id)
    .with_fact("reason", "stream_cleanup_supervisor_exit_timeout")
    .with_cleanup_deadline_elapsed()
}

fn stream_cleanup_supervisor_join_timeout(operation: &'static str, device: cudf::DeviceId, pool_id: u64) -> QueryError {
    QueryError::new(
        QueryErrorCode::ServiceOverloaded,
        operation,
        "timed out waiting to join the service stream cleanup supervisor",
    )
    .with_fact("device_ordinal", device.ordinal())
    .with_fact("stream_pool_id", pool_id)
    .with_fact("reason", "stream_cleanup_supervisor_join_timeout")
    .with_cleanup_deadline_elapsed()
}

fn stream_cleanup_supervisor_state_poisoned(
    operation: &'static str,
    device: cudf::DeviceId,
    pool_id: u64,
) -> QueryError {
    QueryError::new(QueryErrorCode::Internal, operation, "service stream cleanup supervisor state was poisoned")
        .with_fact("device_ordinal", device.ordinal())
        .with_fact("stream_pool_id", pool_id)
        .with_fact("reason", "stream_cleanup_supervisor_state_poisoned")
}

fn stream_cleanup_supervisor_unexpected_exit_error(
    operation: &'static str,
    device: cudf::DeviceId,
    pool_id: u64,
    detail: &str,
) -> QueryError {
    QueryError::new(QueryErrorCode::Internal, operation, "service stream cleanup supervisor exited unexpectedly")
        .with_fact("device_ordinal", device.ordinal())
        .with_fact("stream_pool_id", pool_id)
        .with_fact("supervisor_detail", detail)
        .with_fact("reason", "stream_cleanup_supervisor_unexpected_exit")
}

fn stream_cleanup_supervisor_panic_error(
    operation: &'static str,
    device: cudf::DeviceId,
    pool_id: u64,
    stage: &'static str,
    panic_message: &str,
) -> QueryError {
    QueryError::new(QueryErrorCode::Internal, operation, "service stream cleanup supervisor panicked")
        .with_fact("device_ordinal", device.ordinal())
        .with_fact("stream_pool_id", pool_id)
        .with_fact("supervisor_stage", stage)
        .with_fact("panic_message", panic_message)
        .with_fact("reason", "stream_cleanup_supervisor_panicked")
}

fn stream_cleanup_supervisor_panic_message(panic: &(dyn Any + Send)) -> String {
    panic
        .downcast_ref::<&'static str>()
        .map(|message| (*message).to_string())
        .or_else(|| panic.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "non-string panic payload".to_string())
}

// Control jobs are a test-only probe channel. Production only ever enqueues
// runtime cleanups and shutdown.
#[cfg(test)]
pub(super) trait StreamCleanupControlJob: Send {
    fn run(self: Box<Self>) -> Option<StreamCleanupExecutorExit>;
}

pub(super) enum StreamCleanupJob {
    Runtime(DeferredStreamCleanup),
    Shutdown,
    #[cfg(test)]
    Control(Box<dyn StreamCleanupControlJob>),
}

/// Body of the cleanup thread. A panic inside the executor loop is caught
/// here so the thread can still drop its receiver and return a classified
/// outcome; the joiner reports the panic with stage `cleanup_executor`.
fn run_stream_cleanup_worker(
    receiver: mpsc::Receiver<StreamCleanupJob>,
    _control: &StreamCleanupSupervisorControl,
    probe: &ServiceStreamTestProbeSlot,
) -> StreamCleanupWorkerOutcome {
    let executor = catch_unwind(AssertUnwindSafe(|| run_stream_cleanup_executor(&receiver)));

    // The receiver owns all queued cleanup jobs. Drop it before constructing
    // the provisional outcome so queued recovery completes before any joiner
    // can observe and commit that outcome.
    drop(receiver);

    let outcome = match executor {
        Ok(StreamCleanupExecutorExit::Shutdown) => StreamCleanupWorkerOutcome::Clean,
        Ok(StreamCleanupExecutorExit::ReceiverDisconnected) => {
            StreamCleanupWorkerOutcome::Unexpected { detail: "cleanup receiver disconnected before explicit shutdown" }
        },
        #[cfg(test)]
        Ok(StreamCleanupExecutorExit::Unexpected(detail)) => StreamCleanupWorkerOutcome::Unexpected { detail },
        Err(panic) => StreamCleanupWorkerOutcome::Panicked {
            panic_message: stream_cleanup_supervisor_panic_message(panic.as_ref()),
        },
    };

    probe.pause_before_worker_outcome();
    if probe.panic_after_worker_outcome() {
        panic!("injected service stream cleanup supervisor join panic");
    }
    outcome
}

fn run_stream_cleanup_executor(receiver: &mpsc::Receiver<StreamCleanupJob>) -> StreamCleanupExecutorExit {
    loop {
        let job = match receiver.recv() {
            Ok(job) => job,
            Err(_) => return StreamCleanupExecutorExit::ReceiverDisconnected,
        };
        match job {
            StreamCleanupJob::Runtime(mut cleanup) => cleanup.run(),
            StreamCleanupJob::Shutdown => return StreamCleanupExecutorExit::Shutdown,
            #[cfg(test)]
            StreamCleanupJob::Control(job) => {
                if let Some(exit) = job.run() {
                    return exit;
                }
            },
        }
    }
}

#[cfg(test)]
struct ProbeCleanupJob {
    done: mpsc::Sender<thread::ThreadId>,
}

#[cfg(test)]
impl StreamCleanupControlJob for ProbeCleanupJob {
    fn run(self: Box<Self>) -> Option<StreamCleanupExecutorExit> {
        let _ = self.done.send(thread::current().id());
        None
    }
}

#[cfg(test)]
struct BlockCleanupJob {
    entered: Arc<Barrier>,
    release: Arc<Barrier>,
}

#[cfg(test)]
impl StreamCleanupControlJob for BlockCleanupJob {
    fn run(self: Box<Self>) -> Option<StreamCleanupExecutorExit> {
        self.entered.wait();
        self.release.wait();
        None
    }
}

#[cfg(test)]
struct UnexpectedExitCleanupJob;

#[cfg(test)]
impl StreamCleanupControlJob for UnexpectedExitCleanupJob {
    fn run(self: Box<Self>) -> Option<StreamCleanupExecutorExit> {
        Some(StreamCleanupExecutorExit::Unexpected("test requested unexpected cleanup supervisor exit"))
    }
}

#[cfg(test)]
pub(super) fn probe_cleanup_job_for_test(done: mpsc::Sender<thread::ThreadId>) -> StreamCleanupJob {
    StreamCleanupJob::Control(Box::new(ProbeCleanupJob { done }))
}

#[cfg(test)]
pub(crate) struct StreamCleanupBlocker {
    release: Option<Arc<Barrier>>,
}

#[cfg(test)]
impl Drop for StreamCleanupBlocker {
    fn drop(&mut self) {
        if let Some(release) = self.release.take() {
            release.wait();
        }
    }
}

#[cfg(test)]
pub(crate) fn block_stream_cleanup_thread_for_test(pool: &ServiceStreamPool) -> StreamCleanupBlocker {
    let entered = Arc::new(Barrier::new(2));
    let release = Arc::new(Barrier::new(2));
    pool.inner
        .cleanup_supervisor
        .enqueue(StreamCleanupJob::Control(Box::new(BlockCleanupJob {
            entered: Arc::clone(&entered),
            release: Arc::clone(&release),
        })))
        .expect("cleanup thread accepts blocker");
    entered.wait();
    StreamCleanupBlocker { release: Some(release) }
}

#[cfg(test)]
pub(super) fn stream_cleanup_thread_id_for_test(pool: &ServiceStreamPool) -> thread::ThreadId {
    let (sender, receiver) = mpsc::channel();
    pool.inner.cleanup_supervisor.enqueue(probe_cleanup_job_for_test(sender)).expect("cleanup thread accepts probe");
    receiver.recv().expect("cleanup thread returns probe")
}

#[cfg(test)]
pub(super) fn fail_next_cleanup_supervisor_spawn_for_test() {
    let probe = Arc::new(TestServiceStreamProbe::default());
    probe.fail_next_supervisor_spawn.store(true, Ordering::Release);
    ServiceStreamTestProbeSlot::install_for_next_pool(probe);
}

#[cfg(test)]
pub(super) fn panic_cleanup_supervisor_after_outcome_for_test(pool: &ServiceStreamPool) {
    pool.inner.cleanup_probe.with_test_probe(|probe| {
        probe.panic_after_worker_outcome.store(true, Ordering::Release);
    });
}

#[cfg(test)]
pub(super) fn pause_cleanup_supervisor_before_worker_return_for_test(
    pool: &ServiceStreamPool,
) -> StreamCleanupWorkerOutcomePause {
    let pause =
        StreamCleanupWorkerOutcomePause { arrived: Arc::new(Barrier::new(2)), release: Arc::new(Barrier::new(2)) };
    pool.inner.cleanup_probe.with_test_probe(|probe| {
        *probe.worker_outcome_pause.lock().unwrap() = Some(pause.clone());
    });
    pause
}

#[cfg(test)]
pub(super) fn wait_for_cleanup_supervisor_worker_finished_for_test(pool: &ServiceStreamPool, deadline: Instant) {
    loop {
        let finished =
            pool.inner.cleanup_supervisor.join.lock().unwrap().as_ref().is_some_and(thread::JoinHandle::is_finished);
        if finished {
            return;
        }
        assert!(Instant::now() < deadline, "cleanup supervisor did not finish its provisional outcome");
        thread::yield_now();
    }
}

#[cfg(test)]
pub(super) fn observe_cleanup_supervisor_shutdown_send_failure_for_test(
    pool: &ServiceStreamPool,
) -> mpsc::Receiver<()> {
    let (sender, receiver) = mpsc::channel();
    pool.inner.cleanup_probe.with_test_probe(|probe| {
        *probe.shutdown_send_failed.lock().unwrap() = Some(sender);
    });
    receiver
}

#[cfg(test)]
pub(super) fn observe_cleanup_supervisor_join_waiter_for_test(pool: &ServiceStreamPool) -> mpsc::Receiver<()> {
    let (sender, receiver) = mpsc::channel();
    pool.inner.cleanup_probe.with_test_probe(|probe| {
        *probe.join_waiter.lock().unwrap() = Some(sender);
    });
    receiver
}

#[cfg(test)]
pub(super) fn force_cleanup_supervisor_unexpected_exit_for_test(pool: &ServiceStreamPool) {
    pool.inner
        .cleanup_supervisor
        .enqueue(StreamCleanupJob::Control(Box::new(UnexpectedExitCleanupJob)))
        .expect("cleanup supervisor accepts unexpected-exit test job");
}