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    // Preparing the Run read the whole workspace and built its definitions, all freed by now;
170    // a caller that goes on to wait for the Run should not keep that memory while it waits.
171    crate::platform::release_free_memory();
172    status(context, &id).await
173}
174
175#[derive(Debug, Clone, Serialize)]
176pub struct CancelAcknowledgement {
177    pub run: RunId,
178    pub state: State,
179    pub cancellation_requested: bool,
180}
181pub async fn cancel(context: &Context, id: &RunId) -> Result<CancelAcknowledgement, String> {
182    // The compact observation checks the repository and settles a Run whose worker died.
183    let current = status(context, id).await?;
184    let receipts = Receipts::open(&context.state, &context.repo).await?;
185    let result = receipts.request_cancel(id).await;
186    let closed = receipts.close().await;
187    let (saved, control) = result.and_then(|saved| closed.map(|()| saved))?;
188    Ok(CancelAcknowledgement {
189        run: id.clone(),
190        state: observe::state_for(saved, control.as_ref(), &current.counts),
191        cancellation_requested: control
192            .as_ref()
193            .is_some_and(|control| control.cancellation_requested),
194    })
195}
196
197pub(crate) async fn reconcile(
198    state: &Path,
199    view: store::RunView,
200) -> Result<store::RunView, String> {
201    if view.run.status().is_open()
202        && let Some(control) = &view.run.control
203    {
204        let owner = match control.worker {
205            Worker::Starting { launcher } => Some(launcher),
206            Worker::Admitted { owner } => Some(owner),
207            Worker::Legacy => None,
208        };
209        if owner.is_some() && !process::is_alive(owner).map_err(|e| e.to_string())? {
210            worker::recover(state, &view).await?;
211            return store::read_run(state, &view.run.id).await;
212        }
213    }
214    Ok(view)
215}
artifactize_core/runs/mod.rs

artifactize_core/runs/
mod.rs
