a3s_box_runtime/oci/build/
execution.rs

1//! Plan-bound execution over the one native Box OCI build engine.
2
3use std::path::{Path, PathBuf};
4use std::sync::Arc;
5use std::time::Duration;
6
7use a3s_box_core::error::BoxError;
8use thiserror::Error;
9
10use super::cache::{inspect_build_cache_artifact, BuildCacheExportIdentity, RecordedBuildCache};
11use super::engine::{build_supervised, BuildExecutionControl};
12use super::receipt::{
13    inspect_stored_output, BuildExecutionLease, BuildOperationJournal, LockedBuildOperation,
14    PersistedBuildOperation, PersistedBuildPhase, SupervisedBuildOperation,
15};
16use super::{
17    build, BoxBuildOptions, BoxBuildPlan, BoxBuildPlanError, BuildCachePolicy,
18    BuildCancellationOutcome, BuildOperationIdentity, BuildOutputReceipt, BuildReceiptError,
19    BuildResult, RecordedBuildResult, RecordedBuildStatus,
20};
21use crate::oci::ImageStore;
22
23mod supervision;
24
25use supervision::{fence_run_process, JournalBuildObserver};
26
27const EXECUTION_POLL_INTERVAL: Duration = Duration::from_millis(25);
28
29/// Successful native output bound to the canonical admitted build plan.
30#[derive(Debug)]
31pub(super) struct PlannedBuildResult {
32    /// Canonical A3S ACL plan identity.
33    pub plan_digest: String,
34    /// Durable typed OCI output owned by the Box image store.
35    pub output: BuildResult,
36    /// Portable native cache artifact committed with the image output.
37    pub cache: Option<RecordedBuildCache>,
38}
39
40/// Stable failure boundary for plan admission, compilation, and execution.
41#[derive(Debug, Error)]
42pub enum BuildPlanExecutionError {
43    /// The closed build plan could not be canonicalized or compiled.
44    #[error(transparent)]
45    Plan(#[from] BoxBuildPlanError),
46    /// The native build engine or durable image store rejected the operation.
47    #[error(transparent)]
48    Build(#[from] BoxError),
49    /// Durable receipt identity, persistence, or output validation failed.
50    #[error(transparent)]
51    Receipt(#[from] BuildReceiptError),
52    /// A durable cancellation completed.
53    #[error("Box build operation {operation_id} was cancelled: {message}")]
54    Cancelled {
55        operation_id: String,
56        message: String,
57    },
58    /// A durable supervised execution failed.
59    #[error("Box build operation {operation_id} failed: {message}")]
60    Failed {
61        operation_id: String,
62        message: String,
63    },
64}
65
66/// Compile and execute one canonical plan through Box's existing native engine.
67///
68/// The returned layout path points at the durable image-store copy rather than
69/// the temporary build workspace, so a caller can capture or publish the exact
70/// OCI graph after this future returns.
71#[cfg_attr(not(test), allow(dead_code))]
72pub(super) async fn execute_build_plan(
73    plan: &BoxBuildPlan,
74    source_root: &Path,
75    options: BoxBuildOptions,
76    store: Arc<ImageStore>,
77) -> Result<PlannedBuildResult, BuildPlanExecutionError> {
78    let plan_digest = plan.canonical_digest()?;
79    let config = plan.compile(source_root, options)?;
80    let output = build(config, store).await?;
81    Ok(PlannedBuildResult {
82        plan_digest,
83        output,
84        cache: None,
85    })
86}
87
88async fn execute_supervised_build_plan(
89    identity: &BuildOperationIdentity,
90    plan: &BoxBuildPlan,
91    source_root: &Path,
92    options: BoxBuildOptions,
93    store: Arc<ImageStore>,
94    workspace: &Path,
95    control: BuildExecutionControl,
96) -> Result<PlannedBuildResult, BuildPlanExecutionError> {
97    let plan_digest = plan.canonical_digest()?;
98    let config = plan.compile(source_root, options)?;
99    let cache_identity = (plan.cache() == BuildCachePolicy::ContentAddressed)
100        .then(|| {
101            BuildCacheExportIdentity::new(
102                identity.source_digest(),
103                plan_digest.clone(),
104                plan.platform().clone(),
105            )
106        })
107        .transpose()?;
108    let result = build_supervised(config, store, workspace, control, cache_identity).await?;
109    Ok(PlannedBuildResult {
110        plan_digest,
111        output: result.output,
112        cache: result.cache,
113    })
114}
115
116/// Start or exactly replay one supervised plan-bound native build.
117///
118/// Starting is non-blocking. The returned status comes from the existing
119/// receipt journal; callers use [`inspect_recorded_build_status`] and
120/// [`cancel_recorded_build_plan`] against that same authority.
121pub async fn start_recorded_build_plan(
122    identity: &BuildOperationIdentity,
123    plan: &BoxBuildPlan,
124    source_root: &Path,
125    quiet: bool,
126    store: Arc<ImageStore>,
127) -> Result<RecordedBuildStatus, BuildPlanExecutionError> {
128    start_recorded_build_plan_internal(identity, plan, source_root, quiet, store)
129        .await
130        .map(|start| start.status)
131}
132
133/// Execute or exactly replay one supervised plan-bound native build.
134///
135/// This compatibility API starts through the same durable state machine and
136/// waits for typed inspection to reach a terminal state. It owns no second
137/// execution path, lock, receipt, or cleanup policy.
138pub async fn execute_recorded_build_plan(
139    identity: &BuildOperationIdentity,
140    plan: &BoxBuildPlan,
141    source_root: &Path,
142    quiet: bool,
143    store: Arc<ImageStore>,
144) -> Result<RecordedBuildResult, BuildPlanExecutionError> {
145    let start =
146        start_recorded_build_plan_internal(identity, plan, source_root, quiet, Arc::clone(&store))
147            .await?;
148    let started_here = start.started_here;
149    let mut status = Some(start.status);
150    loop {
151        let current = match status.take() {
152            Some(status) => status,
153            None => inspect_recorded_build_status(identity, plan, &store)
154                .await?
155                .ok_or_else(|| BuildReceiptError::Conflict {
156                    operation_id: identity.operation_id().to_string(),
157                    message: "supervised operation disappeared while execution was waiting"
158                        .to_string(),
159                })?,
160        };
161        match current {
162            RecordedBuildStatus::Running | RecordedBuildStatus::Cancelling => {
163                tokio::time::sleep(EXECUTION_POLL_INTERVAL).await;
164            }
165            RecordedBuildStatus::Cancelled { message } => {
166                return Err(BuildPlanExecutionError::Cancelled {
167                    operation_id: identity.operation_id().to_string(),
168                    message,
169                });
170            }
171            RecordedBuildStatus::Failed { message } => {
172                return Err(BuildPlanExecutionError::Failed {
173                    operation_id: identity.operation_id().to_string(),
174                    message,
175                });
176            }
177            RecordedBuildStatus::Succeeded(mut result) => {
178                result.replayed = !started_here;
179                return Ok(*result);
180            }
181        }
182    }
183}
184
185/// Inspect and reconcile the one durable build-operation state machine.
186///
187/// Inspection never waits for a live build. A nonblocking execution lease
188/// distinguishes a live owner from a crashed one; stale work is fenced and its
189/// operation-owned workspace is reclaimed before a terminal status is written.
190pub async fn inspect_recorded_build_status(
191    identity: &BuildOperationIdentity,
192    plan: &BoxBuildPlan,
193    store: &ImageStore,
194) -> Result<Option<RecordedBuildStatus>, BuildPlanExecutionError> {
195    let plan_digest = plan.canonical_digest()?;
196    let cache_policy = plan.cache();
197    let journal = BuildOperationJournal::for_image_store(store, identity.operation_id()).await?;
198    let locked = journal.lock(identity.operation_id()).await?;
199    let Some(record) = locked.read().await? else {
200        return Ok(None);
201    };
202    match record {
203        PersistedBuildOperation::Succeeded(receipt) => recover_recorded_build(
204            identity,
205            &plan_digest,
206            cache_policy,
207            *receipt,
208            store,
209            &journal,
210        )
211        .await
212        .map(Box::new)
213        .map(RecordedBuildStatus::Succeeded)
214        .map(Some),
215        PersistedBuildOperation::Supervised(operation) => {
216            operation.require_identity(identity, &plan_digest, cache_policy)?;
217            match operation.phase {
218                PersistedBuildPhase::Cancelled | PersistedBuildPhase::Failed => {
219                    Ok(Some(status_from_terminal_operation(&operation)?))
220                }
221                PersistedBuildPhase::Running | PersistedBuildPhase::Cancelling => {
222                    let Some(lease) = journal.try_execution_lease(identity.operation_id()).await?
223                    else {
224                        return Ok(Some(status_from_live_operation(&operation)));
225                    };
226                    recover_stale_operation(
227                        identity,
228                        &plan_digest,
229                        operation,
230                        store,
231                        &journal,
232                        &locked,
233                        lease,
234                    )
235                    .await
236                    .map(Some)
237                }
238            }
239        }
240        PersistedBuildOperation::Pending(pending) => {
241            pending.require_identity(identity, &plan_digest)?;
242            if let Some(result) =
243                adopt_committed_output(identity, &plan_digest, None, store, &journal, &locked)
244                    .await?
245            {
246                return Ok(Some(RecordedBuildStatus::Succeeded(Box::new(result))));
247            }
248            let Some(_lease) = journal.try_execution_lease(identity.operation_id()).await? else {
249                return Ok(Some(RecordedBuildStatus::Running));
250            };
251            journal.cleanup_workspace(identity.operation_id()).await?;
252            journal
253                .cleanup_cache_export(identity.operation_id())
254                .await?;
255            let mut operation = SupervisedBuildOperation::from_pending(
256                &pending,
257                identity,
258                &plan_digest,
259                cache_policy,
260            )?;
261            operation.finish(
262                PersistedBuildPhase::Failed,
263                "legacy build intent has no live execution owner or committed output".to_string(),
264            );
265            let operation = locked.write_supervised(operation).await?;
266            Ok(Some(status_from_terminal_operation(&operation)?))
267        }
268    }
269}
270
271/// Inspect only a successful terminal result for compatibility.
272///
273/// New code should use [`inspect_recorded_build_status`] to distinguish live,
274/// cancelling, cancelled, failed, and successful states.
275pub async fn inspect_recorded_build_plan(
276    identity: &BuildOperationIdentity,
277    plan: &BoxBuildPlan,
278    store: &ImageStore,
279) -> Result<Option<RecordedBuildResult>, BuildPlanExecutionError> {
280    Ok(
281        match inspect_recorded_build_status(identity, plan, store).await? {
282            Some(RecordedBuildStatus::Succeeded(result)) => Some(*result),
283            Some(
284                RecordedBuildStatus::Running
285                | RecordedBuildStatus::Cancelling
286                | RecordedBuildStatus::Cancelled { .. }
287                | RecordedBuildStatus::Failed { .. },
288            )
289            | None => None,
290        },
291    )
292}
293
294/// Durably request cancellation through the existing receipt journal.
295pub async fn cancel_recorded_build_plan(
296    identity: &BuildOperationIdentity,
297    plan: &BoxBuildPlan,
298    store: &ImageStore,
299) -> Result<BuildCancellationOutcome, BuildPlanExecutionError> {
300    let plan_digest = plan.canonical_digest()?;
301    let cache_policy = plan.cache();
302    let journal = BuildOperationJournal::for_image_store(store, identity.operation_id()).await?;
303    let locked = journal.lock(identity.operation_id()).await?;
304    let Some(record) = locked.read().await? else {
305        return Ok(BuildCancellationOutcome::NotFound);
306    };
307    match record {
308        PersistedBuildOperation::Succeeded(receipt) => {
309            receipt.require_identity(identity, &plan_digest, cache_policy)?;
310            Ok(BuildCancellationOutcome::AlreadyTerminal)
311        }
312        PersistedBuildOperation::Pending(pending) => {
313            pending.require_identity(identity, &plan_digest)?;
314            if adopt_committed_output(identity, &plan_digest, None, store, &journal, &locked)
315                .await?
316                .is_some()
317            {
318                return Ok(BuildCancellationOutcome::AlreadyTerminal);
319            }
320            let Some(_lease) = journal.try_execution_lease(identity.operation_id()).await? else {
321                return Err(BuildReceiptError::Conflict {
322                    operation_id: identity.operation_id().to_string(),
323                    message: "legacy pending operation has an unknown live execution owner"
324                        .to_string(),
325                }
326                .into());
327            };
328            journal.cleanup_workspace(identity.operation_id()).await?;
329            journal
330                .cleanup_cache_export(identity.operation_id())
331                .await?;
332            let mut operation = SupervisedBuildOperation::from_pending(
333                &pending,
334                identity,
335                &plan_digest,
336                cache_policy,
337            )?;
338            operation.request_cancellation();
339            operation.finish(
340                PersistedBuildPhase::Cancelled,
341                "cancelled before native execution started".to_string(),
342            );
343            locked.write_supervised(operation).await?;
344            Ok(BuildCancellationOutcome::Requested)
345        }
346        PersistedBuildOperation::Supervised(mut operation) => {
347            operation.require_identity(identity, &plan_digest, cache_policy)?;
348            match operation.phase {
349                PersistedBuildPhase::Cancelled => {
350                    return Ok(BuildCancellationOutcome::AlreadyCancelled);
351                }
352                PersistedBuildPhase::Failed => {
353                    return Ok(BuildCancellationOutcome::AlreadyTerminal);
354                }
355                PersistedBuildPhase::Running | PersistedBuildPhase::Cancelling => {}
356            }
357            if adopt_committed_output(
358                identity,
359                &plan_digest,
360                operation.cache_policy(),
361                store,
362                &journal,
363                &locked,
364            )
365            .await?
366            .is_some()
367            {
368                return Ok(BuildCancellationOutcome::AlreadyTerminal);
369            }
370
371            let newly_requested = operation.request_cancellation();
372            let run_process = operation.run_process;
373            operation = locked.write_supervised(operation).await?;
374            if let Some(lease) = journal.try_execution_lease(identity.operation_id()).await? {
375                fence_run_process(run_process, identity.operation_id()).await?;
376                journal.cleanup_workspace(identity.operation_id()).await?;
377                journal
378                    .cleanup_cache_export(identity.operation_id())
379                    .await?;
380                operation.finish(
381                    PersistedBuildPhase::Cancelled,
382                    "cancelled while recovering an abandoned native execution".to_string(),
383                );
384                locked.write_supervised(operation).await?;
385                drop(lease);
386                return Ok(if newly_requested {
387                    BuildCancellationOutcome::Requested
388                } else {
389                    BuildCancellationOutcome::AlreadyRequested
390                });
391            }
392            drop(locked);
393            fence_run_process(run_process, identity.operation_id()).await?;
394            Ok(if newly_requested {
395                BuildCancellationOutcome::Requested
396            } else {
397                BuildCancellationOutcome::AlreadyRequested
398            })
399        }
400    }
401}
402
403/// Remove one terminal record and its operation-specific ImageStore reference.
404///
405/// Live or cancelling work must reach a terminal state first. The same journal
406/// cleanup removes the operation workspace; there is no parallel garbage
407/// collector or supervisor-owned image store.
408pub async fn remove_recorded_build_plan(
409    identity: &BuildOperationIdentity,
410    plan: &BoxBuildPlan,
411    store: &ImageStore,
412) -> Result<bool, BuildPlanExecutionError> {
413    let plan_digest = plan.canonical_digest()?;
414    let cache_policy = plan.cache();
415    let journal = BuildOperationJournal::for_image_store(store, identity.operation_id()).await?;
416    let locked = journal.lock(identity.operation_id()).await?;
417    let Some(record) = locked.read().await? else {
418        return Ok(false);
419    };
420    let (reference, expected_digest) = match record {
421        PersistedBuildOperation::Pending(pending) => {
422            pending.require_identity(identity, &plan_digest)?;
423            (identity.output_reference().to_string(), None)
424        }
425        PersistedBuildOperation::Succeeded(receipt) => {
426            let receipt = *receipt;
427            receipt.require_identity(identity, &plan_digest, cache_policy)?;
428            (
429                receipt.output.reference,
430                Some(receipt.output.descriptor.digest),
431            )
432        }
433        PersistedBuildOperation::Supervised(operation) => {
434            operation.require_identity(identity, &plan_digest, cache_policy)?;
435            if matches!(
436                operation.phase,
437                PersistedBuildPhase::Running | PersistedBuildPhase::Cancelling
438            ) {
439                return Err(BuildReceiptError::Conflict {
440                    operation_id: identity.operation_id().to_string(),
441                    message: "cancel and reconcile the live build before removal".to_string(),
442                }
443                .into());
444            }
445            (identity.output_reference().to_string(), None)
446        }
447    };
448    if let Some(stored) = store.get_checked(&reference).await? {
449        if expected_digest.is_some_and(|digest| stored.digest != digest) {
450            return Err(BuildReceiptError::OutputInvalid {
451                operation_id: identity.operation_id().to_string(),
452                message: "operation reference was rebound to another digest".to_string(),
453            }
454            .into());
455        }
456        store.remove(&reference).await?;
457    }
458    journal.cleanup_workspace(identity.operation_id()).await?;
459    journal
460        .cleanup_cache_export(identity.operation_id())
461        .await?;
462    locked.delete().await?;
463    Ok(true)
464}
465
466struct StartOutcome {
467    status: RecordedBuildStatus,
468    started_here: bool,
469}
470
471async fn start_recorded_build_plan_internal(
472    identity: &BuildOperationIdentity,
473    plan: &BoxBuildPlan,
474    source_root: &Path,
475    quiet: bool,
476    store: Arc<ImageStore>,
477) -> Result<StartOutcome, BuildPlanExecutionError> {
478    let plan_digest = plan.canonical_digest()?;
479    let cache_policy = plan.cache();
480    let journal = BuildOperationJournal::for_image_store(&store, identity.operation_id()).await?;
481    let locked = journal.lock(identity.operation_id()).await?;
482    let record = locked.read().await?;
483
484    match record {
485        Some(PersistedBuildOperation::Succeeded(receipt)) => {
486            let result = recover_recorded_build(
487                identity,
488                &plan_digest,
489                cache_policy,
490                *receipt,
491                &store,
492                &journal,
493            )
494            .await?;
495            return Ok(StartOutcome {
496                status: RecordedBuildStatus::Succeeded(Box::new(result)),
497                started_here: false,
498            });
499        }
500        Some(PersistedBuildOperation::Supervised(operation)) => {
501            operation.require_identity(identity, &plan_digest, cache_policy)?;
502            if matches!(
503                operation.phase,
504                PersistedBuildPhase::Cancelled | PersistedBuildPhase::Failed
505            ) {
506                return Ok(StartOutcome {
507                    status: status_from_terminal_operation(&operation)?,
508                    started_here: false,
509                });
510            }
511            let Some(lease) = journal.try_execution_lease(identity.operation_id()).await? else {
512                return Ok(StartOutcome {
513                    status: status_from_live_operation(&operation),
514                    started_here: false,
515                });
516            };
517            let status = recover_stale_operation(
518                identity,
519                &plan_digest,
520                operation,
521                &store,
522                &journal,
523                &locked,
524                lease,
525            )
526            .await?;
527            return Ok(StartOutcome {
528                status,
529                started_here: false,
530            });
531        }
532        Some(PersistedBuildOperation::Pending(pending)) => {
533            pending.require_identity(identity, &plan_digest)?;
534            if let Some(result) =
535                adopt_committed_output(identity, &plan_digest, None, &store, &journal, &locked)
536                    .await?
537            {
538                return Ok(StartOutcome {
539                    status: RecordedBuildStatus::Succeeded(Box::new(result)),
540                    started_here: false,
541                });
542            }
543            let lease = journal
544                .try_execution_lease(identity.operation_id())
545                .await?
546                .ok_or_else(|| BuildReceiptError::Conflict {
547                    operation_id: identity.operation_id().to_string(),
548                    message: "pending intent has an unknown live execution owner".to_string(),
549                })?;
550            let workspace = journal.prepare_workspace(identity.operation_id()).await?;
551            let operation = SupervisedBuildOperation::from_pending(
552                &pending,
553                identity,
554                &plan_digest,
555                cache_policy,
556            )?;
557            locked.write_supervised(operation).await?;
558            drop(locked);
559            spawn_supervised_build(SupervisedBuildTask {
560                identity: identity.clone(),
561                plan: plan.clone(),
562                plan_digest,
563                source_root: source_root.to_path_buf(),
564                quiet,
565                store,
566                journal,
567                workspace,
568                lease,
569            });
570            return Ok(StartOutcome {
571                status: RecordedBuildStatus::Running,
572                started_here: true,
573            });
574        }
575        None => {}
576    }
577
578    if store
579        .get_checked(identity.output_reference())
580        .await?
581        .is_some()
582    {
583        return Err(BuildReceiptError::OutputInvalid {
584            operation_id: identity.operation_id().to_string(),
585            message: "operation output exists without a persisted owning intent".to_string(),
586        }
587        .into());
588    }
589    let lease = journal
590        .try_execution_lease(identity.operation_id())
591        .await?
592        .ok_or_else(|| BuildReceiptError::Conflict {
593            operation_id: identity.operation_id().to_string(),
594            message: "execution lease exists without a persisted operation record".to_string(),
595        })?;
596    let workspace = journal.prepare_workspace(identity.operation_id()).await?;
597    let operation = SupervisedBuildOperation::new(identity, plan_digest.clone(), cache_policy)?;
598    locked.write_supervised(operation).await?;
599    drop(locked);
600    spawn_supervised_build(SupervisedBuildTask {
601        identity: identity.clone(),
602        plan: plan.clone(),
603        plan_digest,
604        source_root: source_root.to_path_buf(),
605        quiet,
606        store,
607        journal,
608        workspace,
609        lease,
610    });
611    Ok(StartOutcome {
612        status: RecordedBuildStatus::Running,
613        started_here: true,
614    })
615}
616
617struct SupervisedBuildTask {
618    identity: BuildOperationIdentity,
619    plan: BoxBuildPlan,
620    plan_digest: String,
621    source_root: PathBuf,
622    quiet: bool,
623    store: Arc<ImageStore>,
624    journal: BuildOperationJournal,
625    workspace: PathBuf,
626    lease: BuildExecutionLease,
627}
628
629fn spawn_supervised_build(task: SupervisedBuildTask) {
630    tokio::spawn(async move {
631        let operation_id = task.identity.operation_id().to_string();
632        if let Err(error) = run_supervised_build(task).await {
633            tracing::error!(
634                operation_id,
635                %error,
636                "Supervised native build ended before committing a terminal journal state"
637            );
638        }
639    });
640}
641
642async fn run_supervised_build(task: SupervisedBuildTask) -> Result<(), BuildPlanExecutionError> {
643    let SupervisedBuildTask {
644        identity,
645        plan,
646        plan_digest,
647        source_root,
648        quiet,
649        store,
650        journal,
651        workspace,
652        lease: _lease,
653    } = task;
654    let observer = Arc::new(JournalBuildObserver {
655        journal: journal.clone(),
656        identity: identity.clone(),
657        plan_digest: plan_digest.clone(),
658        cache_policy: plan.cache(),
659    });
660    let control = BuildExecutionControl::new(observer);
661    let result = execute_supervised_build_plan(
662        &identity,
663        &plan,
664        &source_root,
665        BoxBuildOptions {
666            tag: Some(identity.output_reference().to_string()),
667            quiet,
668        },
669        Arc::clone(&store),
670        &workspace,
671        control,
672    )
673    .await;
674
675    if let Ok(planned) = result {
676        journal.cleanup_workspace(identity.operation_id()).await?;
677        let receipt = BuildOutputReceipt::from_result(
678            &identity,
679            planned.plan_digest,
680            &planned.output,
681            plan.cache(),
682            planned.cache.as_ref(),
683        )?;
684        let locked = journal.lock(identity.operation_id()).await?;
685        locked.write_succeeded(receipt).await?;
686        return Ok(());
687    }
688
689    let error = result.unwrap_err();
690    finish_failed_execution(
691        &identity,
692        &plan_digest,
693        plan.cache(),
694        &store,
695        &journal,
696        error.to_string(),
697    )
698    .await
699}
700
701async fn finish_failed_execution(
702    identity: &BuildOperationIdentity,
703    plan_digest: &str,
704    cache_policy: BuildCachePolicy,
705    store: &ImageStore,
706    journal: &BuildOperationJournal,
707    message: String,
708) -> Result<(), BuildPlanExecutionError> {
709    let locked = journal.lock(identity.operation_id()).await?;
710    let Some(record) = locked.read().await? else {
711        return Err(BuildReceiptError::Conflict {
712            operation_id: identity.operation_id().to_string(),
713            message: "operation record disappeared before failure reconciliation".to_string(),
714        }
715        .into());
716    };
717    if matches!(record, PersistedBuildOperation::Succeeded(_)) {
718        return Ok(());
719    }
720    let persisted_cache_policy = match &record {
721        PersistedBuildOperation::Supervised(operation) => operation.cache_policy(),
722        PersistedBuildOperation::Pending(_) => None,
723        PersistedBuildOperation::Succeeded(_) => unreachable!(),
724    };
725    if let Some(result) = adopt_committed_output(
726        identity,
727        plan_digest,
728        persisted_cache_policy,
729        store,
730        journal,
731        &locked,
732    )
733    .await?
734    {
735        drop(result);
736        return Ok(());
737    }
738    let mut operation = match record {
739        PersistedBuildOperation::Supervised(operation) => operation,
740        PersistedBuildOperation::Pending(pending) => {
741            SupervisedBuildOperation::from_pending(&pending, identity, plan_digest, cache_policy)?
742        }
743        PersistedBuildOperation::Succeeded(_) => unreachable!(),
744    };
745    operation.require_identity(identity, plan_digest, cache_policy)?;
746    fence_run_process(operation.run_process, identity.operation_id()).await?;
747    journal.cleanup_workspace(identity.operation_id()).await?;
748    journal
749        .cleanup_cache_export(identity.operation_id())
750        .await?;
751    let phase = if operation.phase == PersistedBuildPhase::Cancelling {
752        PersistedBuildPhase::Cancelled
753    } else {
754        PersistedBuildPhase::Failed
755    };
756    operation.finish(phase, message);
757    locked.write_supervised(operation).await?;
758    Ok(())
759}
760
761async fn recover_stale_operation(
762    identity: &BuildOperationIdentity,
763    plan_digest: &str,
764    mut operation: SupervisedBuildOperation,
765    store: &ImageStore,
766    journal: &BuildOperationJournal,
767    locked: &LockedBuildOperation,
768    _lease: BuildExecutionLease,
769) -> Result<RecordedBuildStatus, BuildPlanExecutionError> {
770    if let Some(result) = adopt_committed_output(
771        identity,
772        plan_digest,
773        operation.cache_policy(),
774        store,
775        journal,
776        locked,
777    )
778    .await?
779    {
780        return Ok(RecordedBuildStatus::Succeeded(Box::new(result)));
781    }
782    fence_run_process(operation.run_process, identity.operation_id()).await?;
783    journal.cleanup_workspace(identity.operation_id()).await?;
784    journal
785        .cleanup_cache_export(identity.operation_id())
786        .await?;
787    let (phase, message) = if operation.phase == PersistedBuildPhase::Cancelling {
788        (
789            PersistedBuildPhase::Cancelled,
790            "cancelled after the native execution owner exited".to_string(),
791        )
792    } else {
793        (
794            PersistedBuildPhase::Failed,
795            "native execution owner exited before committing an output".to_string(),
796        )
797    };
798    operation.finish(phase, message);
799    let operation = locked.write_supervised(operation).await?;
800    status_from_terminal_operation(&operation)
801}
802
803async fn adopt_committed_output(
804    identity: &BuildOperationIdentity,
805    plan_digest: &str,
806    cache_policy: Option<BuildCachePolicy>,
807    store: &ImageStore,
808    journal: &BuildOperationJournal,
809    locked: &LockedBuildOperation,
810) -> Result<Option<RecordedBuildResult>, BuildPlanExecutionError> {
811    let Some(output) =
812        inspect_stored_output(identity.operation_id(), identity.output_reference(), store).await?
813    else {
814        return Ok(None);
815    };
816    journal.cleanup_workspace(identity.operation_id()).await?;
817    let cache =
818        inspect_committed_cache(identity, plan_digest, cache_policy, &output, journal).await?;
819    let receipt = match cache_policy {
820        Some(cache_policy) => BuildOutputReceipt::from_result(
821            identity,
822            plan_digest.to_string(),
823            &output,
824            cache_policy,
825            cache.as_ref(),
826        )?,
827        None => BuildOutputReceipt::from_legacy_result(identity, plan_digest.to_string(), &output)?,
828    };
829    let receipt = locked.write_succeeded(receipt).await?;
830    Ok(Some(RecordedBuildResult {
831        receipt,
832        output,
833        cache,
834        replayed: true,
835    }))
836}
837
838async fn inspect_committed_cache(
839    identity: &BuildOperationIdentity,
840    plan_digest: &str,
841    cache_policy: Option<BuildCachePolicy>,
842    output: &BuildResult,
843    journal: &BuildOperationJournal,
844) -> Result<Option<RecordedBuildCache>, BuildPlanExecutionError> {
845    let path = journal.cache_export_path(identity.operation_id());
846    let cache_exists = match tokio::fs::symlink_metadata(&path).await {
847        Ok(_) => true,
848        Err(error) if error.kind() == std::io::ErrorKind::NotFound => false,
849        Err(error) => {
850            return Err(BuildReceiptError::StoreIo {
851                message: format!(
852                    "failed to inspect cache export for operation {}",
853                    identity.operation_id()
854                ),
855                source: error,
856            }
857            .into())
858        }
859    };
860    match (cache_policy, cache_exists) {
861        (None, true) => {
862            journal
863                .cleanup_cache_export(identity.operation_id())
864                .await?;
865            return Ok(None);
866        }
867        (None | Some(BuildCachePolicy::Disabled), false) => return Ok(None),
868        (Some(BuildCachePolicy::Disabled), true) => {
869            return Err(BuildReceiptError::CacheInvalid {
870                operation_id: identity.operation_id().to_string(),
871                message: "cache export exists for a cache-disabled operation".to_string(),
872            }
873            .into())
874        }
875        (Some(BuildCachePolicy::ContentAddressed), false) => {
876            return Err(BuildReceiptError::CacheInvalid {
877                operation_id: identity.operation_id().to_string(),
878                message: "content-addressed operation committed an image without its cache export"
879                    .to_string(),
880            }
881            .into())
882        }
883        (Some(BuildCachePolicy::ContentAddressed), true) => {}
884    }
885    let cache_identity = BuildCacheExportIdentity::new(
886        identity.source_digest(),
887        plan_digest,
888        output.platform.clone(),
889    )?;
890    let operation = identity.operation_id().to_string();
891    tokio::task::spawn_blocking(move || inspect_build_cache_artifact(&path, &cache_identity, None))
892        .await
893        .map_err(|error| BuildReceiptError::Task {
894            operation_id: operation.clone(),
895            message: format!("cache export validation task failed: {error}"),
896        })?
897        .map(Some)
898        .map_err(|error| {
899            BuildReceiptError::CacheInvalid {
900                operation_id: operation,
901                message: error.to_string(),
902            }
903            .into()
904        })
905}
906
907fn status_from_live_operation(operation: &SupervisedBuildOperation) -> RecordedBuildStatus {
908    if operation.phase == PersistedBuildPhase::Cancelling {
909        RecordedBuildStatus::Cancelling
910    } else {
911        RecordedBuildStatus::Running
912    }
913}
914
915fn status_from_terminal_operation(
916    operation: &SupervisedBuildOperation,
917) -> Result<RecordedBuildStatus, BuildPlanExecutionError> {
918    let message = operation
919        .terminal_message()
920        .ok_or_else(|| BuildReceiptError::InvalidReceipt {
921            operation_id: operation.operation_id().to_string(),
922            message: "terminal operation has no message".to_string(),
923        })?
924        .to_string();
925    match operation.phase {
926        PersistedBuildPhase::Cancelled => Ok(RecordedBuildStatus::Cancelled { message }),
927        PersistedBuildPhase::Failed => Ok(RecordedBuildStatus::Failed { message }),
928        PersistedBuildPhase::Running | PersistedBuildPhase::Cancelling => {
929            Err(BuildReceiptError::InvalidReceipt {
930                operation_id: operation.operation_id().to_string(),
931                message: "live operation was decoded through the terminal boundary".to_string(),
932            }
933            .into())
934        }
935    }
936}
937
938async fn recover_recorded_build(
939    identity: &BuildOperationIdentity,
940    plan_digest: &str,
941    cache_policy: BuildCachePolicy,
942    receipt: BuildOutputReceipt,
943    store: &ImageStore,
944    journal: &BuildOperationJournal,
945) -> Result<RecordedBuildResult, BuildPlanExecutionError> {
946    receipt.require_identity(identity, plan_digest, cache_policy)?;
947    let output = receipt.resolve(store).await?;
948    let cache = receipt
949        .resolve_cache(&journal.cache_export_path(identity.operation_id()))
950        .await?;
951    Ok(RecordedBuildResult {
952        receipt,
953        output,
954        cache,
955        replayed: true,
956    })
957}
958
959#[cfg(test)]
960mod tests;
a3s_box_runtime/oci/build/execution.rs

a3s_box_runtime/oci/build/
execution.rs
