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

artifactize_core/runs/
observe.rs
