onlyne-client 2.0.0

//! Client-side zero-activity stall observation.
//!
//! A freeze past `stall_report_secs` is reported as `Report::Fault` kind
//! `stalled`. The ledger row stays as stored. Complete is not sent.

use onlyne_proto::Report;
use std::collections::{HashMap, HashSet};
use std::time::{Duration, Instant};

/// Fault kind stamped on a stall observation.
pub const STALLED: &str = "stalled";
/// Reason text carried with the stall fault.
pub const STALLED_REASON: &str = "no applied progress";

/// Per-task progress clock and one-shot freeze reporting.
#[derive(Debug, Default)]
pub struct StallWatch {
    last_progress: HashMap<String, Instant>,
    reported: HashSet<String>,
    /// The newest `(generation, seq)` each task's reporter has been seen at.
    /// A heartbeat's version is a watermark: a beat at or below it is a replay,
    /// and the row it names does not move for a beat that observed nothing.
    beats: HashMap<String, (u64, u64)>,
}

impl StallWatch {
    pub fn new() -> Self {
        Self::default()
    }

    /// Start the clock when a task is assigned. A later assign of the same
    /// task keeps the original instant.
    pub fn note_assigned(&mut self, task_id: impl Into<String>, now: Instant) {
        self.last_progress.entry(task_id.into()).or_insert(now);
    }

    /// An Applied persist refreshes the clock of an assigned task and clears
    /// the freeze report bit so a later freeze can be reported again. Clock
    /// creation belongs to [`Self::note_assigned`].
    pub fn note_applied(&mut self, task_id: &str, now: Instant) {
        if let Some(last_progress) = self.last_progress.get_mut(task_id) {
            *last_progress = now;
            self.reported.remove(task_id);
        }
    }

    /// Drop a task that has left the live set.
    pub fn forget(&mut self, task_id: &str) {
        self.last_progress.remove(task_id);
        self.reported.remove(task_id);
        self.beats.remove(task_id);
    }

    /// Record one heartbeat's version, answering whether it is the newest one
    /// this task's reporter has been seen at.
    ///
    /// The version orders the reporter's own frames: the same rule the stored
    /// row's watermark used to apply, kept here because a beat that observed
    /// nothing no longer moves the row. A reporter that came back with a new
    /// generation reads as newer, which is what a rebase is for.
    pub fn note_beat_seq(&mut self, task_id: &str, generation: u64, seq: u64) -> bool {
        match self.beats.get(task_id) {
            Some(&(seen_generation, seen_seq))
                if (generation, seq) <= (seen_generation, seen_seq) =>
            {
                false
            }
            _ => {
                self.beats.insert(task_id.to_string(), (generation, seq));
                true
            }
        }
    }

    /// Task ids whose freeze exceeds `threshold_secs` and have not been
    /// reported in this freeze episode. A threshold of zero reports nothing.
    pub fn due(&self, now: Instant, threshold_secs: u64) -> Vec<String> {
        if threshold_secs == 0 {
            return Vec::new();
        }
        let limit = Duration::from_secs(threshold_secs);
        let mut due: Vec<String> = self
            .last_progress
            .iter()
            .filter(|(task_id, _)| !self.reported.contains(*task_id))
            .filter(|(_, last)| now.saturating_duration_since(**last) > limit)
            .map(|(task_id, _)| task_id.clone())
            .collect();
        due.sort();
        due
    }

    /// Remember that this freeze episode has been reported.
    pub fn mark_reported(&mut self, task_id: &str) {
        self.reported.insert(task_id.to_string());
    }
}

/// Observation-only fault for a stalled task. The server records a fault row
/// and an advisory event; the ledger row is not flipped.
pub fn report(
    task_id: &str,
    session_id: Option<String>,
    generation: Option<u64>,
    seq: Option<u64>,
) -> Report {
    Report::Fault {
        task_id: Some(task_id.to_string()),
        session_id,
        generation,
        seq,
        kind: STALLED.to_string(),
        reason: STALLED_REASON.to_string(),
        desired: None,
        observed: None,
    }
}