feat(tvix/store/pathinfo): add a Cache combinator
This allows querying two PathInfoService implementations sequentially, and inserts into the "near" one if it's not there yet. There is no negative cache, and put / listing is not implemented (for now). Change-Id: I24c3d0e0c3c2f0524a6cc7b2f3cbc33eb20cf92b Reviewed-on: https://cl.tvl.fyi/c/depot/+/11636 Autosubmit: flokli <flokli@flokli.de> Tested-by: BuildkiteCI Reviewed-by: Connor Brewster <cbrewster@hey.com>
This commit is contained in:
parent
30995a0990
commit
adb42959a3
2 changed files with 113 additions and 0 deletions
111
tvix/store/src/pathinfoservice/combinators.rs
Normal file
111
tvix/store/src/pathinfoservice/combinators.rs
Normal file
|
@ -0,0 +1,111 @@
|
||||||
|
use crate::proto::PathInfo;
|
||||||
|
use futures::stream::BoxStream;
|
||||||
|
use nix_compat::nixbase32;
|
||||||
|
use tonic::async_trait;
|
||||||
|
use tracing::{debug, instrument};
|
||||||
|
use tvix_castore::Error;
|
||||||
|
|
||||||
|
use super::PathInfoService;
|
||||||
|
|
||||||
|
/// Asks near first, if not found, asks far.
|
||||||
|
/// If found in there, returns it, and *inserts* it into
|
||||||
|
/// near.
|
||||||
|
/// There is no negative cache.
|
||||||
|
/// Inserts and listings are not implemented for now.
|
||||||
|
pub struct Cache<PS1, PS2> {
|
||||||
|
near: PS1,
|
||||||
|
far: PS2,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<PS1, PS2> Cache<PS1, PS2> {
|
||||||
|
pub fn new(near: PS1, far: PS2) -> Self {
|
||||||
|
Self { near, far }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl<PS1, PS2> PathInfoService for Cache<PS1, PS2>
|
||||||
|
where
|
||||||
|
PS1: PathInfoService,
|
||||||
|
PS2: PathInfoService,
|
||||||
|
{
|
||||||
|
#[instrument(level = "trace", skip_all, fields(path_info.digest = nixbase32::encode(&digest)))]
|
||||||
|
async fn get(&self, digest: [u8; 20]) -> Result<Option<PathInfo>, Error> {
|
||||||
|
match self.near.get(digest).await? {
|
||||||
|
Some(path_info) => {
|
||||||
|
debug!("serving from cache");
|
||||||
|
Ok(Some(path_info))
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
debug!("not found in near, asking remote…");
|
||||||
|
match self.far.get(digest).await? {
|
||||||
|
None => Ok(None),
|
||||||
|
Some(path_info) => {
|
||||||
|
debug!("found in remote, adding to cache");
|
||||||
|
self.near.put(path_info.clone()).await?;
|
||||||
|
Ok(Some(path_info))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn put(&self, _path_info: PathInfo) -> Result<PathInfo, Error> {
|
||||||
|
Err(Error::StorageError("unimplemented".to_string()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn list(&self) -> BoxStream<'static, Result<PathInfo, Error>> {
|
||||||
|
Box::pin(tokio_stream::once(Err(Error::StorageError(
|
||||||
|
"unimplemented".to_string(),
|
||||||
|
))))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod test {
|
||||||
|
use std::num::NonZeroUsize;
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
pathinfoservice::{LruPathInfoService, MemoryPathInfoService, PathInfoService},
|
||||||
|
tests::fixtures::PATH_INFO_WITH_NARINFO,
|
||||||
|
};
|
||||||
|
|
||||||
|
const PATH_INFO_DIGEST: [u8; 20] = [0; 20];
|
||||||
|
|
||||||
|
/// Helper function setting up an instance of a "far" and "near"
|
||||||
|
/// PathInfoService.
|
||||||
|
async fn create_pathinfoservice() -> super::Cache<LruPathInfoService, MemoryPathInfoService> {
|
||||||
|
// Create an instance of a "far" PathInfoService.
|
||||||
|
let far = MemoryPathInfoService::default();
|
||||||
|
|
||||||
|
// … and an instance of a "near" PathInfoService.
|
||||||
|
let near = LruPathInfoService::with_capacity(NonZeroUsize::new(1).unwrap());
|
||||||
|
|
||||||
|
// create a Pathinfoservice combining the two and return it.
|
||||||
|
super::Cache::new(near, far)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Getting from the far backend is gonna insert it into the near one.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn test_populate_cache() {
|
||||||
|
let svc = create_pathinfoservice().await;
|
||||||
|
|
||||||
|
// query the PathInfo, things should not be there.
|
||||||
|
assert!(svc.get(PATH_INFO_DIGEST).await.unwrap().is_none());
|
||||||
|
|
||||||
|
// insert it into the far one.
|
||||||
|
svc.far.put(PATH_INFO_WITH_NARINFO.clone()).await.unwrap();
|
||||||
|
|
||||||
|
// now try getting it again, it should succeed.
|
||||||
|
assert_eq!(
|
||||||
|
Some(PATH_INFO_WITH_NARINFO.clone()),
|
||||||
|
svc.get(PATH_INFO_DIGEST).await.unwrap()
|
||||||
|
);
|
||||||
|
|
||||||
|
// peek near, it should now be there.
|
||||||
|
assert_eq!(
|
||||||
|
Some(PATH_INFO_WITH_NARINFO.clone()),
|
||||||
|
svc.near.get(PATH_INFO_DIGEST).await.unwrap()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
|
@ -1,3 +1,4 @@
|
||||||
|
mod combinators;
|
||||||
mod from_addr;
|
mod from_addr;
|
||||||
mod grpc;
|
mod grpc;
|
||||||
mod lru;
|
mod lru;
|
||||||
|
@ -17,6 +18,7 @@ use tvix_castore::Error;
|
||||||
|
|
||||||
use crate::proto::PathInfo;
|
use crate::proto::PathInfo;
|
||||||
|
|
||||||
|
pub use self::combinators::Cache as CachePathInfoService;
|
||||||
pub use self::from_addr::from_addr;
|
pub use self::from_addr::from_addr;
|
||||||
pub use self::grpc::GRPCPathInfoService;
|
pub use self::grpc::GRPCPathInfoService;
|
||||||
pub use self::lru::LruPathInfoService;
|
pub use self::lru::LruPathInfoService;
|
||||||
|
|
Loading…
Reference in a new issue