mail4agent-messenger-shell 0.4.29

//! Reopening a dead push socket: with back-off between attempts and a warning on stderr when the
//! socket goes down, on every failed attempt, and when it is back. Before this the reopen ran
//! silently every few seconds (or never, on the node and grok listeners), so a listener that had
//! stopped waking sessions left nothing in the log above debug.

use std::time::{Duration, Instant};

/// First wait after a failed reopen; it doubles up to [`MAX_BACKOFF`].
pub(crate) const FIRST_BACKOFF: Duration = Duration::from_secs(5);
/// Longest wait between two reopen attempts.
pub(crate) const MAX_BACKOFF: Duration = Duration::from_secs(120);

/// Wait before the next attempt after `failures` failed ones in a row.
pub(crate) fn backoff(failures: u32) -> Duration {
    if failures == 0 {
        return Duration::ZERO;
    }
    let factor = 1u32.checked_shl(failures.saturating_sub(1).min(16)).unwrap_or(u32::MAX);
    FIRST_BACKOFF.checked_mul(factor).map_or(MAX_BACKOFF, |d| d.min(MAX_BACKOFF))
}

/// Back-off state of one push socket. `label` names the listener in the log (a nick or "client").
#[derive(Debug)]
pub(crate) struct PushHeal {
    failures: u32,
    next_try: Instant,
    down_since: Option<Instant>,
}

impl PushHeal {
    pub(crate) fn new() -> Self {
        Self { failures: 0, next_try: Instant::now(), down_since: None }
    }

    /// True when an attempt may run now.
    pub(crate) fn due(&self) -> bool {
        Instant::now() >= self.next_try
    }

    /// Consecutive failed attempts.
    #[cfg(test)]
    pub(crate) fn failures(&self) -> u32 {
        self.failures
    }

    /// The socket was found dead (or the sessions keep failing): warns once per outage.
    pub(crate) fn down(&mut self, label: &str, reason: Option<&str>) {
        if self.down_since.is_none() {
            self.down_since = Some(Instant::now());
            eprintln!("mail4agent: WARN push socket for {label} is down ({}); reconnecting", reason.unwrap_or("no reason given"));
        }
    }

    /// A reopen attempt failed: the next one waits longer.
    pub(crate) fn failed(&mut self, label: &str, err: &str) {
        self.failures = self.failures.saturating_add(1);
        if self.down_since.is_none() {
            self.down_since = Some(Instant::now());
        }
        let wait = backoff(self.failures);
        self.next_try = Instant::now() + wait;
        eprintln!("mail4agent: WARN push socket for {label} not reopened (attempt {}, next in {}s): {err}", self.failures, wait.as_secs());
    }

    /// A reopen attempt succeeded.
    pub(crate) fn succeeded(&mut self, label: &str) {
        if let Some(since) = self.down_since.take() {
            eprintln!("mail4agent: push socket for {label} is open again after {}s ({} failed attempts)", since.elapsed().as_secs(), self.failures);
        }
        self.failures = 0;
        self.next_try = Instant::now();
    }
}

/// Warns when drives keep failing: on the 3rd failed tick in a row and every 40th after it.
pub(crate) fn warn_streak(label: &str, streak: u32, error: Option<&str>) {
    if streak == 3 || (streak > 3 && streak % 40 == 3) {
        eprintln!("mail4agent: WARN {label}: {streak} ticks in a row failed: {}", error.unwrap_or("-"));
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn backoff_doubles_and_is_capped() {
        assert_eq!(backoff(0), Duration::ZERO);
        assert_eq!(backoff(1), Duration::from_secs(5));
        assert_eq!(backoff(2), Duration::from_secs(10));
        assert_eq!(backoff(3), Duration::from_secs(20));
        assert_eq!(backoff(5), Duration::from_secs(80));
        assert_eq!(backoff(6), MAX_BACKOFF);
        assert_eq!(backoff(u32::MAX), MAX_BACKOFF);
    }

    #[test]
    fn a_failure_defers_the_next_attempt_and_success_resets() {
        let mut heal = PushHeal::new();
        assert!(heal.due());
        heal.down("t", Some("silent"));
        heal.failed("t", "timed out");
        assert_eq!(heal.failures(), 1);
        assert!(!heal.due(), "five seconds of back-off");
        heal.succeeded("t");
        assert_eq!(heal.failures(), 0);
        assert!(heal.due());
    }
}