feat(tvix/eval): implement `builtins.filterSource` r/7578

We add a new set of builtins called `import_builtins`, which
will contain import-related builtins, such as `builtins.path` and
`builtins.filterSource`. Both can import paths into the store, with
various knobs to alter the result, e.g. filtering, renaming, expected
hashes.

We introduce `filtered_ingest` which will drive the filtered ingestion
via the Nix function via the generator machinery, and then we register
the root node to the path info service inside the store.

`builtins.filterSource` is very simple, `builtins.path` is a more
complicated model requiring the same logic albeit more sophisticated
with name customization, file ingestion method and expected SHA-256.

Change-Id: I1083f37808b35f7b37818c8ffb9543d9682b2de2
Reviewed-on: https://cl.tvl.fyi/c/depot/+/10654
Autosubmit: raitobezarius <tvl@lahfa.xyz>
Tested-by: BuildkiteCI
Reviewed-by: flokli <flokli@flokli.de>

1 files changed, 77 insertions, 2 deletions

diff --git a/tvix/glue/src/tvix_store_io.rs b/tvix/glue/src/tvix_store_io.rs
index 025a8a4bf01c..375501b65a5a 100644
--- a/tvix/glue/src/tvix_store_io.rs
+++ b/tvix/glue/src/tvix_store_io.rs
@@ -2,6 +2,7 @@
 
 use async_recursion::async_recursion;
 use bytes::Bytes;
+use futures::Stream;
 use futures::{StreamExt, TryStreamExt};
 use nix_compat::{
     nixhash::CAHash,
@@ -18,6 +19,7 @@ use tokio::io::AsyncReadExt;
 use tracing::{error, instrument, warn, Level};
 use tvix_build::buildservice::BuildService;
 use tvix_eval::{EvalIO, FileType, StdIO};
+use walkdir::DirEntry;
 
 use tvix_castore::{
     blobservice::BlobService,
@@ -282,6 +284,79 @@ impl TvixStoreIO {
         self.tokio_handle
             .block_on(async { self.store_path_to_node(store_path, sub_path).await })
     }
+
+    /// This forwards the ingestion to the [`tvix_castore::import::ingest_entries`]
+    /// with a [`tokio::runtime::Handle::block_on`] call for synchronicity.
+    pub(crate) fn ingest_entries_sync<S>(&self, entries_stream: S) -> io::Result<Node>
+    where
+        S: Stream<Item = DirEntry> + std::marker::Unpin,
+    {
+        self.tokio_handle.block_on(async move {
+            tvix_castore::import::ingest_entries(
+                &self.blob_service,
+                &self.directory_service,
+                entries_stream,
+            )
+            .await
+            .map_err(|err| std::io::Error::new(io::ErrorKind::Other, err))
+        })
+    }
+
+    pub(crate) async fn node_to_path_info(
+        &self,
+        name: &str,
+        path: &Path,
+        root_node: Node,
+    ) -> io::Result<(PathInfo, StorePath)> {
+        // Ask the PathInfoService for the NAR size and sha256
+        let (nar_size, nar_sha256) = self
+            .path_info_service
+            .as_ref()
+            .calculate_nar(&root_node)
+            .await?;
+
+        // Calculate the output path. This might still fail, as some names are illegal.
+        let output_path = nix_compat::store_path::build_nar_based_store_path(&nar_sha256, name)
+            .map_err(|_| {
+                std::io::Error::new(
+                    std::io::ErrorKind::InvalidData,
+                    format!("invalid name: {}", name),
+                )
+            })?;
+
+        // assemble a new root_node with a name that is derived from the nar hash.
+        let root_node = root_node.rename(output_path.to_string().into_bytes().into());
+        tvix_store::import::log_node(&root_node, path);
+
+        let path_info =
+            tvix_store::import::derive_nar_ca_path_info(nar_size, nar_sha256, root_node);
+
+        Ok((path_info, output_path.to_owned()))
+    }
+
+    pub(crate) async fn register_node_in_path_info_service(
+        &self,
+        name: &str,
+        path: &Path,
+        root_node: Node,
+    ) -> io::Result<StorePath> {
+        let (path_info, output_path) = self.node_to_path_info(name, path, root_node).await?;
+        let _path_info = self.path_info_service.as_ref().put(path_info).await?;
+
+        Ok(output_path)
+    }
+
+    pub(crate) fn register_node_in_path_info_service_sync(
+        &self,
+        name: &str,
+        path: &Path,
+        root_node: Node,
+    ) -> io::Result<StorePath> {
+        self.tokio_handle.block_on(async {
+            self.register_node_in_path_info_service(name, path, root_node)
+                .await
+        })
+    }
 }
 
 impl EvalIO for TvixStoreIO {
@@ -475,9 +550,8 @@ mod tests {
     use tvix_eval::{EvalIO, EvaluationResult};
     use tvix_store::pathinfoservice::MemoryPathInfoService;
 
-    use crate::builtins::{add_derivation_builtins, add_fetcher_builtins};
-
     use super::TvixStoreIO;
+    use crate::builtins::{add_derivation_builtins, add_fetcher_builtins, add_import_builtins};
 
     /// evaluates a given nix expression and returns the result.
     /// Takes care of setting up the evaluator so it knows about the
@@ -504,6 +578,7 @@ mod tests {
 
         add_derivation_builtins(&mut eval, io.clone());
         add_fetcher_builtins(&mut eval, io.clone());
+        add_import_builtins(&mut eval, io);
 
         // run the evaluation itself.
         eval.evaluate(str, None)