artifactize_core/runs/
mod.rs

1//! Artifactize-owned verification: durable submission, observation and explicit cancellation.
2#[cfg(test)]
3mod lifetime_tests;
4mod observe;
5mod result;
6mod watch;
7mod worker;
8use crate::{
9    process,
10    project::{VerifyOptions, selection::Selection},
11    store::{self, Receipts},
12    types::{FailureCode, RequestKey, RunId, RunStatus},
13};
14pub use observe::{Counts, RunObservation, State, status, view, wait};
15pub use result::{Finding, ResultCursor, ResultState, RunResultPage, RunVerdict, result};
16use serde::{Deserialize, Serialize};
17use std::path::{Path, PathBuf};
18pub use watch::{RunEvent, watch};
19#[doc(hidden)]
20pub use worker::{execute, supervise};
21
22/// Per-Run execution modes. Future modes extend this type, not transport signatures.
23#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
24#[serde(default)]
25pub struct RunModes {
26    pub speculate: bool,
27    pub defer_human: bool,
28}
29impl RunModes {
30    pub(crate) fn is_default(&self) -> bool {
31        self == &Self::default()
32    }
33}
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(rename_all = "snake_case")]
37pub enum Activity {
38    Preparing,
39    Executing,
40    Publishing,
41    Waiting,
42    Cancelling,
43}
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(tag = "state", rename_all = "snake_case")]
47pub enum Worker {
48    Starting {
49        launcher: process::ChildIdentity,
50    },
51    Admitted {
52        owner: process::ChildIdentity,
53    },
54    /// Old records have no worker identity and are not evidence of owner death.
55    Legacy,
56}
57#[derive(Debug, Clone, Serialize, Deserialize)]
58#[serde(rename_all = "camelCase")]
59pub struct Control {
60    pub worker: Worker,
61    /// The supervisor keeps worker files open until its executor has exited and been checked.
62    #[serde(default, skip_serializing_if = "Option::is_none")]
63    pub supervisor: Option<process::ChildIdentity>,
64    pub activity: Activity,
65    #[serde(default)]
66    pub cancellation_requested: bool,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub failure_code: Option<FailureCode>,
69}
70impl Control {
71    pub(crate) fn starting() -> Result<Self, String> {
72        Ok(Self {
73            worker: Worker::Starting {
74                launcher: process::child_identity(std::process::id()).map_err(|e| e.to_string())?,
75            },
76            supervisor: None,
77            activity: Activity::Preparing,
78            cancellation_requested: false,
79            failure_code: None,
80        })
81    }
82    pub(crate) fn legacy() -> Self {
83        Self {
84            worker: Worker::Legacy,
85            supervisor: None,
86            activity: Activity::Executing,
87            cancellation_requested: false,
88            failure_code: None,
89        }
90    }
91}
92
93/// Bound repository/state and trusted executable, resolved once by the interface at startup.
94#[derive(Debug, Clone)]
95pub struct Context {
96    pub(crate) repo: PathBuf,
97    pub(crate) state: PathBuf,
98    executable: PathBuf,
99}
100impl Context {
101    /// The canonical repository this endpoint serves.
102    pub fn repo(&self) -> &Path {
103        &self.repo
104    }
105
106    /// The canonical state directory shared by this endpoint and its Run workers.
107    pub fn state(&self) -> &Path {
108        &self.state
109    }
110
111    pub fn new(repo: &Path, state: Option<&Path>, executable: &Path) -> Result<Self, String> {
112        Ok(Self {
113            repo: crate::workspace::canonical_target(repo).map_err(|e| e.to_string())?,
114            state: crate::workspace::canonical_target(&store::state_dir(state)?)
115                .map_err(|e| e.to_string())?,
116            executable: executable.to_path_buf(),
117        })
118    }
119    /// Public-library convenience; packaged callers normally pass their own executable to `new`.
120    pub fn discover(repo: &Path, state: Option<&Path>) -> Result<Self, String> {
121        let executable = process::artifactize_executable()?;
122        Self::new(repo, state, &executable)
123    }
124}
125
126pub async fn start(
127    context: &Context,
128    selection: &Selection,
129    options: &VerifyOptions,
130    request_key: Option<RequestKey>,
131) -> Result<RunObservation, String> {
132    // Keys are attempt identity, not checkout identity. Retry before discovering declarations.
133    if let Some(key) = &request_key
134        && context.state.join(store::DATABASE).is_file()
135    {
136        let receipts = Receipts::open(&context.state, &context.repo).await?;
137        let result = receipts.keyed_run(&context.repo, key).await;
138        let closed = receipts.close().await;
139        if let Some(id) = result.and_then(|id| closed.map(|()| id))? {
140            return status(context, &id).await;
141        }
142    }
143    let (id, accepted) = crate::project::verify::reserve(
144        &context.repo,
145        Some(&context.state),
146        selection,
147        options,
148        request_key.as_ref(),
149    )
150    .await?;
151    if accepted {
152        let receipts = Receipts::open(&context.state, &context.repo).await?;
153        let result = async {
154            let launch =
155                process::launch_run(&context.executable, &id, &context.repo, &context.state);
156            let admitted = match launch {
157                Ok(identity) => receipts.admit(&id, identity).await,
158                Err(error) => Err(error.to_string()),
159            };
160            if let Err(error) = admitted {
161                worker::fail(&receipts, &id, FailureCode::SpawnFailed, error.clone()).await?;
162                return Err(format!("Run {id} handoff failed: {error}"));
163            }
164            Ok(())
165        }
166        .await;
167        let closed = receipts.close().await;
168        result.and(closed)?;
169    }
170    status(context, &id).await
171}
172
173#[derive(Debug, Clone, Serialize)]
174pub struct CancelAcknowledgement {
175    pub run: RunId,
176    pub state: State,
177    pub cancellation_requested: bool,
178}
179pub async fn cancel(context: &Context, id: &RunId) -> Result<CancelAcknowledgement, String> {
180    let current = view(context, id).await?;
181    let receipts = Receipts::open(&context.state, &context.repo).await?;
182    let result = receipts.request_cancel(&current.run.id).await;
183    let closed = receipts.close().await;
184    let run = result.and_then(|run| closed.map(|()| run))?;
185    Ok(CancelAcknowledgement {
186        run: id.clone(),
187        state: observe::state(&run, &current.requests),
188        cancellation_requested: run
189            .control
190            .as_ref()
191            .is_some_and(|control| control.cancellation_requested),
192    })
193}
194
195pub(crate) async fn reconcile(
196    state: &Path,
197    view: store::RunView,
198) -> Result<store::RunView, String> {
199    if view.run.status() == RunStatus::Running
200        && let Some(control) = &view.run.control
201    {
202        let owner = match control.worker {
203            Worker::Starting { launcher } => Some(launcher),
204            Worker::Admitted { owner } => Some(owner),
205            Worker::Legacy => None,
206        };
207        if owner.is_some() && !process::is_alive(owner).map_err(|e| e.to_string())? {
208            worker::recover(state, &view).await?;
209            return store::read_run(state, &view.run.id).await;
210        }
211    }
212    Ok(view)
213}
artifactize_core/runs/mod.rs

artifactize_core/runs/
mod.rs
