artifactize_core/runs/
observe.rs

1//! One subscribed observer, independent of execution and Eval kind.
2use super::{Activity, Context};
3use crate::{
4    store::{self, Request, Run, RunView},
5    types::{FailureCode, RequestStatus, RunId, RunStatus, Timestamp},
6};
7use serde::Serialize;
8use std::time::Duration;
9use tokio_util::sync::CancellationToken;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
12#[serde(rename_all = "snake_case")]
13pub enum State {
14    Preparing,
15    Running,
16    Waiting,
17    /// Open while its sign-offs are claimed or its idle grace runs.
18    WaitingHuman,
19    Cancelling,
20    Green,
21    Red,
22    Incomplete,
23    Yellow,
24    Error,
25    Cancelled,
26}
27impl State {
28    pub fn is_terminal(self) -> bool {
29        matches!(
30            self,
31            Self::Green
32                | Self::Red
33                | Self::Incomplete
34                | Self::Yellow
35                | Self::Error
36                | Self::Cancelled
37        )
38    }
39}
40#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
41pub struct Counts {
42    pub total: u64,
43    pub completed: u64,
44    pub running: u64,
45    pub waiting: u64,
46    pub pending: u64,
47}
48#[derive(Debug, Clone, Serialize)]
49pub struct RunObservation {
50    pub run: RunId,
51    pub state: State,
52    pub counts: Counts,
53    #[serde(serialize_with = "seconds")]
54    pub elapsed: Duration,
55    #[serde(serialize_with = "seconds")]
56    pub next_check: Duration,
57}
58
59fn seconds<S: serde::Serializer>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error> {
60    serializer.serialize_u64(duration.as_secs())
61}
62
63fn counts(states: impl IntoIterator<Item = (RequestStatus, u64)>) -> Counts {
64    states
65        .into_iter()
66        .fold(Counts::default(), |mut counts, (state, count)| {
67            counts.total += count;
68            match state {
69                RequestStatus::Green
70                | RequestStatus::Provisional
71                | RequestStatus::Red
72                | RequestStatus::Error => counts.completed += count,
73                RequestStatus::Running => counts.running += count,
74                RequestStatus::WaitingHuman
75                | RequestStatus::Deferred
76                | RequestStatus::WaitDependency => counts.waiting += count,
77                _ => counts.pending += count,
78            }
79            counts
80        })
81}
82pub(super) fn state_for(
83    status: RunStatus,
84    control: Option<&super::Control>,
85    counts: &Counts,
86) -> State {
87    match status {
88        RunStatus::Green => State::Green,
89        RunStatus::Red => State::Red,
90        RunStatus::Incomplete => State::Incomplete,
91        RunStatus::Yellow => State::Yellow,
92        RunStatus::Error
93            if control
94                .is_some_and(|control| control.failure_code == Some(FailureCode::Cancelled)) =>
95        {
96            State::Cancelled
97        }
98        RunStatus::Error => State::Error,
99        RunStatus::WaitingHuman
100            if control.is_some_and(|control| control.cancellation_requested) =>
101        {
102            State::Cancelling
103        }
104        RunStatus::WaitingHuman => State::WaitingHuman,
105        RunStatus::Running => {
106            if let Some(control) = control {
107                if control.cancellation_requested {
108                    return State::Cancelling;
109                }
110                return match control.activity {
111                    Activity::Preparing => State::Preparing,
112                    Activity::Executing | Activity::Publishing => State::Running,
113                    Activity::Waiting => State::Waiting,
114                    Activity::Cancelling => State::Cancelling,
115                };
116            }
117            if counts.waiting > 0 && counts.running == 0 {
118                State::Waiting
119            } else {
120                State::Running
121            }
122        }
123    }
124}
125pub(super) fn state(run: &Run, requests: &[std::sync::Arc<Request>]) -> State {
126    state_for(
127        run.status(),
128        run.control.as_ref(),
129        &counts(requests.iter().map(|request| (request.status(), 1))),
130    )
131}
132
133pub async fn view(context: &Context, id: &RunId) -> Result<RunView, String> {
134    let view = store::read_run(&context.state, id).await?;
135    if view.run.repo_path != context.repo {
136        return Err("Run not found in this repository.".into());
137    }
138    super::reconcile(&context.state, view).await
139}
140pub async fn status(context: &Context, id: &RunId) -> Result<RunObservation, String> {
141    const NEXT_CHECK: Duration = Duration::from_secs(30); // One ordinary bounded observation, not a polling cadence.
142    let mut saved = store::run_control::observe(&context.state, id).await?;
143    if saved.repo != context.repo {
144        return Err("Run not found in this repository.".into());
145    }
146    let owner = saved
147        .control
148        .as_ref()
149        .and_then(|control| match control.worker {
150            super::Worker::Starting { launcher } => Some(launcher),
151            super::Worker::Admitted { owner } => Some(owner),
152            super::Worker::Legacy => None,
153        });
154    if saved.status.is_open()
155        && owner.is_some()
156        && !crate::process::is_alive(owner).map_err(|e| e.to_string())?
157    {
158        view(context, id).await?;
159        saved = store::run_control::observe(&context.state, id).await?;
160    }
161    let counts = counts(saved.counts);
162    let state = state_for(saved.status, saved.control.as_ref(), &counts);
163    Ok(RunObservation {
164        run: id.clone(),
165        state,
166        counts,
167        elapsed: saved
168            .completed_at
169            .unwrap_or_else(Timestamp::now)
170            .since(saved.created_at),
171        next_check: if state.is_terminal() {
172            Duration::ZERO
173        } else {
174            NEXT_CHECK
175        },
176    })
177}
178
179/// The optional bound covers registration and all reads. At its deadline return the last
180/// authoritative snapshot; cancellation disposes only this observer, never the Run.
181pub async fn wait(
182    context: &Context,
183    id: &RunId,
184    bound: Option<Duration>,
185    cancellation: CancellationToken,
186) -> Result<RunObservation, String> {
187    if bound.is_some_and(|bound| bound.is_zero()) {
188        return status(context, id).await;
189    }
190    let deadline = bound.map(|bound| tokio::time::Instant::now() + bound);
191    let mut last = None;
192    let mut subscription = None;
193    let result = {
194        let observing = async {
195            let changes =
196                subscription.insert(crate::changes::Subscription::new(&context.state).await);
197            loop {
198                let current = status(context, id).await?;
199                let ended = current.state.is_terminal();
200                last = Some(current);
201                if ended {
202                    return Ok::<(), String>(());
203                }
204                changes.next_reconciled().await;
205            }
206        };
207        tokio::pin!(observing);
208        tokio::select! {
209            result = &mut observing => result,
210            _ = cancellation.cancelled() => Err("Run observation was cancelled.".into()),
211            _ = async { if let Some(deadline) = deadline { tokio::time::sleep_until(deadline).await; } else { std::future::pending::<()>().await; } } => Ok(()),
212        }
213    };
214    // Completion, deadline and cancellation all release this observer before returning.
215    // Otherwise CLI runtime shutdown can interrupt the detached watch's SQLite cleanup.
216    let closed = match subscription {
217        Some(changes) => changes.close().await,
218        None => Ok(()),
219    };
220    match (result, closed) {
221        (Ok(()), Ok(())) => {
222            last.ok_or_else(|| "State unavailable within the observation bound.".into())
223        }
224        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
225        (Err(error), Err(close)) => Err(format!(
226            "{error}; closing observation subscription: {close}"
227        )),
228    }
229}
230
231#[cfg(test)]
232mod tests {
233    use super::*;
234
235    #[test]
236    fn core_durations_keep_the_existing_integer_second_wire_values() {
237        let observed = RunObservation {
238            run: "run-wire".parse().unwrap(),
239            state: State::Running,
240            counts: Counts::default(),
241            elapsed: Duration::from_millis(1250),
242            next_check: Duration::from_secs(30),
243        };
244        let wire = serde_json::to_value(observed).unwrap();
245        assert_eq!(wire["elapsed"], 1);
246        assert_eq!(wire["next_check"], 30);
247    }
248}
artifactize_core/runs/observe.rs

artifactize_core/runs/
observe.rs
