dpm_engine/run/
projection.rs

1//! Deriving what can be said about a run at one clock reading.
2//!
3//! The durable lifecycle is never edited by silence. Staleness is computed from the time DPM last
4//! received anything from the run, on DPM's own clock; a time the executor reports about itself is
5//! kept but never consulted, so a run cannot stay fresh by claiming a future instant.
6
7use chrono::{DateTime, TimeDelta, Utc};
8use dpm_model::{
9    LineageId, LineageStatus, ObservedStatus, Orphan, Plan, RunSnapshot, RunState, RunView,
10    WorkStatus,
11};
12
13/// How long a working or waiting run may go unheard before it is reported stale.
14pub const STALE_AFTER: TimeDelta = TimeDelta::seconds(300);
15
16/// Compute a run's view at `now`.
17///
18/// `lineage` is the history the project store continues (`None` without a store) and `plan` is the
19/// snapshot read from it, used only to notice that an unfinished run's task has moved on.
20#[must_use]
21pub fn project(
22    snapshot: RunSnapshot,
23    plan: Option<&Plan>,
24    lineage: Option<LineageId>,
25    now: DateTime<Utc>,
26) -> RunView {
27    let RunSnapshot {
28        record,
29        last,
30        activity,
31        operations,
32    } = snapshot;
33    let current = super::same_history(record.contract.lineage_id, lineage);
34    let last_receipt_at = activity.latest.as_ref().map_or(last.recorded_at, |latest| {
35        latest.recorded_at.max(last.recorded_at)
36    });
37    let silent = now.signed_duration_since(last_receipt_at) >= STALE_AFTER;
38    let status = match last.state {
39        RunState::Failed => ObservedStatus::Failed,
40        RunState::Interrupted => ObservedStatus::Interrupted,
41        RunState::Completed => ObservedStatus::Completed,
42        _ if !current => ObservedStatus::Unknown,
43        _ if silent => ObservedStatus::Stale,
44        RunState::Working => ObservedStatus::Working,
45        RunState::Waiting => ObservedStatus::Waiting,
46    };
47    let stale_at = matches!(status, ObservedStatus::Working | ObservedStatus::Waiting)
48        .then(|| super::stale_after(last_receipt_at))
49        .flatten();
50    let orphan = (!last.state.is_terminal() && current)
51        .then(|| plan.and_then(|plan| orphan(&record, plan)))
52        .flatten();
53    RunView {
54        state: last.state,
55        state_since: last.recorded_at,
56        state_detail: last.detail,
57        status,
58        last_receipt_at,
59        stale_at,
60        activity,
61        lineage: if current {
62            LineageStatus::Current
63        } else {
64            LineageStatus::Foreign
65        },
66        orphan,
67        operations,
68        evaluated_at: now,
69        run: record,
70    }
71}
72
73fn orphan(record: &dpm_model::RunRecord, plan: &Plan) -> Option<Orphan> {
74    let Some(work) = plan.work_items.get(&record.work) else {
75        return Some(Orphan::WorkMissing);
76    };
77    if work.execution.owner.as_ref() != Some(&record.executor) {
78        return Some(Orphan::OwnerChanged {
79            owner: work.execution.owner.clone(),
80        });
81    }
82    match work.execution.status {
83        WorkStatus::Claimed | WorkStatus::InProgress => None,
84        status => Some(Orphan::NotExecuting { status }),
85    }
86}
dpm_engine/run/projection.rs

dpm_engine/run/
projection.rs
