feat(build): refactor BuildService to streaming RPC

snix/build/Cargo.toml

@@ -21,6 +21,7 @@ mimalloc.workspace = true
 tonic-reflection = { workspace = true, optional = true }
 
 anyhow = "1.0.79"
+async-stream = "0.3"
 blake3 = "1.5.0"
 bstr = "1.6.0"
 data-encoding = "2.5.0"

snix/build/protos/build.proto

@@ -155,8 +155,47 @@ message BuildRequest {
   // TODO: allow describing something like "preferLocal", to influence composition?
 }
 
-// A BuildResponse is (one possible) outcome of executing a [BuildRequest].
-message BuildResponse {
+// BuildEvent represents a single event during a build process.
+// The stream of events will always end with either BuildCompleted or BuildFailed.
+message BuildEvent {
+  oneof event {
+    BuildStarted started = 1;
+    LogOutput log = 2;
+    RefscanResult refscan = 3;
+    BuildCompleted completed = 4;
+    BuildFailed failed = 5;
+  }
+}
+
+// Emitted at the start of a build.
+message BuildStarted {
+  // A unique identifier for this build.
+  string build_id = 1;
+}
+
+// A line of log output from the build process.
+message LogOutput {
+  enum Stream {
+    STREAM_UNSPECIFIED = 0;
+    STREAM_STDOUT = 1;
+    STREAM_STDERR = 2;
+  }
+  // Which output stream this line came from.
+  Stream stream = 1;
+  // The log line data (including newline).
+  bytes data = 2;
+}
+
+// Reference scanning result for a single output.
+message RefscanResult {
+  // Index of the output in the original BuildRequest.outputs.
+  uint32 output_index = 1;
+  // Indexes into [BuildRequest::refscan_needles] found in this output.
+  repeated uint64 found_needles = 2;
+}
+
+// Emitted when a build completes successfully.
+message BuildCompleted {
   // The outputs that were produced after successfully building.
   // They are provided in the same order as specified in the [BuildRequest].
   repeated Output outputs = 1;
@@ -170,8 +209,14 @@ message BuildResponse {
     // Indexes into the found [BuildRequest::refscan_needles] in this output.
     repeated uint64 needles = 2;
   }
+}
 
-  // TODO: where did this run, how long, logs, …
+// Emitted when a build fails.
+message BuildFailed {
+  // Human-readable error message.
+  string message = 1;
+  // Exit code of the build process, if available.
+  optional int32 exit_code = 2;
 }
 
 /// TODO: check remarkable notes on constraints again

snix/build/protos/rpc_build.proto

@@ -10,5 +10,7 @@ import "snix/build/protos/build.proto";
 option go_package = "snix.dev/build/proto;buildv1";
 
 service BuildService {
-  rpc DoBuild(BuildRequest) returns (BuildResponse);
+  // Execute a build and stream events back.
+  // Dropping the stream signals cancellation.
+  rpc DoBuild(BuildRequest) returns (stream BuildEvent);
 }

snix/build/src/buildservice/build_request.rs

@@ -146,3 +146,59 @@ pub struct BuildOutput {
     /// Indexes into the found [BuildRequest::refscan_needles] in that output.
     pub output_needles: BTreeSet<u64>,
 }
+
+/// An event emitted during a build process.
+#[derive(Debug, Clone)]
+pub enum BuildEvent {
+    /// Emitted at the start of a build.
+    Started(BuildStarted),
+    /// A line of log output from the build process.
+    Log(LogOutput),
+    /// Reference scanning result for an output.
+    RefscanResult(RefscanResultEvent),
+    /// The build completed successfully.
+    Completed(BuildResult),
+    /// The build failed.
+    Failed(BuildError),
+}
+
+/// Emitted at the start of a build.
+#[derive(Debug, Clone)]
+pub struct BuildStarted {
+    /// A unique identifier for this build.
+    pub build_id: String,
+}
+
+/// Which output stream a log line came from.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LogStream {
+    Stdout,
+    Stderr,
+}
+
+/// A line of log output from the build process.
+#[derive(Debug, Clone)]
+pub struct LogOutput {
+    /// Which output stream this line came from.
+    pub stream: LogStream,
+    /// The log line data (including newline).
+    pub data: Bytes,
+}
+
+/// Reference scanning result for a single output.
+#[derive(Debug, Clone)]
+pub struct RefscanResultEvent {
+    /// Index of the output in the original BuildRequest.outputs.
+    pub output_index: usize,
+    /// Indexes into [BuildRequest::refscan_needles] found in this output.
+    pub found_needles: Vec<u64>,
+}
+
+/// Describes a build failure.
+#[derive(Debug, Clone)]
+pub struct BuildError {
+    /// Human-readable error message.
+    pub message: String,
+    /// Exit code of the build process, if available.
+    pub exit_code: Option<i32>,
+}

snix/build/src/buildservice/dummy.rs

@@ -1,18 +1,66 @@
-use tonic::async_trait;
+use futures::stream;
 use tracing::instrument;
 
-use super::BuildService;
-use crate::buildservice::{BuildRequest, BuildResult};
+use super::{BuildEventStream, BuildService};
+use crate::buildservice::BuildRequest;
 
 #[derive(Default)]
 pub struct DummyBuildService {}
 
-#[async_trait]
 impl BuildService for DummyBuildService {
-    #[instrument(skip(self), ret, err)]
-    async fn do_build(&self, _request: BuildRequest) -> std::io::Result<BuildResult> {
-        Err(std::io::Error::other(
-            "builds are not supported with DummyBuildService",
-        ))
+    #[instrument(skip(self))]
+    fn do_build(&self, _request: BuildRequest) -> BuildEventStream {
+        Box::pin(stream::once(async {
+            Err(std::io::Error::other(
+                "builds are not supported with DummyBuildService",
+            ))
+        }))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::StreamExt;
+    use std::collections::BTreeMap;
+    use std::path::PathBuf;
+
+    fn make_dummy_request() -> BuildRequest {
+        BuildRequest {
+            inputs: BTreeMap::new(),
+            command_args: vec!["echo".to_string(), "hello".to_string()],
+            working_dir: PathBuf::from("build"),
+            scratch_paths: vec![PathBuf::from("build")],
+            inputs_dir: PathBuf::from("nix/store"),
+            outputs: vec![PathBuf::from("build/out")],
+            environment_vars: vec![],
+            constraints: Default::default(),
+            additional_files: vec![],
+            refscan_needles: vec![],
+        }
+    }
+
+    #[tokio::test]
+    async fn test_dummy_build_service_returns_error_stream() {
+        let service = DummyBuildService::default();
+        let mut stream = service.do_build(make_dummy_request());
+
+        // First item should be an error
+        let first = stream.next().await;
+        assert!(first.is_some(), "stream should yield at least one item");
+
+        let result = first.unwrap();
+        assert!(result.is_err(), "should be an error");
+
+        let err = result.unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("builds are not supported with DummyBuildService"),
+            "error message should explain builds are not supported"
+        );
+
+        // Stream should end after the error
+        let second = stream.next().await;
+        assert!(second.is_none(), "stream should end after the error");
     }
 }

snix/build/src/buildservice/grpc.rs

@@ -1,9 +1,9 @@
-use tonic::{async_trait, transport::Channel};
+use tonic::transport::Channel;
 
-use crate::buildservice::BuildRequest;
+use crate::buildservice::{BuildEvent, BuildRequest};
 use crate::proto::{self, build_service_client::BuildServiceClient};
 
-use super::{BuildResult, BuildService};
+use super::{BuildEventStream, BuildService};
 
 pub struct GRPCBuildService {
     client: BuildServiceClient<Channel>,
@@ -16,16 +16,25 @@ impl GRPCBuildService {
     }
 }
 
-#[async_trait]
 impl BuildService for GRPCBuildService {
-    async fn do_build(&self, request: BuildRequest) -> std::io::Result<BuildResult> {
+    fn do_build(&self, request: BuildRequest) -> BuildEventStream {
         let mut client = self.client.clone();
-        let resp = client
-            .do_build(Into::<proto::BuildRequest>::into(request))
-            .await
-            .map_err(std::io::Error::other)?
-            .into_inner();
+        let proto_request: proto::BuildRequest = request.into();
 
-        Ok::<BuildResult, _>(resp.try_into().map_err(std::io::Error::other)?)
+        let stream = async_stream::try_stream! {
+            let response = client
+                .do_build(proto_request)
+                .await
+                .map_err(std::io::Error::other)?;
+
+            let mut stream = response.into_inner();
+
+            while let Some(event) = stream.message().await.map_err(std::io::Error::other)? {
+                let event: BuildEvent = event.try_into().map_err(std::io::Error::other)?;
+                yield event;
+            }
+        };
+
+        Box::pin(stream)
     }
 }

snix/build/src/buildservice/mod.rs

@@ -1,4 +1,4 @@
-use tonic::async_trait;
+use futures::stream::BoxStream;
 
 pub mod build_request;
 pub use crate::buildservice::build_request::*;
@@ -12,8 +12,19 @@ mod oci;
 pub use dummy::DummyBuildService;
 pub use from_addr::from_addr;
 
-#[async_trait]
+/// A stream of build events.
+pub type BuildEventStream = BoxStream<'static, Result<BuildEvent, std::io::Error>>;
+
+/// Service for executing builds.
 pub trait BuildService: Send + Sync {
-    /// TODO: document
-    async fn do_build(&self, request: BuildRequest) -> std::io::Result<BuildResult>;
+    /// Execute a build and return a stream of events.
+    ///
+    /// The stream will emit events as the build progresses:
+    /// - `BuildStarted` at the beginning
+    /// - `Log` events for stdout/stderr output (line by line)
+    /// - `RefscanResult` events for each output after ingestion
+    /// - Either `Completed` or `Failed` at the end
+    ///
+    /// Dropping the stream signals cancellation - the build will be aborted.
+    fn do_build(&self, request: BuildRequest) -> BuildEventStream;
 }