aion-server 0.31.0

//! Rows parked for want of worker capacity, and the pending→confirmed protocol
//! that stops a re-offer racing the park it depends on.
//!
//! A row answered BUSY is remembered here so a freed capacity slot can re-offer
//! it long before its durable fence — which rides the operator's FAILURE
//! backoff — expires. It is remembered BEFORE that durable write, because a
//! pulse landing during the write would otherwise find nothing and spend itself.
//!
//! # Why an entry is PENDING until its park is durable
//!
//! Remembering first opens a narrower window than the one it closes. The row's
//! delivery task is spawned, so the run loop can be in its wake arm while the
//! busy arm's `retry_outbox_row(now + backoff)` is still in flight. A re-offer
//! that acted on the row there would issue `retry_outbox_row(now)` against the
//! same key while the park write was unserialised — and if the park's write
//! commits second, the row ends up fenced at `now + backoff` AND forgotten,
//! which is the exact outcome the whole capacity wake exists to prevent.
//!
//! So the property this type enforces is: **a re-offer never acts on a row whose
//! durable park is not yet confirmed.** A pulse arriving during the window is
//! not lost — it is recorded on the entry, and [`CapacityParkedRows::confirm`]
//! re-checks it the moment the write lands and tells the caller to re-offer
//! immediately. No lost wake, no two writers on one key, and no interval to
//! tune.

use std::collections::BTreeMap;
use std::sync::Mutex;

use tracing::error;

/// One row waiting for a worker slot.
#[derive(Debug)]
struct ParkedRow {
    /// The attempt the row was parked at, passed through unchanged on re-offer
    /// so a capacity wait never costs the row an attempt.
    attempt: u32,
    /// Whether the durable park has been written. A re-offer skips a row that
    /// is still pending.
    confirmed: bool,
    /// Whether a capacity pulse arrived while this row was pending. Re-checked
    /// by [`CapacityParkedRows::confirm`], so the skipped pulse is deferred
    /// rather than dropped.
    woken_while_pending: bool,
}

/// The set of rows waiting for capacity, shared between the busy arm that parks
/// them and the wake arm that re-offers them.
#[derive(Debug, Default)]
pub(super) struct CapacityParkedRows {
    rows: Mutex<BTreeMap<String, ParkedRow>>,
}

impl CapacityParkedRows {
    /// Remember a row BEFORE its durable park is written.
    ///
    /// Pending until [`Self::confirm`]: a pulse arriving now finds the row and
    /// records itself on it rather than acting on a park that may not land.
    pub(super) fn park_pending(&self, dispatch_key: &str, attempt: u32) {
        let Ok(mut rows) = self.rows.lock() else {
            error!(
                dispatch_key,
                "capacity-parked set is poisoned; this row waits for its durable fence instead \
                 of the next freed slot"
            );
            return;
        };
        rows.insert(
            dispatch_key.to_owned(),
            ParkedRow {
                attempt,
                confirmed: false,
                woken_while_pending: false,
            },
        );
    }

    /// The durable park FAILED, so forget the row.
    ///
    /// A row left remembered with no fence behind it would be re-offered on
    /// every freed slot, pulling forward a `visible_after` that was never
    /// written — a phantom the sweep can never satisfy.
    pub(super) fn abandon(&self, dispatch_key: &str) {
        let Ok(mut rows) = self.rows.lock() else {
            error!(
                dispatch_key,
                "capacity-parked set is poisoned; a row whose park failed stays remembered"
            );
            return;
        };
        rows.remove(dispatch_key);
    }

    /// The durable park landed. Returns `true` when a pulse arrived while the
    /// row was pending, in which case the caller must re-offer it NOW — the
    /// entry has been removed and no further wake is coming for it.
    pub(super) fn confirm(&self, dispatch_key: &str) -> bool {
        let Ok(mut rows) = self.rows.lock() else {
            error!(
                dispatch_key,
                "capacity-parked set is poisoned; this row waits for its durable fence instead \
                 of the next freed slot"
            );
            return false;
        };
        let Some(row) = rows.get_mut(dispatch_key) else {
            return false;
        };
        row.confirmed = true;
        if row.woken_while_pending {
            rows.remove(dispatch_key);
            return true;
        }
        false
    }

    /// Take every CONFIRMED row, for re-offering against a freed slot.
    ///
    /// Pending rows stay and are marked as woken, so the pulse they could not be
    /// acted on for is handed to [`Self::confirm`] rather than lost.
    pub(super) fn take_confirmed(&self) -> Vec<(String, u32)> {
        let Ok(mut rows) = self.rows.lock() else {
            error!(
                "capacity-parked set is poisoned; rows wait for their durable fences instead of \
                 the freed slot"
            );
            return Vec::new();
        };
        let mut ready = Vec::new();
        rows.retain(|dispatch_key, row| {
            if row.confirmed {
                ready.push((dispatch_key.clone(), row.attempt));
                return false;
            }
            row.woken_while_pending = true;
            true
        });
        ready
    }

    /// Whether `dispatch_key` is remembered but not yet confirmed.
    ///
    /// Test-only, and `#[cfg(test)]` rather than allowed-dead because that is
    /// what it is: production never asks, it only parks and confirms. The tests
    /// that pin this ordering do have to ask — they must establish that their
    /// simulated pulse really landed inside the pending window, or they would be
    /// asserting a property they never reached.
    #[cfg(test)]
    pub(super) fn is_pending(&self, dispatch_key: &str) -> bool {
        self.rows
            .lock()
            .is_ok_and(|rows| rows.get(dispatch_key).is_some_and(|row| !row.confirmed))
    }

    /// Whether `dispatch_key` is remembered at all.
    ///
    /// Test-only for the same reason: the failed-park test asserts the absence
    /// of a phantom entry, which nothing in production needs to read.
    #[cfg(test)]
    pub(super) fn contains(&self, dispatch_key: &str) -> bool {
        self.rows
            .lock()
            .is_ok_and(|rows| rows.contains_key(dispatch_key))
    }
}