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

artifactize_core/runs/
mod.rs
