aion-server 0.31.0

//! The expiry sweep's nomination read and the positive-fact predicates that
//! acquit a nominated worker.
//!
//! Split out of `heartbeat.rs`: deciding WHICH workers a sweep deregisters is
//! a self-contained read over the connection clock plus two corroborating
//! facts, and keeping it beside the tracker's mutation surface pushed that
//! file past the per-file length budget.

use std::time::Instant;

use crate::error::ServerError;
use crate::worker::registry::{ConnectedWorkerRegistry, WorkerId};

use super::{HeartbeatTracker, TaskLiveness};

impl HeartbeatTracker {
    /// Return the workers this sweep will DEREGISTER: those whose silence is
    /// corroborated by a push leg that is no longer open.
    ///
    /// # One clock nominates; two positive facts acquit
    ///
    /// This read used to consult two silence clocks — the connection clock and
    /// the PER-TASK beat clock — and NOTHING about whether the server still held
    /// a channel to the worker. Both inputs are facts about ABSENCE, so neither
    /// could tell a worker that is gone from one that is busy, and a worker
    /// running at its full concurrency looked identical to a dead one. It was
    /// then deregistered while connected and holding live work, its in-flight
    /// activities were surfaced as transport losses, and the queue census — with
    /// the worker gone from the pool — reported that no worker was connected
    /// about a worker on an open stream.
    ///
    /// Two things changed. The per-task clock no longer nominates anybody:
    /// `is_expired` is a predicate about ONE activity and was being used to
    /// condemn a whole process, so a single lapsed beat took down every other
    /// activity that worker was holding.
    ///
    /// # What now resolves a lapsed per-task beat — and what does not
    ///
    /// Stated exactly, because the honest answer is narrower than "something
    /// else handles it" and an earlier draft of this doc promised a mechanism
    /// "below" that does not exist.
    ///
    /// Three things still end a stuck activity, none of them this sweep:
    ///
    /// - the CONNECTION clock, when the worker itself also goes quiet — this
    ///   function nominates it, and `fail_expired_workers` then fails every
    ///   activity it was holding;
    /// - the engine seam's authored per-attempt bound, measured from the lease
    ///   (`aion::LeaseSignal`), where an operator authored one;
    /// - the outbox row's own attempt budget, backoff and dead-letter, once the
    ///   dispatch call returns.
    ///
    /// What is left uncovered, and is a real gap rather than a tidy hand-off: an
    /// activity whose beat lapses while its worker KEEPS answering. The worker
    /// stays in service, correctly — it is demonstrably alive — and no mechanism
    /// fails that single activity. `is_expired` now reaches production through
    /// nothing at all: its only reader is [`HeartbeatTracker::is_live`], which
    /// has no production caller. Before this change the whole-worker reap was
    /// what (brutally, and wrongly) covered that case.
    ///
    /// This is recorded rather than fixed here because inventing a per-activity
    /// deadline is a semantics decision about how long work may run without
    /// reporting, and that belongs to whoever owns the activity timeout — not to
    /// a sweep that was being asked to do it as a side effect. It is tracked as
    /// a server-lane issue; the number is deliberately not written here, because
    /// an issue reference in a doc comment rots the moment the issue moves.
    ///
    /// And the surviving clock alone no longer condemns anybody either. Silence
    /// is the SUSPICION; the verdict needs a second, positive fact to survive,
    /// and the worker keeps its registration unless BOTH of these hold:
    ///
    /// - **The push leg is still open.** [`WorkerHandle::is_connected`] asks
    ///   the transport, not the noise floor: gRPC answers from its stream
    ///   sender, liminal from its supervisor's connection table. A worker with
    ///   no leg left has nothing that could carry a beat, and reaping it is the
    ///   honest reading of both facts together.
    /// - **The server has PROVED it can reach that leg.** An open channel is
    ///   not an answering worker — the server can hold a perfectly healthy
    ///   looking `mpsc` to a wedged process indefinitely, which is the whole
    ///   reason the liveness ping exists. So a connected worker is spared only
    ///   while its dispatch path is proved reachable.
    ///
    /// # Why BOTH, and what each one alone would cost
    ///
    /// Connection alone is not enough: on gRPC a sender closes only when the
    /// stream task ends, and that is already the teardown sweep's business. A
    /// reaper gated on connection alone would therefore never fire for a gRPC
    /// worker at all, and a WEDGED process — stream open, nothing moving,
    /// in-flight work that will never complete — would hold its activities for
    /// ever. That is the failure #176 was written to end, and it must not be
    /// re-opened by fixing a different one. The connection clock is what catches
    /// it: a wedged process performs neither of the two acts that refresh it, so
    /// it is nominated within one window with its reachability unproved. That is
    /// why dropping the per-task arm costs #176 nothing.
    ///
    /// Reachability alone is not enough either: it is a measurement with a
    /// cadence, and a worker mid-probation or one probe behind has not yet
    /// proved anything about a link that is fine.
    ///
    /// # The reachability conjunct is currently DEFENSIVE, and here is why
    ///
    /// Say this rather than let a future reader assume it is load-bearing. Every
    /// call that proves reachability — `record_dispatch_reachability` — also
    /// refreshes the connection clock, and so do `record_heartbeat` and
    /// `record_liveness`. Since the only nomination is a lapsed connection
    /// clock, a worker that reaches this gate has by construction not been
    /// proved recently either: the two clocks are advanced by the same calls, so
    /// "overdue but provably reachable" is not a state this tracker can be in
    /// today. The conjunct therefore cannot currently spare anybody the
    /// connection arm would have condemned.
    ///
    /// It stays because it is the correct predicate and it is free. If a future
    /// change proves reachability without touching the connection clock — a
    /// side-channel probe, a transport that answers without traffic — this gate
    /// is already the right one, and its absence would be a silent re-opening of
    /// the reap-a-busy-worker hole rather than a visible one.
    ///
    /// Together they say what the sweep actually needs: this worker's leg is
    /// open AND the server has recently pushed something down it that came
    /// back, so its silence is BUSY, not gone. That is exactly the state a
    /// worker at its full concurrency is now in — the receive loop no longer
    /// parks on admission, so a saturated worker answers pings while it works —
    /// and it is exactly the state the old code could not see.
    ///
    /// # What a connected, unreachable worker gets first
    ///
    /// Dispatch INELIGIBILITY, published by the liveness probe on this same
    /// evidence and several probe cadences before this sweep can fire: the
    /// worker stays registered, keeps its in-flight work, and is sent nothing
    /// new. This method neither reads nor re-derives that verdict — one reaper,
    /// one eligibility verdict, and each reaches its own conclusion from the
    /// evidence it owns.
    ///
    /// A read failure on either input yields the reaping answer for that
    /// worker, which is the conservative direction: it can only fall back to
    /// the behaviour this method had before it could ask at all, never invent a
    /// reap the silence clocks did not already ask for.
    ///
    /// # Errors
    ///
    /// Returns [`ServerError::LockPoisoned`] if tracker state cannot be trusted.
    pub fn expired_workers(
        &self,
        registry: &ConnectedWorkerRegistry,
        now: Instant,
    ) -> Result<Vec<WorkerId>, ServerError> {
        let candidates = {
            let state = self.state()?;
            // ONE candidate source: the CONNECTION silence clock. A worker whose
            // connection has been quiet past the window is suspected; nothing
            // else nominates anybody.
            //
            // The per-task clock used to nominate too, and that was the outage's
            // own mechanism wearing a second hat: `is_expired` is a predicate
            // about ONE activity, and using it to condemn the whole worker meant
            // a single lapsed beat deregistered a connected process along with
            // every other activity it was holding. A per-task fact stays a
            // per-task fact — the sweep still names those tasks below, and
            // whatever fails an individual activity still does.
            //
            // Nothing is lost by dropping it. `state.connections` is refreshed
            // only by what the WORKER does — `record_heartbeat` and
            // `record_liveness` — so a wedged process refreshes nothing, goes
            // stale within one window, and is nominated here with its
            // reachability unproved. That is #176's case, decided entirely by
            // this arm.
            state
                .connections
                .iter()
                .filter(|(_, last_activity)| {
                    now.checked_duration_since(**last_activity)
                        .is_some_and(|elapsed| elapsed > self.heartbeat_window)
                })
                .map(|(worker_id, _)| *worker_id)
                .collect::<Vec<_>>()
        };
        let mut workers = Vec::new();
        for worker_id in candidates {
            if is_still_connected(registry, worker_id) && self.is_provably_reachable(worker_id, now)
            {
                continue;
            }
            workers.push(worker_id);
        }
        workers.sort_unstable();
        Ok(workers)
    }

    /// Whether the server has PROVED it can reach this worker's dispatch path,
    /// reading through the expiry sweep's own error policy.
    ///
    /// [`Self::is_dispatch_reachable`] is the fact; this is what the sweep does
    /// when it cannot read it. An unreadable tracker answers "not proved",
    /// which reaps — the same conservative direction the connection check
    /// takes, and the behaviour this sweep had before it could consult
    /// reachability at all.
    fn is_provably_reachable(&self, worker_id: WorkerId, now: Instant) -> bool {
        match self.is_dispatch_reachable(worker_id, now) {
            Ok(reachable) => reachable,
            Err(error) => {
                tracing::error!(
                    worker_id = worker_id.value(),
                    %error,
                    "worker reachability is unreadable; treating this worker's dispatch path as \
                     unproved, which is the reading the expiry sweep had before it could ask"
                );
                false
            }
        }
    }
}

/// Whether the server still holds an open push leg to `worker_id`.
///
/// An UNREGISTERED worker answers `false`: there is no leg because there is no
/// registration, and a worker already out of the registry is not one this sweep
/// needs to protect. A poisoned registry lock also answers `false`, which is the
/// conservative direction — it degrades this decision to what it was before the
/// connected fact existed, and never manufactures a reap that the two silence
/// clocks did not already ask for.
pub(super) fn is_still_connected(registry: &ConnectedWorkerRegistry, worker_id: WorkerId) -> bool {
    match registry.worker_by_id(worker_id) {
        Ok(worker) => worker.is_some_and(|worker| worker.is_connected()),
        Err(error) => {
            tracing::error!(
                worker_id = worker_id.value(),
                %error,
                "connected-worker registry is unreadable; judging this worker's push leg closed, \
                 which is the reading the expiry sweep had before it could ask at all"
            );
            false
        }
    }
}

/// Whether ONE activity's beat has lapsed past its own heartbeat window.
pub(super) fn is_expired(liveness: &TaskLiveness, now: Instant) -> bool {
    now.checked_duration_since(liveness.last_heartbeat_at)
        .is_some_and(|elapsed| elapsed > liveness.heartbeat_window)
}