snix_build/buildservice/
streaming.rs

1use std::future::Future;
2
3use futures::{
4    StreamExt,
5    stream::{BoxStream, FuturesUnordered},
6};
7use snix_castore::{
8    blobservice::BlobService,
9    directoryservice::DirectoryService,
10    import::{IngestionError, fs::ingest_path},
11    refscan::{ReferencePattern, ReferenceScanner},
12};
13use tracing::{debug, warn};
14use tracing_futures::Instrument;
15
16use super::{BuildFailure, BuildUpdate};
17use crate::sandbox::{SandboxOutputs, event::SandboxEvent};
18
19/// Executes a sandboxed build and streams its progress and outputs.
20pub(crate) fn run_build_streaming<'a, BS, DS, R, SpawnFut>(
21    semaphore: &'a tokio::sync::Semaphore,
22    blob_service: BS,
23    directory_service: DS,
24    expected_outputs: Vec<String>,
25    refscan_needles: Vec<String>,
26    spawn: impl FnOnce() -> SpawnFut + Send + 'a,
27) -> BoxStream<'a, BuildUpdate>
28where
29    BS: BlobService + Clone + 'a,
30    DS: DirectoryService + Clone + 'a,
31    R: SandboxOutputs + Send + Sync + 'static,
32    SpawnFut:
33        Future<Output = Result<(BoxStream<'static, SandboxEvent>, R), BuildFailure>> + Send + 'a,
34{
35    async_stream::try_stream! {
36        let _permit = semaphore.acquire().await?;
37
38        let (mut event_stream, output_resolver) = spawn().await?;
39
40        let mut exit_code = None;
41        while let Some(event) = event_stream.next().await {
42            match event {
43                SandboxEvent::Stdout(chunk) => yield BuildUpdate::ProducedStdout(chunk),
44                SandboxEvent::Stderr(chunk) => yield BuildUpdate::ProducedStderr(chunk),
45                SandboxEvent::ExitCode(code) => exit_code = Some(code),
46            }
47        }
48
49        let exit_code = exit_code.unwrap_or(1);
50        if exit_code != 0 {
51            warn!(exit_code=%exit_code, "build failed");
52            Err(BuildFailure::NonzeroExitCode)?;
53        }
54
55        let patterns = ReferencePattern::new(refscan_needles);
56
57        let mut tasks : FuturesUnordered<_> = (expected_outputs.into_iter().enumerate().map(|(idx, o)| {
58            let blob_service = &blob_service;
59            let directory_service = &directory_service;
60            let output_resolver = &output_resolver;
61            let scanner = ReferenceScanner::new(patterns.clone());
62            async move {
63                let host_output_path = output_resolver.find_path(&o).await.ok_or(BuildFailure::MissingOutput{output: o})?;
64                ingest_host_output(idx, &host_output_path, scanner, blob_service, directory_service).await.map_err(BuildFailure::from)
65            }.in_current_span()
66        })).collect();
67
68        while let Some(elem) = tasks.next().await {
69            yield elem?;
70        }
71    }
72    .map(|res| match res {
73        Ok(update) => update,
74        Err(failure) => BuildUpdate::BuildFailure(failure),
75    })
76    .in_current_span()
77    .boxed()
78}
79
80/// Ingests the given host output path, while running reference scanning.
81/// Returns either a [BuildUpdate::ProducedOutput], or a [BuildFailure]
82#[tracing::instrument(skip_all, err, fields(host.path = ?host_output_path))]
83async fn ingest_host_output<BS, DS>(
84    idx: usize,
85    host_output_path: &std::path::Path,
86    scanner: ReferenceScanner<String>,
87    blob_service: BS,
88    directory_service: DS,
89) -> Result<BuildUpdate, IngestionError<snix_castore::import::fs::Error>>
90where
91    BS: BlobService,
92    DS: DirectoryService,
93{
94    debug!("ingesting path");
95    let node = ingest_path(
96        &blob_service,
97        &directory_service,
98        host_output_path,
99        Some(&scanner),
100    )
101    .await?;
102
103    let refscan_needles = scanner
104        .matches()
105        .into_iter()
106        .enumerate()
107        .filter(|(_, val)| *val)
108        .map(|(idx, _)| idx as u64)
109        .collect();
110
111    Ok(BuildUpdate::ProducedOutput {
112        node,
113        idx: idx as u64,
114        refscan_needles,
115    })
116}
snix_build/buildservice/streaming.rs

snix_build/buildservice/
streaming.rs
