mlua_swarm/worker/
output.rs

1//! Output path.
2//!
3//! The worker → `SpawnerAdapter` → engine output path. All structuring is
4//! completed inside the `SpawnerAdapter`; by the time an event reaches the
5//! engine it is a Rust-typed `OutputEvent`. The wire form the worker uses
6//! (stdout / NDJSON / file path / IPC) is opaque to the engine.
7//!
8//! The `WorkerResult` type was folded into `OutputEvent::Final`.
9//!
10//! # Canonical type locations
11//!
12//! `OutputEvent` and `ContentRef` are canonical in [`crate::store::output`];
13//! this module is narrowed to re-exports plus the engine-specific
14//! `OutputSink` / `EngineSink`.
15
16use crate::core::errors::EngineError;
17use async_trait::async_trait;
18
19pub use crate::store::output::{ContentRef, OutputEvent};
20
21/// Sink used inside a worker function to emit events. The `InProcSpawner`
22/// injects one into `WorkerInvocation`. The `ProcessSpawner` / child-process
23/// pull path folds stdout / IPC into `OutputEvent` internally and calls
24/// `engine.submit_output` directly, so it does not go through `OutputSink`
25/// (it lands in the same engine state, but not via this trait).
26#[async_trait]
27pub trait OutputSink: Send + Sync {
28    /// Emits one `OutputEvent` (progress, final, etc.) into the engine's
29    /// output stream for this attempt.
30    async fn emit(&self, event: OutputEvent) -> Result<(), EngineError>;
31}
32
33/// Concrete `OutputSink` — the default implementation that closes over
34/// `engine`, `token`, `task_id`, and `attempt`, and calls
35/// `engine.submit_output` for every `emit`. Injected by the `InProcSpawner`
36/// into `WorkerInvocation`.
37///
38/// `InProcSpawner::spawn` is this type's **sole constructor**, which is
39/// what makes it the in-process lane's "the worker itself did this"
40/// boundary: an `OutputEvent::Artifact` arriving here is a part the worker
41/// staged, as opposed to one some other producer appended to the same
42/// shared tail (`AfterRunAuditMiddleware` calls `submit_output` directly
43/// for exactly that reason). `emit` therefore records `Artifact` names
44/// into `EngineState.worker_artifact_names` — the allowlist the Final-pull
45/// folds `{out, parts}` from. Adding a second constructor for a non-worker
46/// producer would break that invariant; such a producer should call
47/// `Engine::submit_output` directly instead.
48#[derive(Clone)]
49pub struct EngineSink {
50    engine: crate::core::engine::Engine,
51    token: crate::types::CapToken,
52    task_id: crate::types::StepId,
53    attempt: u32,
54}
55
56impl EngineSink {
57    /// Binds a sink to one attempt's identity so every `emit` call knows
58    /// where to route the event without the caller repeating the
59    /// coordinates each time.
60    pub fn new(
61        engine: crate::core::engine::Engine,
62        token: crate::types::CapToken,
63        task_id: crate::types::StepId,
64        attempt: u32,
65    ) -> Self {
66        Self {
67            engine,
68            token,
69            task_id,
70            attempt,
71        }
72    }
73}
74
75#[async_trait]
76impl OutputSink for EngineSink {
77    async fn emit(&self, event: OutputEvent) -> Result<(), EngineError> {
78        // Read the part name off the event before handing it over.
79        let staged_part = match &event {
80            OutputEvent::Artifact { name, .. } => Some(name.clone()),
81            _ => None,
82        };
83        self.engine
84            .submit_output(&self.token, &self.task_id, self.attempt, event)
85            .await?;
86        // Only after the tail write lands: a rejected or failed submit must
87        // not leave a name in the allowlist pointing at an `Artifact` that
88        // is not on the tail. See the type doc for why recording here (and
89        // not inside `submit_output`) is what keeps a non-worker producer's
90        // artifact out of the BP-chain value.
91        if let Some(name) = staged_part {
92            self.engine
93                .record_worker_artifact_name(&self.task_id, self.attempt, name)
94                .await?;
95        }
96        Ok(())
97    }
98}
mlua_swarm/worker/output.rs

mlua_swarm/worker/
output.rs
