aion_server/worker/
attempt_progress.rs

1//! Reading the volatile worker progress note of one attempt — honestly.
2//!
3//! A worker heartbeat may carry an opaque progress payload ("what I am doing
4//! right now"). The server holds the most recent one per in-flight attempt in
5//! [`HeartbeatTracker`], in memory, and writes it nowhere. That makes the read
6//! path a measurement problem rather than a lookup: the absence of a note has
7//! two completely different causes, and collapsing them would tell an operator
8//! that a busy worker has gone silent every time the server was restarted.
9//!
10//! So the read returns [`AttemptProgress`], which never says "no note" without
11//! saying which kind of no:
12//!
13//! - [`AttemptProgress::Reported`] — the tracker holds a note for this attempt.
14//! - [`AttemptProgress::NoneSent`] — the tracker holds the ATTEMPT and no note
15//!   on it. This is a measurement: the worker really has reported nothing.
16//! - [`AttemptProgress::Untracked`] — the tracker does not hold the attempt, so
17//!   there is nothing to measure. A restart is the usual cause (the notes were
18//!   in a process that no longer exists), and the other causes — the owner was
19//!   swept off the tracker, or nothing has leased the dispatch — are equally
20//!   unmeasurable from here. Every one of them is this variant; none of them is
21//!   ever reported as silence.
22
23use aion_core::{ActivityId, Payload, WorkflowId};
24use chrono::{DateTime, Utc};
25
26use super::heartbeat::{HeartbeatTracker, TaskLiveness};
27use crate::error::ServerError;
28
29/// What this server process can say about one attempt's progress note.
30#[derive(Clone, Debug, PartialEq, Eq)]
31pub enum AttemptProgress {
32    /// The worker serving this attempt reported this note, and this process
33    /// received it at `reported_at`.
34    Reported {
35        /// The worker's opaque progress payload, exactly as it was sent.
36        payload: Payload,
37        /// Wall-clock instant this process received the reporting heartbeat.
38        reported_at: DateTime<Utc>,
39    },
40    /// This process is tracking the attempt and no note has been reported on it.
41    NoneSent,
42    /// This process is not tracking the attempt, so it cannot say whether a
43    /// note was reported.
44    Untracked,
45}
46
47/// The progress note this process holds for `(workflow, activity, attempt)`.
48///
49/// When a within-attempt failover has both a dying owner and its adopter
50/// tracked, the entry with the most recent heartbeat wins: it is the one still
51/// being fed, and taking the other would report a note the live worker has
52/// already superseded. `last_heartbeat_at` is a monotonic [`std::time::Instant`]
53/// minted by this process, so comparing entries is meaningful even though the
54/// value itself would be meaningless to report.
55///
56/// # Errors
57///
58/// Returns [`ServerError::LockPoisoned`] when the tracker state cannot be read.
59/// An unreadable tracker is reported, never rendered as "no note" — that would
60/// be the exact confusion this module exists to prevent.
61pub fn attempt_progress(
62    tracker: &HeartbeatTracker,
63    workflow_id: &WorkflowId,
64    activity_id: &ActivityId,
65    attempt: u32,
66) -> Result<AttemptProgress, ServerError> {
67    let entries = tracker.attempt_entries(workflow_id, activity_id, attempt)?;
68    let Some(freshest) = freshest(entries) else {
69        return Ok(AttemptProgress::Untracked);
70    };
71    match (freshest.last_progress, freshest.last_progress_at) {
72        (Some(payload), Some(reported_at)) => Ok(AttemptProgress::Reported {
73            payload,
74            reported_at,
75        }),
76        // A note is stamped with its receive instant in the same assignment, so
77        // a note without a stamp is unreachable. It is handled as UNTRACKED
78        // rather than as `NoneSent` because a note that exists without a receive
79        // instant is a note this process cannot describe — reporting it as
80        // "nothing was sent" would be the one lie the type exists to prevent.
81        (Some(_payload), None) => Ok(AttemptProgress::Untracked),
82        (None, _) => Ok(AttemptProgress::NoneSent),
83    }
84}
85
86/// The tracked entry with the most recent heartbeat.
87fn freshest(entries: Vec<TaskLiveness>) -> Option<TaskLiveness> {
88    entries
89        .into_iter()
90        .max_by_key(|liveness| liveness.last_heartbeat_at)
91}
92
93#[cfg(test)]
94#[path = "attempt_progress_tests.rs"]
95mod tests;
aion_server/worker/attempt_progress.rs

aion_server/worker/
attempt_progress.rs
