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}
82fn state_for(status: RunStatus, control: Option<&super::Control>, counts: &Counts) -> State {
83    match status {
84        RunStatus::Green => State::Green,
85        RunStatus::Red => State::Red,
86        RunStatus::Incomplete => State::Incomplete,
87        RunStatus::Yellow => State::Yellow,
88        RunStatus::Error
89            if control
90                .is_some_and(|control| control.failure_code == Some(FailureCode::Cancelled)) =>
91        {
92            State::Cancelled
93        }
94        RunStatus::Error => State::Error,
95        RunStatus::WaitingHuman
96            if control.is_some_and(|control| control.cancellation_requested) =>
97        {
98            State::Cancelling
99        }
100        RunStatus::WaitingHuman => State::WaitingHuman,
101        RunStatus::Running => {
102            if let Some(control) = control {
103                if control.cancellation_requested {
104                    return State::Cancelling;
105                }
106                return match control.activity {
107                    Activity::Preparing => State::Preparing,
108                    Activity::Executing | Activity::Publishing => State::Running,
109                    Activity::Waiting => State::Waiting,
110                    Activity::Cancelling => State::Cancelling,
111                };
112            }
113            if counts.waiting > 0 && counts.running == 0 {
114                State::Waiting
115            } else {
116                State::Running
117            }
118        }
119    }
120}
121pub(super) fn state(run: &Run, requests: &[Request]) -> State {
122    state_for(
123        run.status(),
124        run.control.as_ref(),
125        &counts(requests.iter().map(|request| (request.status(), 1))),
126    )
127}
128
129pub async fn view(context: &Context, id: &RunId) -> Result<RunView, String> {
130    let view = store::read_run(&context.state, id).await?;
131    if view.run.repo_path != context.repo {
132        return Err("Run not found in this repository.".into());
133    }
134    super::reconcile(&context.state, view).await
135}
136pub async fn status(context: &Context, id: &RunId) -> Result<RunObservation, String> {
137    const NEXT_CHECK: Duration = Duration::from_secs(30); // One ordinary bounded observation, not a polling cadence.
138    let mut saved = store::run_control::observe(&context.state, id).await?;
139    if saved.repo != context.repo {
140        return Err("Run not found in this repository.".into());
141    }
142    let owner = saved
143        .control
144        .as_ref()
145        .and_then(|control| match control.worker {
146            super::Worker::Starting { launcher } => Some(launcher),
147            super::Worker::Admitted { owner } => Some(owner),
148            super::Worker::Legacy => None,
149        });
150    if saved.status.is_open()
151        && owner.is_some()
152        && !crate::process::is_alive(owner).map_err(|e| e.to_string())?
153    {
154        view(context, id).await?;
155        saved = store::run_control::observe(&context.state, id).await?;
156    }
157    let counts = counts(saved.counts);
158    let state = state_for(saved.status, saved.control.as_ref(), &counts);
159    Ok(RunObservation {
160        run: id.clone(),
161        state,
162        counts,
163        elapsed: saved
164            .completed_at
165            .unwrap_or_else(Timestamp::now)
166            .since(saved.created_at),
167        next_check: if state.is_terminal() {
168            Duration::ZERO
169        } else {
170            NEXT_CHECK
171        },
172    })
173}
174
175/// The optional bound covers registration and all reads. At its deadline return the last
176/// authoritative snapshot; cancellation disposes only this observer, never the Run.
177pub async fn wait(
178    context: &Context,
179    id: &RunId,
180    bound: Option<Duration>,
181    cancellation: CancellationToken,
182) -> Result<RunObservation, String> {
183    if bound.is_some_and(|bound| bound.is_zero()) {
184        return status(context, id).await;
185    }
186    let deadline = bound.map(|bound| tokio::time::Instant::now() + bound);
187    let mut last = None;
188    let mut subscription = None;
189    let result = {
190        let observing = async {
191            let changes =
192                subscription.insert(crate::changes::Subscription::new(&context.state).await);
193            loop {
194                let current = status(context, id).await?;
195                let ended = current.state.is_terminal();
196                last = Some(current);
197                if ended {
198                    return Ok::<(), String>(());
199                }
200                changes.next_reconciled().await;
201            }
202        };
203        tokio::pin!(observing);
204        tokio::select! {
205            result = &mut observing => result,
206            _ = cancellation.cancelled() => Err("Run observation was cancelled.".into()),
207            _ = async { if let Some(deadline) = deadline { tokio::time::sleep_until(deadline).await; } else { std::future::pending::<()>().await; } } => Ok(()),
208        }
209    };
210    // Completion, deadline and cancellation all release this observer before returning.
211    // Otherwise CLI runtime shutdown can interrupt the detached watch's SQLite cleanup.
212    let closed = match subscription {
213        Some(changes) => changes.close().await,
214        None => Ok(()),
215    };
216    match (result, closed) {
217        (Ok(()), Ok(())) => {
218            last.ok_or_else(|| "State unavailable within the observation bound.".into())
219        }
220        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
221        (Err(error), Err(close)) => Err(format!(
222            "{error}; closing observation subscription: {close}"
223        )),
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use super::*;
230
231    #[test]
232    fn core_durations_keep_the_existing_integer_second_wire_values() {
233        let observed = RunObservation {
234            run: "run-wire".parse().unwrap(),
235            state: State::Running,
236            counts: Counts::default(),
237            elapsed: Duration::from_millis(1250),
238            next_check: Duration::from_secs(30),
239        };
240        let wire = serde_json::to_value(observed).unwrap();
241        assert_eq!(wire["elapsed"], 1);
242        assert_eq!(wire["next_check"], 30);
243    }
244}
artifactize_core/runs/observe.rs

artifactize_core/runs/
observe.rs
