aion-rs 0.31.0

//! The no-progress bound every engine stop path waits under (AE-017).
//!
//! An engine stop drains three worker threads — the cleanup executor, the
//! process-exit drainer, and the process-exit callback dispatcher — and each
//! of them may be in the middle of real work when the stop arrives: a
//! terminal being recorded to the store, an abort being delivered, an exit
//! event being observed. Before this module, each drain waited under ONE
//! total bound derived from the signal-delivery readiness window (50 ms × 9
//! = 450 ms), and a box that was merely busy blew it: on 2026-08-29 a tree
//! with no change at all failed its own stop 3/3 under a foreign build and
//! passed 3/3 on the same binary thirteen minutes later.
//!
//! # What the bound means now
//!
//! It is a NO-PROGRESS bound. The clock restarts every time the drained
//! worker completes a job and fires only when nothing has completed for the
//! whole window. A loaded box that is still finishing callbacks is therefore
//! never called a wedge, however long its queue; a genuinely blocked callback
//! is named within one window of blocking; and the progress warnings emitted
//! while the drain waits are the evidence of which one it was. The bound is
//! GIVEN — by the engine builder, from the server's configuration — never
//! derived from an unrelated policy and never defaulted here.
//!
//! The contract a blocked job used to satisfy still holds: a callback that
//! never returns produces a bounded, retryable failure, not a hang. It is
//! bounded from its last progress rather than from the stop request.

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// What a drained worker reports about itself: jobs completed, the job in
/// flight, and the depth of its queue. Updated by the worker thread; read
/// by the stop path.
#[derive(Debug, Default)]
pub(super) struct DrainProgress {
    completed: AtomicU64,
    in_flight_since: Mutex<Option<Instant>>,
    queued: AtomicUsize,
}

/// A point-in-time reading of a worker's progress, for a warning or an error.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct ProgressSnapshot {
    /// Jobs the worker has completed since it started.
    pub(super) completed: u64,
    /// How long the job in flight has been running, when one is.
    pub(super) in_flight_for: Option<Duration>,
    /// Jobs waiting behind the one in flight.
    pub(super) queued: usize,
}

impl DrainProgress {
    pub(super) fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// The worker took a job.
    pub(super) fn begin(&self) {
        // A poisoned lock here means a worker panicked mid-update; the stop
        // path must still be able to read, so the poison is cleared rather
        // than propagated — this is a gauge, not a ledger.
        let mut in_flight = self
            .in_flight_since
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        *in_flight = Some(Instant::now());
    }

    /// The worker finished the job it took.
    pub(super) fn finish(&self) {
        let mut in_flight = self
            .in_flight_since
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        *in_flight = None;
        drop(in_flight);
        self.completed.fetch_add(1, Ordering::AcqRel);
    }

    pub(super) fn enqueued(&self) {
        self.queued.fetch_add(1, Ordering::AcqRel);
    }

    pub(super) fn dequeued(&self) {
        // Saturating: a dequeue racing a stop must never wrap the gauge.
        let _ = self
            .queued
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
                current.checked_sub(1)
            });
    }

    pub(super) fn queued(&self) -> usize {
        self.queued.load(Ordering::Acquire)
    }

    pub(super) fn snapshot(&self) -> ProgressSnapshot {
        let in_flight_for = self
            .in_flight_since
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .map(|since| since.elapsed());
        ProgressSnapshot {
            completed: self.completed.load(Ordering::Acquire),
            in_flight_for,
            queued: self.queued(),
        }
    }
}

/// Why a drain gave up: nothing completed for the whole bound.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct DrainTimedOut {
    /// The bound the drain waited under.
    pub(super) bound: Duration,
    /// How long it has been since the worker last completed a job.
    pub(super) since_progress: Duration,
    /// The worker's queue depth at the moment the drain gave up.
    pub(super) queued: usize,
}

/// The bound, and the waiting that interprets it.
#[derive(Clone, Copy, Debug)]
pub(super) struct DrainBound {
    bound: Duration,
}

impl DrainBound {
    pub(super) const fn new(bound: Duration) -> Self {
        Self { bound }
    }

    /// How often a waiting loop wakes to re-read progress and the stop
    /// signal: one hundredth of the bound, so a stop is observed within one
    /// percent of the operator's patience whatever that patience is. Never
    /// zero — a zero wait would spin.
    pub(super) fn poll_interval(self) -> Duration {
        (self.bound / 100).max(Duration::from_micros(1))
    }

    /// How often the drain says aloud that it is still waiting: a quarter of
    /// the bound, so an operator watching a 30 s drain sees the in-flight job
    /// named three times before the drain gives up.
    pub(super) fn progress_interval(self) -> Duration {
        (self.bound / 4).max(self.poll_interval())
    }

    /// Wait for `stopped` to deliver (or disconnect), restarting the clock
    /// every time `progress` records a completed job, warning at
    /// [`Self::progress_interval`] while it waits, and giving up only when
    /// nothing has completed for the whole bound.
    pub(super) fn wait(
        self,
        worker: &'static str,
        stopped: &Receiver<()>,
        progress: &DrainProgress,
    ) -> Result<(), DrainTimedOut> {
        let poll = self.poll_interval();
        let warn_every = self.progress_interval();
        let mut last_completed = progress.snapshot().completed;
        let mut last_progress_at = Instant::now();
        let mut last_warned_at = last_progress_at;
        loop {
            match stopped.recv_timeout(poll) {
                Ok(()) | Err(RecvTimeoutError::Disconnected) => return Ok(()),
                Err(RecvTimeoutError::Timeout) => {}
            }
            let now = Instant::now();
            let snapshot = progress.snapshot();
            if snapshot.completed != last_completed {
                last_completed = snapshot.completed;
                last_progress_at = now;
            }
            let since_progress = now.duration_since(last_progress_at);
            if since_progress >= self.bound {
                tracing::error!(
                    worker,
                    bound_ms = self.bound.as_millis(),
                    since_progress_ms = since_progress.as_millis(),
                    in_flight_ms = snapshot.in_flight_for.map(|age| age.as_millis()),
                    queued = snapshot.queued,
                    "engine stop gave up draining: nothing completed for the whole bound"
                );
                return Err(DrainTimedOut {
                    bound: self.bound,
                    since_progress,
                    queued: snapshot.queued,
                });
            }
            if now.duration_since(last_warned_at) >= warn_every {
                last_warned_at = now;
                tracing::warn!(
                    worker,
                    bound_ms = self.bound.as_millis(),
                    since_progress_ms = since_progress.as_millis(),
                    in_flight_ms = snapshot.in_flight_for.map(|age| age.as_millis()),
                    queued = snapshot.queued,
                    completed = snapshot.completed,
                    "engine stop still draining; the bound restarts on every completed job"
                );
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Barrier, mpsc};
    use std::thread;
    use std::time::{Duration, Instant};

    use super::{DrainBound, DrainProgress};

    /// A worker whose jobs each finish inside the bound but whose queue
    /// takes many bounds in total is a slow box, not a wedge: the drain
    /// waits it out.
    #[test]
    fn a_progressing_worker_is_waited_out_past_the_total_bound()
    -> Result<(), Box<dyn std::error::Error>> {
        let bound = DrainBound::new(Duration::from_millis(80));
        let progress = DrainProgress::new();
        let (stopped_sender, stopped) = mpsc::sync_channel(1);
        let worker_progress = std::sync::Arc::clone(&progress);
        let rendezvous = Arc::new(Barrier::new(2));
        let worker_rendezvous = Arc::clone(&rendezvous);
        let worker = thread::spawn(move || {
            // Five jobs of 40 ms: 200 ms total, every gap under the 80 ms bound.
            for index in 0..5 {
                worker_progress.begin();
                if index == 0 {
                    worker_rendezvous.wait();
                }
                thread::sleep(Duration::from_millis(40));
                worker_progress.finish();
            }
            let _ = stopped_sender.send(());
        });
        rendezvous.wait();
        let started = Instant::now();
        bound
            .wait("test-worker", &stopped, &progress)
            .map_err(|timed_out| {
                format!("a progressing worker was called a wedge: {timed_out:?}")
            })?;
        assert!(
            started.elapsed() >= Duration::from_millis(150),
            "the drain waited for the whole queue, not one bound"
        );
        worker.join().map_err(|_| "worker panicked")?;
        Ok(())
    }

    /// A job that never completes is named within one bound of its last
    /// progress, with the in-flight age and queue depth on the error.
    #[test]
    fn a_blocked_worker_fails_bounded_from_its_last_progress()
    -> Result<(), Box<dyn std::error::Error>> {
        let bound = DrainBound::new(Duration::from_millis(60));
        let progress = DrainProgress::new();
        let (_stopped_sender, stopped) = mpsc::sync_channel::<()>(1);
        progress.enqueued();
        progress.enqueued();
        progress.dequeued();
        progress.begin();
        let started = Instant::now();
        let timed_out = bound
            .wait("test-worker", &stopped, &progress)
            .err()
            .ok_or("a blocked worker must time the drain out")?;
        let elapsed = started.elapsed();
        assert!(
            elapsed >= Duration::from_millis(60),
            "not before the bound: {elapsed:?}"
        );
        assert!(
            elapsed < Duration::from_millis(600),
            "not long after it: {elapsed:?}"
        );
        assert_eq!(timed_out.bound, Duration::from_millis(60));
        assert!(timed_out.since_progress >= Duration::from_millis(60));
        assert_eq!(timed_out.queued, 1, "the queue depth rides the error");
        Ok(())
    }

    /// The clock restarts on progress: a worker that completes one job late
    /// in the first window and then blocks fails one bound after THAT job,
    /// not one bound after the stop request.
    #[test]
    fn the_clock_restarts_on_progress_not_on_the_stop_request()
    -> Result<(), Box<dyn std::error::Error>> {
        let bound = DrainBound::new(Duration::from_millis(100));
        let progress = DrainProgress::new();
        let (_stopped_sender, stopped) = mpsc::sync_channel::<()>(1);
        let worker_progress = std::sync::Arc::clone(&progress);
        let rendezvous = Arc::new(Barrier::new(2));
        let worker_rendezvous = Arc::clone(&rendezvous);
        let worker = thread::spawn(move || {
            worker_progress.begin();
            worker_rendezvous.wait();
            thread::sleep(Duration::from_millis(70));
            worker_progress.finish();
            worker_progress.begin();
            // Never finishes.
            thread::sleep(Duration::from_millis(250));
        });
        rendezvous.wait();
        let started = Instant::now();
        let timed_out = bound
            .wait("test-worker", &stopped, &progress)
            .err()
            .ok_or("the blocked second job must time the drain out")?;
        let elapsed = started.elapsed();
        assert!(
            elapsed >= Duration::from_millis(160),
            "the bound restarted after the 70 ms job: {elapsed:?}"
        );
        assert!(timed_out.since_progress < Duration::from_millis(160));
        worker.join().map_err(|_| "worker panicked")?;
        Ok(())
    }

    #[test]
    fn a_stopped_worker_returns_at_once() -> Result<(), Box<dyn std::error::Error>> {
        let bound = DrainBound::new(Duration::from_secs(30));
        let progress = DrainProgress::new();
        let (stopped_sender, stopped) = mpsc::sync_channel(1);
        stopped_sender.send(())?;
        let started = Instant::now();
        bound
            .wait("test-worker", &stopped, &progress)
            .map_err(|timed_out| format!("{timed_out:?}"))?;
        assert!(started.elapsed() < Duration::from_secs(1));
        Ok(())
    }

    #[test]
    fn intervals_derive_from_the_bound_and_never_reach_zero() {
        let bound = DrainBound::new(Duration::from_secs(30));
        assert_eq!(bound.poll_interval(), Duration::from_millis(300));
        assert_eq!(bound.progress_interval(), Duration::from_millis(7_500));
        let tiny = DrainBound::new(Duration::from_nanos(1));
        assert!(tiny.poll_interval() > Duration::ZERO);
        assert!(tiny.progress_interval() >= tiny.poll_interval());
    }
}