dpm_engine/run/
start.rs

1//! Turning a start request into the record of what the run observed.
2
3use super::RunError;
4use chrono::{DateTime, Utc};
5use dpm_model::{
6    ActorId, ActorKind, ContractObservation, LineageId, Observation, Plan, RunRecord, RunStart,
7    WorkStatus,
8};
9
10/// Validate a start against the plan and capture the contract it observes.
11///
12/// The task must be an executable task its executor already holds, claimed or started. `writer` is
13/// the principal recording the run: the executor itself, or a human or service acting for it.
14/// `lineage` is the history `plan` was read from; the record keeps it with the plan revision so the
15/// observation stays attributable after a restore. `now` is DPM's own clock, the receipt time.
16pub fn observe_start(
17    plan: &Plan,
18    lineage: LineageId,
19    writer: &ActorId,
20    request: &RunStart,
21    now: DateTime<Utc>,
22) -> Result<RunRecord, RunError> {
23    request.validate()?;
24    super::transition::authorize_writer(writer, &request.executor, request.id, "start")?;
25    if request.observation == Observation::Managed && writer.kind != ActorKind::Service {
26        return Err(RunError::ManagedNeedsService(request.id));
27    }
28    let work = plan
29        .work_items
30        .get(&request.work)
31        .ok_or(RunError::MissingWork(request.work))?;
32    if !work.is_executable() {
33        return Err(RunError::NotATask(work.id));
34    }
35    if !matches!(
36        work.execution.status,
37        WorkStatus::Claimed | WorkStatus::InProgress
38    ) {
39        return Err(RunError::NotExecuting {
40            work: work.id,
41            status: work.execution.status,
42        });
43    }
44    if work.execution.owner.as_ref() != Some(&request.executor) {
45        return Err(RunError::NotOwner {
46            work: work.id,
47            executor: request.executor.clone(),
48            owner: work.execution.owner.clone(),
49        });
50    }
51    let exact_sources = request
52        .sources
53        .iter()
54        .map(|source| super::source::resolve(plan, work, source))
55        .collect::<Result<Vec<_>, _>>()?;
56    Ok(RunRecord {
57        id: request.id,
58        work: work.id,
59        executor: request.executor.clone(),
60        recorded_by: writer.clone(),
61        parent: request.parent,
62        session: request.session.clone(),
63        observation: request.observation,
64        sources: request.sources.clone(),
65        exact_sources,
66        contract: ContractObservation {
67            workspace_id: plan.workspace.id,
68            lineage_id: lineage,
69            revision: plan.revision,
70            work_key: work.key.clone(),
71            prior_submissions: u32::try_from(work.execution.attempts.len()).unwrap_or(u32::MAX),
72            contract: work.contract.clone(),
73        },
74        observed_started_at: request.observed_at,
75        started_at: now,
76    })
77}
dpm_engine/run/start.rs

dpm_engine/run/
start.rs
