4 files changed, 218 insertions, 0 deletions
diff --git a/tvix/store/src/pathinfoservice/grpc.rs b/tvix/store/src/pathinfoservice/grpc.rs
new file mode 100644
index 000000000000..6bb774c668a3
--- /dev/null
+++ b/tvix/store/src/pathinfoservice/grpc.rs
@@ -0,0 +1,81 @@
+use super::PathInfoService;
+use crate::proto;
+use tonic::{transport::Channel, Code, Status};
+
+/// Connects to a (remote) tvix-store PathInfoService over gRPC.
+#[derive(Clone)]
+pub struct GRPCPathInfoService {
+    /// A handle into the active tokio runtime. Necessary to spawn tasks.
+    tokio_handle: tokio::runtime::Handle,
+
+    /// The internal reference to a gRPC client.
+    /// Cloning it is cheap, and it internally handles concurrent requests.
+    grpc_client: proto::path_info_service_client::PathInfoServiceClient<Channel>,
+}
+
+impl GRPCPathInfoService {
+    /// Construct a new [GRPCPathInfoService], by passing a handle to the tokio
+    /// runtime, and a gRPC client.
+    pub fn new(
+        tokio_handle: tokio::runtime::Handle,
+        grpc_client: proto::path_info_service_client::PathInfoServiceClient<Channel>,
+    ) -> Self {
+        Self {
+            tokio_handle,
+            grpc_client,
+        }
+    }
+
+    /// construct a [GRPCDirectoryService] from a [proto::path_info_service_client::PathInfoServiceClient<Channel>].
+    /// panics if called outside the context of a tokio runtime.
+    pub fn from_client(
+        grpc_client: proto::path_info_service_client::PathInfoServiceClient<Channel>,
+    ) -> Self {
+        Self {
+            tokio_handle: tokio::runtime::Handle::current(),
+            grpc_client,
+        }
+    }
+}
+
+impl PathInfoService for GRPCPathInfoService {
+    fn get(&self, digest: [u8; 20]) -> Result<Option<proto::PathInfo>, crate::Error> {
+        // Get a new handle to the gRPC client.
+        let mut grpc_client = self.grpc_client.clone();
+
+        let task: tokio::task::JoinHandle<Result<proto::PathInfo, Status>> =
+            self.tokio_handle.spawn(async move {
+                let path_info = grpc_client
+                    .get(proto::GetPathInfoRequest {
+                        by_what: Some(proto::get_path_info_request::ByWhat::ByOutputHash(
+                            digest.to_vec(),
+                        )),
+                    })
+                    .await?
+                    .into_inner();
+
+                Ok(path_info)
+            });
+
+        match self.tokio_handle.block_on(task)? {
+            Ok(path_info) => Ok(Some(path_info)),
+            Err(e) if e.code() == Code::NotFound => Ok(None),
+            Err(e) => Err(crate::Error::StorageError(e.to_string())),
+        }
+    }
+
+    fn put(&self, path_info: proto::PathInfo) -> Result<proto::PathInfo, crate::Error> {
+        // Get a new handle to the gRPC client.
+        let mut grpc_client = self.grpc_client.clone();
+
+        let task: tokio::task::JoinHandle<Result<proto::PathInfo, Status>> =
+            self.tokio_handle.spawn(async move {
+                let path_info = grpc_client.put(path_info).await?.into_inner();
+                Ok(path_info)
+            });
+
+        self.tokio_handle
+            .block_on(task)?
+            .map_err(|e| crate::Error::StorageError(e.to_string()))
+    }
+}
diff --git a/tvix/store/src/pathinfoservice/memory.rs b/tvix/store/src/pathinfoservice/memory.rs
new file mode 100644
index 000000000000..d0ff1976efab
--- /dev/null
+++ b/tvix/store/src/pathinfoservice/memory.rs
@@ -0,0 +1,41 @@
+use super::PathInfoService;
+use crate::{proto, Error};
+use std::{
+    collections::HashMap,
+    sync::{Arc, RwLock},
+};
+
+#[derive(Default)]
+pub struct MemoryPathInfoService {
+    db: Arc<RwLock<HashMap<[u8; 20], proto::PathInfo>>>,
+}
+
+impl PathInfoService for MemoryPathInfoService {
+    fn get(&self, digest: [u8; 20]) -> Result<Option<proto::PathInfo>, Error> {
+        let db = self.db.read().unwrap();
+
+        match db.get(&digest) {
+            None => Ok(None),
+            Some(path_info) => Ok(Some(path_info.clone())),
+        }
+    }
+
+    fn put(&self, path_info: proto::PathInfo) -> Result<proto::PathInfo, Error> {
+        // Call validate on the received PathInfo message.
+        match path_info.validate() {
+            Err(e) => Err(Error::InvalidRequest(format!(
+                "failed to validate PathInfo: {}",
+                e
+            ))),
+
+            // In case the PathInfo is valid, and we were able to extract a NixPath, store it in the database.
+            // This overwrites existing PathInfo objects.
+            Ok(nix_path) => {
+                let mut db = self.db.write().unwrap();
+                db.insert(nix_path.digest, path_info.clone());
+
+                Ok(path_info)
+            }
+        }
+    }
+}
diff --git a/tvix/store/src/pathinfoservice/mod.rs b/tvix/store/src/pathinfoservice/mod.rs
new file mode 100644
index 000000000000..6a34e09af478
--- /dev/null
+++ b/tvix/store/src/pathinfoservice/mod.rs
@@ -0,0 +1,20 @@
+mod grpc;
+mod memory;
+mod sled;
+
+use crate::{proto, Error};
+
+pub use self::grpc::GRPCPathInfoService;
+pub use self::memory::MemoryPathInfoService;
+pub use self::sled::SledPathInfoService;
+
+/// The base trait all PathInfo services need to implement.
+/// This is a simple get and put of [proto::Directory], returning their digest.
+pub trait PathInfoService {
+    /// Retrieve a PathInfo message by the output digest.
+    fn get(&self, digest: [u8; 20]) -> Result<Option<proto::PathInfo>, Error>;
+
+    /// Store a PathInfo message. Implementations MUST call validate and reject
+    /// invalid messages.
+    fn put(&self, path_info: proto::PathInfo) -> Result<proto::PathInfo, Error>;
+}
diff --git a/tvix/store/src/pathinfoservice/sled.rs b/tvix/store/src/pathinfoservice/sled.rs
new file mode 100644
index 000000000000..8776ebcbc106
--- /dev/null
+++ b/tvix/store/src/pathinfoservice/sled.rs
@@ -0,0 +1,76 @@
+use super::PathInfoService;
+use crate::{proto, Error};
+use prost::Message;
+use std::path::PathBuf;
+use tracing::warn;
+
+/// SledPathInfoService stores PathInfo in a [sled](https://github.com/spacejam/sled).
+///
+/// The PathInfo messages are stored as encoded protos, and keyed by their output hash,
+/// as that's currently the only request type available.
+#[derive(Clone)]
+pub struct SledPathInfoService {
+    db: sled::Db,
+}
+
+impl SledPathInfoService {
+    pub fn new(p: PathBuf) -> Result<Self, sled::Error> {
+        let config = sled::Config::default().use_compression(true).path(p);
+        let db = config.open()?;
+
+        Ok(Self { db })
+    }
+
+    pub fn new_temporary() -> Result<Self, sled::Error> {
+        let config = sled::Config::default().temporary(true);
+        let db = config.open()?;
+
+        Ok(Self { db })
+    }
+}
+
+impl PathInfoService for SledPathInfoService {
+    fn get(&self, digest: [u8; 20]) -> Result<Option<proto::PathInfo>, Error> {
+        match self.db.get(digest) {
+            Ok(None) => Ok(None),
+            Ok(Some(data)) => match proto::PathInfo::decode(&*data) {
+                Ok(path_info) => Ok(Some(path_info)),
+                Err(e) => {
+                    warn!("failed to decode stored PathInfo: {}", e);
+                    Err(Error::StorageError(format!(
+                        "failed to decode stored PathInfo: {}",
+                        e
+                    )))
+                }
+            },
+            Err(e) => {
+                warn!("failed to retrieve PathInfo: {}", e);
+                Err(Error::StorageError(format!(
+                    "failed to retrieve PathInfo: {}",
+                    e
+                )))
+            }
+        }
+    }
+
+    fn put(&self, path_info: proto::PathInfo) -> Result<proto::PathInfo, Error> {
+        // Call validate on the received PathInfo message.
+        match path_info.validate() {
+            Err(e) => Err(Error::InvalidRequest(format!(
+                "failed to validate PathInfo: {}",
+                e
+            ))),
+            // In case the PathInfo is valid, and we were able to extract a NixPath, store it in the database.
+            // This overwrites existing PathInfo objects.
+            Ok(nix_path) => match self.db.insert(nix_path.digest, path_info.encode_to_vec()) {
+                Ok(_) => Ok(path_info),
+                Err(e) => {
+                    warn!("failed to insert PathInfo: {}", e);
+                    Err(Error::StorageError(format! {
+                        "failed to insert PathInfo: {}", e
+                    }))
+                }
+            },
+        }
+    }
+}