mdebray/tvl-depot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(tvix/castore): remove name from Nodes

Browse source 
Nodes only have names if they're contained inside a Directory, or if
they're a root node and have something else possibly giving them a name
externally.

This removes all `name` fields in the three different Nodes, and instead
maintains it inside a BTreeMap inside the Directory.

It also removes the NamedNode trait (they don't have a get_name()), as
well as Node::rename(self, name), and all [Partial]Ord implementations
for Node (as they don't have names to use for sorting).

The `nodes()`, `directories()`, `files()` iterators inside a `Directory`
now return a tuple of Name and Node, as does the RootNodesProvider.

The different {Directory,File,Symlink}Node struct constructors got
simpler, and the {Directory,File}Node ones became infallible - as
there's no more possibility to represent invalid state.

The proto structs stayed the same - there's now from_name_and_node and
into_name_and_node to convert back and forth between the two `Node`
structs.

Some further cleanups:

The error types for Node validation were renamed. Everything related to
names is now in the DirectoryError (not yet happy about the naming)

There's some leftover cleanups to do:
 - There should be a from_(sorted_)iter and into_iter in Directory, so
   we can construct and deconstruct in one go.
   That should also enable us to implement conversions from and to the
   proto representation that moves, rather than clones.

 - The BuildRequest and PathInfo structs are still proto-based, so we
   still do a bunch of conversions back and forth there (and have some
   ugly expect there). There's not much point for error handling here,
   this will be moved to stricter types in a followup CL.

Change-Id: I7369a8e3a426f44419c349077cb4fcab2044ebb6
Reviewed-on: https://cl.tvl.fyi/c/depot/+/12205
Tested-by: BuildkiteCI
Reviewed-by: yuka <yuka@yuka.dev>
Autosubmit: flokli <flokli@flokli.de>
Reviewed-by: benjaminedwardwebb <benjaminedwardwebb@gmail.com>
Reviewed-by: Connor Brewster <cbrewster@hey.com>
This commit is contained in:
Florian Klink 2024-08-14 22:00:12 +03:00 committed by clbot
parent 04e9531e65
commit 49b173786c
46 changed files with 785 additions and 1002 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
tvix/build/src/proto/mod.rs
View file

@ -1,8 +1,7 @@
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use itertools::Itertools; use itertools::Itertools;
use tvix_castore::ValidateNodeError; use tvix_castore::DirectoryError;
use tvix_castore::{NamedNode, Node};
mod grpc_buildservice_wrapper; mod grpc_buildservice_wrapper;
@ -20,7 +19,7 @@ pub const FILE_DESCRIPTOR_SET: &[u8] = tonic::include_file_descriptor_set!("tvix
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum ValidateBuildRequestError { pub enum ValidateBuildRequestError {
#[error("invalid input node at position {0}: {1}")] #[error("invalid input node at position {0}: {1}")]
InvalidInputNode(usize, ValidateNodeError), InvalidInputNode(usize, DirectoryError),
#[error("input nodes are not sorted by name")] #[error("input nodes are not sorted by name")]
InputNodesNotSorted, InputNodesNotSorted,
@ -124,19 +123,21 @@ impl BuildRequest {
/// and all restrictions around paths themselves (relative, clean, …) need /// and all restrictions around paths themselves (relative, clean, …) need
// to be fulfilled. // to be fulfilled.
pub fn validate(&self) -> Result<(), ValidateBuildRequestError> { pub fn validate(&self) -> Result<(), ValidateBuildRequestError> {
// now we can look at the names, and make sure they're sorted. // validate names. Make sure they're sorted
if !is_sorted(
self.inputs let mut last_name = bytes::Bytes::new();
.iter() for (i, node) in self.inputs.iter().enumerate() {
// TODO(flokli) handle conversion errors and store result somewhere // TODO(flokli): store result somewhere
.map(|e| { let (name, _node) = node
Node::try_from(e.node.as_ref().unwrap())
.unwrap()
.get_name()
.clone() .clone()
}), .into_name_and_node()
) { .map_err(|e| ValidateBuildRequestError::InvalidInputNode(i, e))?;
Err(ValidateBuildRequestError::InputNodesNotSorted)?
if name <= last_name {
return Err(ValidateBuildRequestError::InputNodesNotSorted);
} else {
last_name = name
}
} }
// validate working_dir // validate working_dir

23
tvix/castore/src/directoryservice/directory_graph.rs
View file

@ -10,7 +10,7 @@ use petgraph::{
use tracing::instrument; use tracing::instrument;
use super::order_validator::{LeavesToRootValidator, OrderValidator, RootToLeavesValidator}; use super::order_validator::{LeavesToRootValidator, OrderValidator, RootToLeavesValidator};
use crate::{B3Digest, Directory, DirectoryNode, NamedNode}; use crate::{B3Digest, Directory, DirectoryNode};
#[derive(thiserror::Error, Debug)] #[derive(thiserror::Error, Debug)]
pub enum Error { pub enum Error {
@ -69,12 +69,12 @@ pub struct ValidatedDirectoryGraph {
root: Option<NodeIndex>, root: Option<NodeIndex>,
} }
fn check_edge(dir: &DirectoryNode, child: &Directory) -> Result<(), Error> { fn check_edge(dir: &DirectoryNode, dir_name: &[u8], child: &Directory) -> Result<(), Error> {
// Ensure the size specified in the child node matches our records. // Ensure the size specified in the child node matches our records.
if dir.size() != child.size() { if dir.size() != child.size() {
return Err(Error::ValidationError(format!( return Err(Error::ValidationError(format!(
"'{}' has wrong size, specified {}, recorded {}", "'{}' has wrong size, specified {}, recorded {}",
dir.get_name().as_bstr(), dir_name.as_bstr(),
dir.size(), dir.size(),
child.size(), child.size(),
))); )));
@ -141,19 +141,19 @@ impl<O: OrderValidator> DirectoryGraph<O> {
} }
// set up edges to all child directories // set up edges to all child directories
for subdir in directory.directories() { for (subdir_name, subdir_node) in directory.directories() {
let child_ix = *self let child_ix = *self
.digest_to_node_ix .digest_to_node_ix
.entry(subdir.digest().clone()) .entry(subdir_node.digest().clone())
.or_insert_with(|| self.graph.add_node(None)); .or_insert_with(|| self.graph.add_node(None));
let pending_edge_check = match &self.graph[child_ix] { let pending_edge_check = match &self.graph[child_ix] {
Some(child) => { Some(child) => {
// child is already available, validate the edge now // child is already available, validate the edge now
check_edge(subdir, child)?; check_edge(subdir_node, subdir_name, child)?;
None None
} }
None => Some(subdir.clone()), // pending validation None => Some(subdir_node.clone()), // pending validation
}; };
self.graph.add_edge(ix, child_ix, pending_edge_check); self.graph.add_edge(ix, child_ix, pending_edge_check);
} }
@ -173,7 +173,9 @@ impl<O: OrderValidator> DirectoryGraph<O> {
.expect("edge not found") .expect("edge not found")
.take() .take()
.expect("edge is already validated"); .expect("edge is already validated");
check_edge(&edge_weight, &directory)?;
// TODO: where's the name here?
check_edge(&edge_weight, b"??", &directory)?;
} }
// finally, store the directory information in the node weight // finally, store the directory information in the node weight
@ -277,11 +279,12 @@ mod tests {
lazy_static! { lazy_static! {
pub static ref BROKEN_PARENT_DIRECTORY: Directory = { pub static ref BROKEN_PARENT_DIRECTORY: Directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
"foo".into(), "foo".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_A.digest(), DIRECTORY_A.digest(),
DIRECTORY_A.size() + 42, // wrong! DIRECTORY_A.size() + 42, // wrong!
).unwrap())).unwrap(); ))).unwrap();
dir dir
}; };
} }

5
tvix/castore/src/directoryservice/grpc.rs
View file

@ -3,8 +3,7 @@ use std::collections::HashSet;
use super::{Directory, DirectoryPutter, DirectoryService}; use super::{Directory, DirectoryPutter, DirectoryService};
use crate::composition::{CompositionContext, ServiceBuilder}; use crate::composition::{CompositionContext, ServiceBuilder};
use crate::proto::{self, get_directory_request::ByWhat}; use crate::proto::{self, get_directory_request::ByWhat};
use crate::ValidateDirectoryError; use crate::{B3Digest, DirectoryError, Error};
use crate::{B3Digest, Error};
use async_stream::try_stream; use async_stream::try_stream;
use futures::stream::BoxStream; use futures::stream::BoxStream;
use std::sync::Arc; use std::sync::Arc;
@ -154,7 +153,7 @@ where
} }
let directory = directory.try_into() let directory = directory.try_into()
.map_err(|e: ValidateDirectoryError| Error::StorageError(e.to_string()))?; .map_err(|e: DirectoryError| Error::StorageError(e.to_string()))?;
yield directory; yield directory;
}, },

10
tvix/castore/src/directoryservice/order_validator.rs
View file

@ -48,9 +48,9 @@ impl RootToLeavesValidator {
self.expected_digests.insert(directory.digest()); self.expected_digests.insert(directory.digest());
} }
for subdir in directory.directories() { for (_, subdir_node) in directory.directories() {
// Allow the children to appear next // Allow the children to appear next
self.expected_digests.insert(subdir.digest().clone()); self.expected_digests.insert(subdir_node.digest().clone());
} }
} }
} }
@ -79,11 +79,11 @@ impl OrderValidator for LeavesToRootValidator {
fn add_directory(&mut self, directory: &Directory) -> bool { fn add_directory(&mut self, directory: &Directory) -> bool {
let digest = directory.digest(); let digest = directory.digest();
for subdir in directory.directories() { for (_, subdir_node) in directory.directories() {
if !self.allowed_references.contains(subdir.digest()) { if !self.allowed_references.contains(subdir_node.digest()) {
warn!( warn!(
directory.digest = %digest, directory.digest = %digest,
subdirectory.digest = %subdir.digest(), subdirectory.digest = %subdir_node.digest(),
"unexpected directory reference" "unexpected directory reference"
); );
return false; return false;

7
tvix/castore/src/directoryservice/tests/mod.rs
View file

@ -218,14 +218,13 @@ async fn upload_reject_dangling_pointer(directory_service: impl DirectoryService
async fn upload_reject_wrong_size(directory_service: impl DirectoryService) { async fn upload_reject_wrong_size(directory_service: impl DirectoryService) {
let wrong_parent_directory = { let wrong_parent_directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::Directory( dir.add(
DirectoryNode::new(
"foo".into(), "foo".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_A.digest(), DIRECTORY_A.digest(),
DIRECTORY_A.size() + 42, // wrong! DIRECTORY_A.size() + 42, // wrong!
)),
) )
.unwrap(),
))
.unwrap(); .unwrap();
dir dir
}; };

29
tvix/castore/src/directoryservice/traverse.rs
View file

@ -1,4 +1,4 @@
use crate::{directoryservice::DirectoryService, Error, NamedNode, Node, Path}; use crate::{directoryservice::DirectoryService, Error, Node, Path};
use tracing::{instrument, warn}; use tracing::{instrument, warn};
/// This descends from a (root) node to the given (sub)path, returning the Node /// This descends from a (root) node to the given (sub)path, returning the Node
@ -37,9 +37,10 @@ where
})?; })?;
// look for the component in the [Directory]. // look for the component in the [Directory].
// FUTUREWORK: as the nodes() iterator returns in a sorted fashion, we if let Some((_child_name, child_node)) = directory
// could stop as soon as e.name is larger than the search string. .nodes()
if let Some(child_node) = directory.nodes().find(|n| n.get_name() == component) { .find(|(name, _node)| name.as_ref() == component)
{
// child node found, update prev_node to that and continue. // child node found, update prev_node to that and continue.
parent_node = child_node.clone(); parent_node = child_node.clone();
} else { } else {
@ -81,21 +82,23 @@ mod tests {
handle.close().await.expect("must upload"); handle.close().await.expect("must upload");
// construct the node for DIRECTORY_COMPLICATED // construct the node for DIRECTORY_COMPLICATED
let node_directory_complicated = Node::Directory( let node_directory_complicated = Node::Directory(DirectoryNode::new(
DirectoryNode::new(
"doesntmatter".into(),
DIRECTORY_COMPLICATED.digest(), DIRECTORY_COMPLICATED.digest(),
DIRECTORY_COMPLICATED.size(), DIRECTORY_COMPLICATED.size(),
) ));
.unwrap(),
);
// construct the node for DIRECTORY_COMPLICATED // construct the node for DIRECTORY_COMPLICATED
let node_directory_with_keep = let node_directory_with_keep = Node::Directory(
Node::Directory(DIRECTORY_COMPLICATED.directories().next().unwrap().clone()); DIRECTORY_COMPLICATED
.directories()
.next()
.unwrap()
.1
.clone(),
);
// construct the node for the .keep file // construct the node for the .keep file
let node_file_keep = Node::File(DIRECTORY_WITH_KEEP.files().next().unwrap().clone()); let node_file_keep = Node::File(DIRECTORY_WITH_KEEP.files().next().unwrap().1.clone());
// traversal to an empty subpath should return the root node. // traversal to an empty subpath should return the root node.
{ {

11
tvix/castore/src/directoryservice/utils.rs
View file

@ -57,16 +57,15 @@ pub fn traverse_directory<'a, DS: DirectoryService + 'static>(
// enqueue all child directory digests to the work queue, as // enqueue all child directory digests to the work queue, as
// long as they're not part of the worklist or already sent. // long as they're not part of the worklist or already sent.
// This panics if the digest looks invalid, it's supposed to be checked first. // This panics if the digest looks invalid, it's supposed to be checked first.
for child_directory_node in current_directory.directories() { for (_, child_directory_node) in current_directory.directories() {
// TODO: propagate error let child_digest = child_directory_node.digest();
let child_digest: B3Digest = child_directory_node.digest().clone();
if worklist_directory_digests.contains(&child_digest) if worklist_directory_digests.contains(child_digest)
|| sent_directory_digests.contains(&child_digest) || sent_directory_digests.contains(child_digest)
{ {
continue; continue;
} }
worklist_directory_digests.push_back(child_digest); worklist_directory_digests.push_back(child_digest.clone());
} }
yield current_directory; yield current_directory;

46
tvix/castore/src/errors.rs
View file

@ -13,33 +13,12 @@ pub enum Error {
StorageError(String), StorageError(String),
} }
/// Errors that can occur during the validation of [Directory] messages. /// Errors that occur during construction of [crate::Node]
#[derive(Debug, thiserror::Error, PartialEq)]
pub enum ValidateDirectoryError {
/// Elements are not in sorted order
#[error("{:?} is not sorted", .0.as_bstr())]
WrongSorting(Vec<u8>),
/// Multiple elements with the same name encountered
#[error("{:?} is a duplicate name", .0.as_bstr())]
DuplicateName(Vec<u8>),
/// Invalid node
#[error("invalid node with name {:?}: {:?}", .0.as_bstr(), .1.to_string())]
InvalidNode(Vec<u8>, ValidateNodeError),
#[error("Total size exceeds u32::MAX")]
SizeOverflow,
}
/// Errors that occur during Node validation
#[derive(Debug, thiserror::Error, PartialEq)] #[derive(Debug, thiserror::Error, PartialEq)]
pub enum ValidateNodeError { pub enum ValidateNodeError {
#[error("No node set")]
NoNodeSet,
/// Invalid digest length encountered /// Invalid digest length encountered
#[error("invalid digest length: {0}")] #[error("invalid digest length: {0}")]
InvalidDigestLen(usize), InvalidDigestLen(usize),
/// Invalid name encountered
#[error("Invalid name: {}", .0.as_bstr())]
InvalidName(bytes::Bytes),
/// Invalid symlink target /// Invalid symlink target
#[error("Invalid symlink target: {}", .0.as_bstr())] #[error("Invalid symlink target: {}", .0.as_bstr())]
InvalidSymlinkTarget(bytes::Bytes), InvalidSymlinkTarget(bytes::Bytes),
@ -53,6 +32,29 @@ impl From<crate::digests::Error> for ValidateNodeError {
} }
} }
/// Errors that can occur when populating [crate::Directory] messages,
/// or parsing [crate::proto::Directory]
#[derive(Debug, thiserror::Error, PartialEq)]
pub enum DirectoryError {
/// Multiple elements with the same name encountered
#[error("{:?} is a duplicate name", .0.as_bstr())]
DuplicateName(Vec<u8>),
/// Node failed validation
#[error("invalid node with name {:?}: {:?}", .0.as_bstr(), .1.to_string())]
InvalidNode(bytes::Bytes, ValidateNodeError),
#[error("Total size exceeds u32::MAX")]
SizeOverflow,
/// Invalid name encountered
#[error("Invalid name: {}", .0.as_bstr())]
InvalidName(bytes::Bytes),
/// Elements are not in sorted order. Can only happen on protos
#[error("{:?} is not sorted", .0.as_bstr())]
WrongSorting(bytes::Bytes),
/// This can only happen if there's an unknown node type (on protos)
#[error("No node set")]
NoNodeSet,
}
impl From<JoinError> for Error { impl From<JoinError> for Error {
fn from(value: JoinError) -> Self { fn from(value: JoinError) -> Self {
Error::StorageError(value.to_string()) Error::StorageError(value.to_string())

67
tvix/castore/src/fixtures.rs
View file

@ -35,70 +35,83 @@ lazy_static! {
// Directories // Directories
pub static ref DIRECTORY_WITH_KEEP: Directory = { pub static ref DIRECTORY_WITH_KEEP: Directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::File(FileNode::new( dir.add(
b".keep".to_vec().into(), ".keep".into(),
Node::File(FileNode::new(
EMPTY_BLOB_DIGEST.clone(), EMPTY_BLOB_DIGEST.clone(),
0, 0,
false false
).unwrap())).unwrap(); ))).unwrap();
dir dir
}; };
pub static ref DIRECTORY_COMPLICATED: Directory = { pub static ref DIRECTORY_COMPLICATED: Directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
b"keep".to_vec().into(), "keep".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_WITH_KEEP.digest(), DIRECTORY_WITH_KEEP.digest(),
DIRECTORY_WITH_KEEP.size() DIRECTORY_WITH_KEEP.size()
).unwrap())).unwrap(); ))).unwrap();
dir.add(Node::File(FileNode::new( dir.add(
b".keep".to_vec().into(), ".keep".into(),
Node::File(FileNode::new(
EMPTY_BLOB_DIGEST.clone(), EMPTY_BLOB_DIGEST.clone(),
0, 0,
false false
).unwrap())).unwrap(); ))).unwrap();
dir.add(Node::Symlink(SymlinkNode::new( dir.add(
b"aa".to_vec().into(), "aa".into(),
Node::Symlink(SymlinkNode::new(
b"/nix/store/somewhereelse".to_vec().into() b"/nix/store/somewhereelse".to_vec().into()
).unwrap())).unwrap(); ).unwrap())).unwrap();
dir dir
}; };
pub static ref DIRECTORY_A: Directory = Directory::new(); pub static ref DIRECTORY_A: Directory = Directory::new();
pub static ref DIRECTORY_B: Directory = { pub static ref DIRECTORY_B: Directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
b"a".to_vec().into(), "a".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_A.digest(), DIRECTORY_A.digest(),
DIRECTORY_A.size(), DIRECTORY_A.size(),
).unwrap())).unwrap(); ))).unwrap();
dir dir
}; };
pub static ref DIRECTORY_C: Directory = { pub static ref DIRECTORY_C: Directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
b"a".to_vec().into(), "a".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_A.digest(), DIRECTORY_A.digest(),
DIRECTORY_A.size(), DIRECTORY_A.size(),
).unwrap())).unwrap(); ))).unwrap();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
b"a'".to_vec().into(), "a'".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_A.digest(), DIRECTORY_A.digest(),
DIRECTORY_A.size(), DIRECTORY_A.size(),
).unwrap())).unwrap(); ))).unwrap();
dir dir
}; };
pub static ref DIRECTORY_D: Directory = { pub static ref DIRECTORY_D: Directory = {
let mut dir = Directory::new(); let mut dir = Directory::new();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
b"a".to_vec().into(), "a".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_A.digest(), DIRECTORY_A.digest(),
DIRECTORY_A.size(), DIRECTORY_A.size(),
).unwrap())).unwrap(); ))).unwrap();
dir.add(Node::Directory(DirectoryNode::new( dir.add(
b"b".to_vec().into(), "b".into(),
Node::Directory(DirectoryNode::new(
DIRECTORY_B.digest(), DIRECTORY_B.digest(),
DIRECTORY_B.size(), DIRECTORY_B.size(),
).unwrap())).unwrap(); ))).unwrap();
dir
dir
}; };
} }

62
tvix/castore/src/fs/fuse/tests.rs
View file

@ -68,15 +68,11 @@ async fn populate_blob_a(
root_nodes.insert( root_nodes.insert(
BLOB_A_NAME.into(), BLOB_A_NAME.into(),
Node::File( Node::File(FileNode::new(
FileNode::new(
BLOB_A_NAME.into(),
fixtures::BLOB_A_DIGEST.clone(), fixtures::BLOB_A_DIGEST.clone(),
fixtures::BLOB_A.len() as u64, fixtures::BLOB_A.len() as u64,
false, false,
) )),
.unwrap(),
),
); );
} }
@ -92,15 +88,11 @@ async fn populate_blob_b(
root_nodes.insert( root_nodes.insert(
BLOB_B_NAME.into(), BLOB_B_NAME.into(),
Node::File( Node::File(FileNode::new(
FileNode::new(
BLOB_B_NAME.into(),
fixtures::BLOB_B_DIGEST.clone(), fixtures::BLOB_B_DIGEST.clone(),
fixtures::BLOB_B.len() as u64, fixtures::BLOB_B.len() as u64,
false, false,
) )),
.unwrap(),
),
); );
} }
@ -120,22 +112,18 @@ async fn populate_blob_helloworld(
root_nodes.insert( root_nodes.insert(
HELLOWORLD_BLOB_NAME.into(), HELLOWORLD_BLOB_NAME.into(),
Node::File( Node::File(FileNode::new(
FileNode::new(
HELLOWORLD_BLOB_NAME.into(),
fixtures::HELLOWORLD_BLOB_DIGEST.clone(), fixtures::HELLOWORLD_BLOB_DIGEST.clone(),
fixtures::HELLOWORLD_BLOB_CONTENTS.len() as u64, fixtures::HELLOWORLD_BLOB_CONTENTS.len() as u64,
true, true,
) )),
.unwrap(),
),
); );
} }
async fn populate_symlink(root_nodes: &mut BTreeMap<Bytes, Node>) { async fn populate_symlink(root_nodes: &mut BTreeMap<Bytes, Node>) {
root_nodes.insert( root_nodes.insert(
SYMLINK_NAME.into(), SYMLINK_NAME.into(),
Node::Symlink(SymlinkNode::new(SYMLINK_NAME.into(), BLOB_A_NAME.into()).unwrap()), Node::Symlink(SymlinkNode::new(BLOB_A_NAME.into()).unwrap()),
); );
} }
@ -144,9 +132,7 @@ async fn populate_symlink(root_nodes: &mut BTreeMap<Bytes, Node>) {
async fn populate_symlink2(root_nodes: &mut BTreeMap<Bytes, Node>) { async fn populate_symlink2(root_nodes: &mut BTreeMap<Bytes, Node>) {
root_nodes.insert( root_nodes.insert(
SYMLINK_NAME2.into(), SYMLINK_NAME2.into(),
Node::Symlink( Node::Symlink(SymlinkNode::new("/nix/store/somewhereelse".into()).unwrap()),
SymlinkNode::new(SYMLINK_NAME2.into(), "/nix/store/somewhereelse".into()).unwrap(),
),
); );
} }
@ -170,14 +156,10 @@ async fn populate_directory_with_keep(
root_nodes.insert( root_nodes.insert(
DIRECTORY_WITH_KEEP_NAME.into(), DIRECTORY_WITH_KEEP_NAME.into(),
Node::Directory( Node::Directory(DirectoryNode::new(
DirectoryNode::new(
DIRECTORY_WITH_KEEP_NAME.into(),
fixtures::DIRECTORY_WITH_KEEP.digest(), fixtures::DIRECTORY_WITH_KEEP.digest(),
fixtures::DIRECTORY_WITH_KEEP.size(), fixtures::DIRECTORY_WITH_KEEP.size(),
) )),
.unwrap(),
),
); );
} }
@ -186,14 +168,10 @@ async fn populate_directory_with_keep(
async fn populate_directorynode_without_directory(root_nodes: &mut BTreeMap<Bytes, Node>) { async fn populate_directorynode_without_directory(root_nodes: &mut BTreeMap<Bytes, Node>) {
root_nodes.insert( root_nodes.insert(
DIRECTORY_WITH_KEEP_NAME.into(), DIRECTORY_WITH_KEEP_NAME.into(),
Node::Directory( Node::Directory(DirectoryNode::new(
DirectoryNode::new(
DIRECTORY_WITH_KEEP_NAME.into(),
fixtures::DIRECTORY_WITH_KEEP.digest(), fixtures::DIRECTORY_WITH_KEEP.digest(),
fixtures::DIRECTORY_WITH_KEEP.size(), fixtures::DIRECTORY_WITH_KEEP.size(),
) )),
.unwrap(),
),
); );
} }
@ -201,15 +179,11 @@ async fn populate_directorynode_without_directory(root_nodes: &mut BTreeMap<Byte
async fn populate_filenode_without_blob(root_nodes: &mut BTreeMap<Bytes, Node>) { async fn populate_filenode_without_blob(root_nodes: &mut BTreeMap<Bytes, Node>) {
root_nodes.insert( root_nodes.insert(
BLOB_A_NAME.into(), BLOB_A_NAME.into(),
Node::File( Node::File(FileNode::new(
FileNode::new(
BLOB_A_NAME.into(),
fixtures::BLOB_A_DIGEST.clone(), fixtures::BLOB_A_DIGEST.clone(),
fixtures::BLOB_A.len() as u64, fixtures::BLOB_A.len() as u64,
false, false,
) )),
.unwrap(),
),
); );
} }
@ -239,14 +213,10 @@ async fn populate_directory_complicated(
root_nodes.insert( root_nodes.insert(
DIRECTORY_COMPLICATED_NAME.into(), DIRECTORY_COMPLICATED_NAME.into(),
Node::Directory( Node::Directory(DirectoryNode::new(
DirectoryNode::new(
DIRECTORY_COMPLICATED_NAME.into(),
fixtures::DIRECTORY_COMPLICATED.digest(), fixtures::DIRECTORY_COMPLICATED.digest(),
fixtures::DIRECTORY_COMPLICATED.size(), fixtures::DIRECTORY_COMPLICATED.size(),
) )),
.unwrap(),
),
); );
} }

23
tvix/castore/src/fs/inodes.rs
View file

@ -2,9 +2,7 @@
//! about inodes, which present tvix-castore nodes in a filesystem. //! about inodes, which present tvix-castore nodes in a filesystem.
use std::time::Duration; use std::time::Duration;
use bytes::Bytes; use crate::{B3Digest, Node};
use crate::{B3Digest, NamedNode, Node};
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub enum InodeData { pub enum InodeData {
@ -20,23 +18,18 @@ pub enum InodeData {
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub enum DirectoryInodeData { pub enum DirectoryInodeData {
Sparse(B3Digest, u64), // digest, size Sparse(B3Digest, u64), // digest, size
Populated(B3Digest, Vec<(u64, Node)>), // [(child_inode, node)] Populated(B3Digest, Vec<(u64, bytes::Bytes, Node)>), // [(child_inode, name, node)]
} }
impl InodeData { impl InodeData {
/// Constructs a new InodeData by consuming a [Node]. /// Constructs a new InodeData by consuming a [Node].
/// It splits off the orginal name, so it can be used later. pub fn from_node(node: &Node) -> Self {
pub fn from_node(node: &Node) -> (Self, Bytes) {
match node { match node {
Node::Directory(n) => ( Node::Directory(n) => {
Self::Directory(DirectoryInodeData::Sparse(n.digest().clone(), n.size())), Self::Directory(DirectoryInodeData::Sparse(n.digest().clone(), n.size()))
n.get_name().clone(), }
), Node::File(n) => Self::Regular(n.digest().clone(), n.size(), n.executable()),
Node::File(n) => ( Node::Symlink(n) => Self::Symlink(n.target().clone()),
Self::Regular(n.digest().clone(), n.size(), n.executable()),
n.get_name().clone(),
),
Node::Symlink(n) => (Self::Symlink(n.target().clone()), n.get_name().clone()),
} }
} }

54
tvix/castore/src/fs/mod.rs
View file

@ -18,7 +18,7 @@ use self::{
use crate::{ use crate::{
blobservice::{BlobReader, BlobService}, blobservice::{BlobReader, BlobService},
directoryservice::DirectoryService, directoryservice::DirectoryService,
{B3Digest, NamedNode, Node}, {B3Digest, Node},
}; };
use bstr::ByteVec; use bstr::ByteVec;
use bytes::Bytes; use bytes::Bytes;
@ -103,7 +103,7 @@ pub struct TvixStoreFs<BS, DS, RN> {
u64, u64,
( (
Span, Span,
Arc<Mutex<mpsc::Receiver<(usize, Result<Node, crate::Error>)>>>, Arc<Mutex<mpsc::Receiver<(usize, Result<(Bytes, Node), crate::Error>)>>>,
), ),
>, >,
>, >,
@ -165,8 +165,12 @@ where
/// It is ok if it's a [DirectoryInodeData::Sparse] - in that case, a lookup /// It is ok if it's a [DirectoryInodeData::Sparse] - in that case, a lookup
/// in self.directory_service is performed, and self.inode_tracker is updated with the /// in self.directory_service is performed, and self.inode_tracker is updated with the
/// [DirectoryInodeData::Populated]. /// [DirectoryInodeData::Populated].
#[allow(clippy::type_complexity)]
#[instrument(skip(self), err)] #[instrument(skip(self), err)]
fn get_directory_children(&self, ino: u64) -> io::Result<(B3Digest, Vec<(u64, Node)>)> { fn get_directory_children(
&self,
ino: u64,
) -> io::Result<(B3Digest, Vec<(u64, bytes::Bytes, Node)>)> {
let data = self.inode_tracker.read().get(ino).unwrap(); let data = self.inode_tracker.read().get(ino).unwrap();
match *data { match *data {
// if it's populated already, return children. // if it's populated already, return children.
@ -196,13 +200,13 @@ where
let children = { let children = {
let mut inode_tracker = self.inode_tracker.write(); let mut inode_tracker = self.inode_tracker.write();
let children: Vec<(u64, Node)> = directory let children: Vec<(u64, Bytes, Node)> = directory
.nodes() .nodes()
.map(|child_node| { .map(|(child_name, child_node)| {
let (inode_data, _) = InodeData::from_node(child_node); let inode_data = InodeData::from_node(child_node);
let child_ino = inode_tracker.put(inode_data); let child_ino = inode_tracker.put(inode_data);
(child_ino, child_node.clone()) (child_ino, child_name.to_owned(), child_node.clone())
}) })
.collect(); .collect();
@ -263,12 +267,6 @@ where
Ok(None) => Err(io::Error::from_raw_os_error(libc::ENOENT)), Ok(None) => Err(io::Error::from_raw_os_error(libc::ENOENT)),
// The root node does exist // The root node does exist
Ok(Some(root_node)) => { Ok(Some(root_node)) => {
// The name must match what's passed in the lookup, otherwise this is also a ENOENT.
if root_node.get_name() != name.to_bytes() {
debug!(root_node.name=?root_node.get_name(), found_node.name=%name.to_string_lossy(), "node name mismatch");
return Err(io::Error::from_raw_os_error(libc::ENOENT));
}
// Let's check if someone else beat us to updating the inode tracker and // Let's check if someone else beat us to updating the inode tracker and
// root_nodes map. This avoids locking inode_tracker for writing. // root_nodes map. This avoids locking inode_tracker for writing.
if let Some(ino) = self.root_nodes.read().get(name.to_bytes()) { if let Some(ino) = self.root_nodes.read().get(name.to_bytes()) {
@ -285,9 +283,9 @@ where
// insert the (sparse) inode data and register in // insert the (sparse) inode data and register in
// self.root_nodes. // self.root_nodes.
let (inode_data, name) = InodeData::from_node(&root_node); let inode_data = InodeData::from_node(&root_node);
let ino = inode_tracker.put(inode_data.clone()); let ino = inode_tracker.put(inode_data.clone());
root_nodes.insert(name, ino); root_nodes.insert(bytes::Bytes::copy_from_slice(name.to_bytes()), ino);
Ok((ino, Arc::new(inode_data))) Ok((ino, Arc::new(inode_data)))
} }
@ -362,7 +360,7 @@ where
// Search for that name in the list of children and return the FileAttrs. // Search for that name in the list of children and return the FileAttrs.
// in the children, find the one with the desired name. // in the children, find the one with the desired name.
if let Some((child_ino, _)) = children.iter().find(|e| e.1.get_name() == name.to_bytes()) { if let Some((child_ino, _, _)) = children.iter().find(|(_, n, _)| n == name.to_bytes()) {
// lookup the child [InodeData] in [self.inode_tracker]. // lookup the child [InodeData] in [self.inode_tracker].
// We know the inodes for children have already been allocated. // We know the inodes for children have already been allocated.
let child_inode_data = self.inode_tracker.read().get(*child_ino).unwrap(); let child_inode_data = self.inode_tracker.read().get(*child_ino).unwrap();
@ -398,8 +396,8 @@ where
self.tokio_handle.spawn( self.tokio_handle.spawn(
async move { async move {
let mut stream = root_nodes_provider.list().enumerate(); let mut stream = root_nodes_provider.list().enumerate();
while let Some(node) = stream.next().await { while let Some(e) = stream.next().await {
if tx.send(node).await.is_err() { if tx.send(e).await.is_err() {
// If we get a send error, it means the sync code // If we get a send error, it means the sync code
// doesn't want any more entries. // doesn't want any more entries.
break; break;
@ -461,12 +459,12 @@ where
.map_err(|_| crate::Error::StorageError("mutex poisoned".into()))?; .map_err(|_| crate::Error::StorageError("mutex poisoned".into()))?;
while let Some((i, n)) = rx.blocking_recv() { while let Some((i, n)) = rx.blocking_recv() {
let root_node = n.map_err(|e| { let (name, node) = n.map_err(|e| {
warn!("failed to retrieve root node: {}", e); warn!("failed to retrieve root node: {}", e);
io::Error::from_raw_os_error(libc::EIO) io::Error::from_raw_os_error(libc::EIO)
})?; })?;
let (inode_data, name) = InodeData::from_node(&root_node); let inode_data = InodeData::from_node(&node);
// obtain the inode, or allocate a new one. // obtain the inode, or allocate a new one.
let ino = self.get_inode_for_root_name(&name).unwrap_or_else(|| { let ino = self.get_inode_for_root_name(&name).unwrap_or_else(|| {
@ -495,15 +493,17 @@ where
let (parent_digest, children) = self.get_directory_children(inode)?; let (parent_digest, children) = self.get_directory_children(inode)?;
Span::current().record("directory.digest", parent_digest.to_string()); Span::current().record("directory.digest", parent_digest.to_string());
for (i, (ino, child_node)) in children.into_iter().skip(offset as usize).enumerate() { for (i, (ino, child_name, child_node)) in
let (inode_data, name) = InodeData::from_node(&child_node); children.into_iter().skip(offset as usize).enumerate()
{
let inode_data = InodeData::from_node(&child_node);
// the second parameter will become the "offset" parameter on the next call. // the second parameter will become the "offset" parameter on the next call.
let written = add_entry(fuse_backend_rs::api::filesystem::DirEntry { let written = add_entry(fuse_backend_rs::api::filesystem::DirEntry {
ino, ino,
offset: offset + (i as u64) + 1, offset: offset + (i as u64) + 1,
type_: inode_data.as_fuse_type(), type_: inode_data.as_fuse_type(),
name: &name, name: &child_name,
})?; })?;
// If the buffer is full, add_entry will return `Ok(0)`. // If the buffer is full, add_entry will return `Ok(0)`.
if written == 0 { if written == 0 {
@ -548,12 +548,12 @@ where
.map_err(|_| crate::Error::StorageError("mutex poisoned".into()))?; .map_err(|_| crate::Error::StorageError("mutex poisoned".into()))?;
while let Some((i, n)) = rx.blocking_recv() { while let Some((i, n)) = rx.blocking_recv() {
let root_node = n.map_err(|e| { let (name, node) = n.map_err(|e| {
warn!("failed to retrieve root node: {}", e); warn!("failed to retrieve root node: {}", e);
io::Error::from_raw_os_error(libc::EPERM) io::Error::from_raw_os_error(libc::EPERM)
})?; })?;
let (inode_data, name) = InodeData::from_node(&root_node); let inode_data = InodeData::from_node(&node);
// obtain the inode, or allocate a new one. // obtain the inode, or allocate a new one.
let ino = self.get_inode_for_root_name(&name).unwrap_or_else(|| { let ino = self.get_inode_for_root_name(&name).unwrap_or_else(|| {
@ -585,8 +585,8 @@ where
let (parent_digest, children) = self.get_directory_children(inode)?; let (parent_digest, children) = self.get_directory_children(inode)?;
Span::current().record("directory.digest", parent_digest.to_string()); Span::current().record("directory.digest", parent_digest.to_string());
for (i, (ino, child_node)) in children.into_iter().skip(offset as usize).enumerate() { for (i, (ino, name, child_node)) in children.into_iter().skip(offset as usize).enumerate() {
let (inode_data, name) = InodeData::from_node(&child_node); let inode_data = InodeData::from_node(&child_node);
// the second parameter will become the "offset" parameter on the next call. // the second parameter will become the "offset" parameter on the next call.
let written = add_entry( let written = add_entry(

13
tvix/castore/src/fs/root_nodes.rs
View file

@ -12,9 +12,10 @@ pub trait RootNodes: Send + Sync {
/// directory of the filesystem. /// directory of the filesystem.
async fn get_by_basename(&self, name: &[u8]) -> Result<Option<Node>, Error>; async fn get_by_basename(&self, name: &[u8]) -> Result<Option<Node>, Error>;
/// Lists all root CA nodes in the filesystem. An error can be returned /// Lists all root CA nodes in the filesystem, as a tuple of (base)name
/// in case listing is not allowed /// and Node.
fn list(&self) -> BoxStream<Result<Node, Error>>; /// An error can be returned in case listing is not allowed.
fn list(&self) -> BoxStream<Result<(bytes::Bytes, Node), Error>>;
} }
#[async_trait] #[async_trait]
@ -28,9 +29,11 @@ where
Ok(self.as_ref().get(name).cloned()) Ok(self.as_ref().get(name).cloned())
} }
fn list(&self) -> BoxStream<Result<Node, Error>> { fn list(&self) -> BoxStream<Result<(bytes::Bytes, Node), Error>> {
Box::pin(tokio_stream::iter( Box::pin(tokio_stream::iter(
self.as_ref().iter().map(|(_, v)| Ok(v.clone())), self.as_ref()
.iter()
.map(|(name, node)| Ok((name.to_owned(), node.to_owned()))),
)) ))
} }
} }

54
tvix/castore/src/import/mod.rs
View file

@ -59,14 +59,6 @@ where
// we break the loop manually. // we break the loop manually.
.expect("Tvix bug: unexpected end of stream")?; .expect("Tvix bug: unexpected end of stream")?;
let name = entry
.path()
.file_name()
// If this is the root node, it will have an empty name.
.unwrap_or_default()
.to_owned()
.into();
let node = match &mut entry { let node = match &mut entry {
IngestionEntry::Dir { .. } => { IngestionEntry::Dir { .. } => {
// If the entry is a directory, we traversed all its children (and // If the entry is a directory, we traversed all its children (and
@ -92,36 +84,22 @@ where
IngestionError::UploadDirectoryError(entry.path().to_owned(), e) IngestionError::UploadDirectoryError(entry.path().to_owned(), e)
})?; })?;
Node::Directory( Node::Directory(DirectoryNode::new(directory_digest, directory_size))
DirectoryNode::new(name, directory_digest, directory_size).map_err(|e| {
IngestionError::UploadDirectoryError(
entry.path().to_owned(),
crate::Error::StorageError(e.to_string()),
)
})?,
)
} }
IngestionEntry::Symlink { ref target, .. } => Node::Symlink( IngestionEntry::Symlink { ref target, .. } => {
SymlinkNode::new(name, target.to_owned().into()).map_err(|e| { Node::Symlink(SymlinkNode::new(target.to_owned().into()).map_err(|e| {
IngestionError::UploadDirectoryError( IngestionError::UploadDirectoryError(
entry.path().to_owned(), entry.path().to_owned(),
crate::Error::StorageError(e.to_string()), crate::Error::StorageError(e.to_string()),
) )
})?, })?)
), }
IngestionEntry::Regular { IngestionEntry::Regular {
size, size,
executable, executable,
digest, digest,
.. ..
} => Node::File( } => Node::File(FileNode::new(digest.clone(), *size, *executable)),
FileNode::new(name, digest.clone(), *size, *executable).map_err(|e| {
IngestionError::UploadDirectoryError(
entry.path().to_owned(),
crate::Error::StorageError(e.to_string()),
)
})?,
),
}; };
let parent = entry let parent = entry
@ -132,11 +110,19 @@ where
if parent == crate::Path::ROOT { if parent == crate::Path::ROOT {
break node; break node;
} else { } else {
let name = entry
.path()
.file_name()
// If this is the root node, it will have an empty name.
.unwrap_or_default()
.to_owned()
.into();
// record node in parent directory, creating a new [Directory] if not there yet. // record node in parent directory, creating a new [Directory] if not there yet.
directories directories
.entry(parent.to_owned()) .entry(parent.to_owned())
.or_default() .or_default()
.add(node) .add(name, node)
.map_err(|e| { .map_err(|e| {
IngestionError::UploadDirectoryError( IngestionError::UploadDirectoryError(
entry.path().to_owned(), entry.path().to_owned(),
@ -227,18 +213,18 @@ mod test {
executable: true, executable: true,
digest: DUMMY_DIGEST.clone(), digest: DUMMY_DIGEST.clone(),
}], }],
Node::File(FileNode::new("foo".into(), DUMMY_DIGEST.clone(), 42, true).unwrap()) Node::File(FileNode::new(DUMMY_DIGEST.clone(), 42, true))
)] )]
#[case::single_symlink(vec![IngestionEntry::Symlink { #[case::single_symlink(vec![IngestionEntry::Symlink {
path: "foo".parse().unwrap(), path: "foo".parse().unwrap(),
target: b"blub".into(), target: b"blub".into(),
}], }],
Node::Symlink(SymlinkNode::new("foo".into(), "blub".into()).unwrap()) Node::Symlink(SymlinkNode::new("blub".into()).unwrap())
)] )]
#[case::single_dir(vec![IngestionEntry::Dir { #[case::single_dir(vec![IngestionEntry::Dir {
path: "foo".parse().unwrap(), path: "foo".parse().unwrap(),
}], }],
Node::Directory(DirectoryNode::new("foo".into(), Directory::default().digest(), Directory::default().size()).unwrap()) Node::Directory(DirectoryNode::new(Directory::default().digest(), Directory::default().size()))
)] )]
#[case::dir_with_keep(vec![ #[case::dir_with_keep(vec![
IngestionEntry::Regular { IngestionEntry::Regular {
@ -251,7 +237,7 @@ mod test {
path: "foo".parse().unwrap(), path: "foo".parse().unwrap(),
}, },
], ],
Node::Directory(DirectoryNode::new("foo".into(), DIRECTORY_WITH_KEEP.digest(), DIRECTORY_WITH_KEEP.size()).unwrap()) Node::Directory(DirectoryNode::new(DIRECTORY_WITH_KEEP.digest(), DIRECTORY_WITH_KEEP.size()))
)] )]
/// This is intentionally a bit unsorted, though it still satisfies all /// This is intentionally a bit unsorted, though it still satisfies all
/// requirements we have on the order of elements in the stream. /// requirements we have on the order of elements in the stream.
@ -279,7 +265,7 @@ mod test {
path: "blub".parse().unwrap(), path: "blub".parse().unwrap(),
}, },
], ],
Node::Directory(DirectoryNode::new("blub".into(), DIRECTORY_COMPLICATED.digest(), DIRECTORY_COMPLICATED.size()).unwrap()) Node::Directory(DirectoryNode::new(DIRECTORY_COMPLICATED.digest(), DIRECTORY_COMPLICATED.size()))
)] )]
#[tokio::test] #[tokio::test]
async fn test_ingestion(#[case] entries: Vec<IngestionEntry>, #[case] exp_root_node: Node) { async fn test_ingestion(#[case] entries: Vec<IngestionEntry>, #[case] exp_root_node: Node) {

2
tvix/castore/src/lib.rs
View file

@ -21,7 +21,7 @@ pub mod proto;
pub mod tonic; pub mod tonic;
pub use digests::{B3Digest, B3_LEN}; pub use digests::{B3Digest, B3_LEN};
pub use errors::{Error, ValidateDirectoryError, ValidateNodeError}; pub use errors::{DirectoryError, Error, ValidateNodeError};
pub use hashing_reader::{B3HashingReader, HashingReader}; pub use hashing_reader::{B3HashingReader, HashingReader};
#[cfg(test)] #[cfg(test)]

208
tvix/castore/src/nodes/directory.rs
View file

@ -1,23 +1,25 @@
use crate::{ use std::collections::BTreeMap;
proto, B3Digest, DirectoryNode, FileNode, NamedNode, Node, SymlinkNode, ValidateDirectoryError,
ValidateNodeError,
};
/// A Directory can contain Directory, File or Symlink nodes. use crate::{errors::DirectoryError, proto, B3Digest, DirectoryNode, FileNode, Node, SymlinkNode};
/// Each of these nodes have a name attribute, which is the basename in that
/// directory and node type specific attributes. /// A Directory contains nodes, which can be Directory, File or Symlink nodes.
/// While a Node by itself may have any name, the names of Directory entries: /// It attached names to these nodes, which is the basename in that directory.
/// These names:
/// - MUST not contain slashes or null bytes /// - MUST not contain slashes or null bytes
/// - MUST not be '.' or '..' /// - MUST not be '.' or '..'
/// - MUST be unique across all three lists /// - MUST be unique across all three lists
#[derive(Default, Debug, Clone, PartialEq, Eq)] #[derive(Default, Debug, Clone, PartialEq, Eq)]
pub struct Directory { pub struct Directory {
nodes: Vec<Node>, nodes: BTreeMap<bytes::Bytes, Node>,
} }
impl Directory { impl Directory {
/// Constructs a new, empty Directory.
/// FUTUREWORK: provide a constructor from an interator of (sorted) names and nodes.
pub fn new() -> Self { pub fn new() -> Self {
Directory { nodes: vec![] } Directory {
nodes: BTreeMap::new(),
}
} }
/// The size of a directory is the number of all regular and symlink elements, /// The size of a directory is the number of all regular and symlink elements,
@ -25,7 +27,7 @@ impl Directory {
pub fn size(&self) -> u64 { pub fn size(&self) -> u64 {
// It's impossible to create a Directory where the size overflows, because we // It's impossible to create a Directory where the size overflows, because we
// check before every add() that the size won't overflow. // check before every add() that the size won't overflow.
(self.nodes.len() as u64) + self.directories().map(|e| e.size()).sum::<u64>() (self.nodes.len() as u64) + self.directories().map(|(_name, dn)| dn.size()).sum::<u64>()
} }
/// Calculates the digest of a Directory, which is the blake3 hash of a /// Calculates the digest of a Directory, which is the blake3 hash of a
@ -35,62 +37,66 @@ impl Directory {
} }
/// Allows iterating over all nodes (directories, files and symlinks) /// Allows iterating over all nodes (directories, files and symlinks)
/// ordered by their name. /// For each, it returns a tuple of its name and node.
pub fn nodes(&self) -> impl Iterator<Item = &Node> + Send + Sync + '_ { /// The elements are sorted by their names.
pub fn nodes(&self) -> impl Iterator<Item = (&bytes::Bytes, &Node)> + Send + Sync + '_ {
self.nodes.iter() self.nodes.iter()
} }
/// Allows iterating over the FileNode entries of this directory /// Allows iterating over the FileNode entries of this directory.
/// ordered by their name /// For each, it returns a tuple of its name and node.
pub fn files(&self) -> impl Iterator<Item = &FileNode> + Send + Sync + '_ { /// The elements are sorted by their names.
self.nodes.iter().filter_map(|node| match node { pub fn files(&self) -> impl Iterator<Item = (&bytes::Bytes, &FileNode)> + Send + Sync + '_ {
Node::File(n) => Some(n), self.nodes.iter().filter_map(|(name, node)| match node {
Node::File(n) => Some((name, n)),
_ => None, _ => None,
}) })
} }
/// Allows iterating over the subdirectories of this directory /// Allows iterating over the DirectoryNode entries (subdirectories) of this directory.
/// ordered by their name /// For each, it returns a tuple of its name and node.
pub fn directories(&self) -> impl Iterator<Item = &DirectoryNode> + Send + Sync + '_ { /// The elements are sorted by their names.
self.nodes.iter().filter_map(|node| match node { pub fn directories(
Node::Directory(n) => Some(n), &self,
) -> impl Iterator<Item = (&bytes::Bytes, &DirectoryNode)> + Send + Sync + '_ {
self.nodes.iter().filter_map(|(name, node)| match node {
Node::Directory(n) => Some((name, n)),
_ => None, _ => None,
}) })
} }
/// Allows iterating over the SymlinkNode entries of this directory /// Allows iterating over the SymlinkNode entries of this directory
/// ordered by their name /// For each, it returns a tuple of its name and node.
pub fn symlinks(&self) -> impl Iterator<Item = &SymlinkNode> + Send + Sync + '_ { /// The elements are sorted by their names.
self.nodes.iter().filter_map(|node| match node { pub fn symlinks(
Node::Symlink(n) => Some(n), &self,
) -> impl Iterator<Item = (&bytes::Bytes, &SymlinkNode)> + Send + Sync + '_ {
self.nodes.iter().filter_map(|(name, node)| match node {
Node::Symlink(n) => Some((name, n)),
_ => None, _ => None,
}) })
} }
/// Checks a Node name for validity as a directory entry /// Checks a Node name for validity as a directory entry
/// We disallow slashes, null bytes, '.', '..' and the empty string. /// We disallow slashes, null bytes, '.', '..' and the empty string.
pub(crate) fn validate_node_name(name: &[u8]) -> Result<(), ValidateNodeError> { pub(crate) fn is_valid_name(name: &[u8]) -> bool {
if name.is_empty() !(name.is_empty()
|| name == b".." || name == b".."
|| name == b"." || name == b"."
|| name.contains(&0x00) || name.contains(&0x00)
|| name.contains(&b'/') || name.contains(&b'/'))
{
Err(ValidateNodeError::InvalidName(name.to_owned().into()))
} else {
Ok(())
}
} }
/// Adds the specified [Node] to the [Directory], preserving sorted entries. /// Adds the specified [Node] to the [Directory] with a given name.
/// ///
/// Inserting an element that already exists with the same name in the directory will yield an /// Inserting an element that already exists with the same name in the directory will yield an
/// error. /// error.
/// Inserting an element will validate that its name fulfills the stricter requirements for /// Inserting an element will validate that its name fulfills the
/// directory entries and yield an error if it is not. /// requirements for directory entries and yield an error if it is not.
pub fn add(&mut self, node: Node) -> Result<(), ValidateDirectoryError> { pub fn add(&mut self, name: bytes::Bytes, node: Node) -> Result<(), DirectoryError> {
Self::validate_node_name(node.get_name()) if !Self::is_valid_name(&name) {
.map_err(|e| ValidateDirectoryError::InvalidNode(node.get_name().clone().into(), e))?; return Err(DirectoryError::InvalidName(name));
}
// Check that the even after adding this new directory entry, the size calculation will not // Check that the even after adding this new directory entry, the size calculation will not
// overflow // overflow
@ -104,25 +110,18 @@ impl Directory {
_ => 0, _ => 0,
}, },
]) ])
.ok_or(ValidateDirectoryError::SizeOverflow)?; .ok_or(DirectoryError::SizeOverflow)?;
// This assumes the [Directory] is sorted, since we don't allow accessing the nodes list match self.nodes.entry(name) {
// directly and all previous inserts should have been in-order std::collections::btree_map::Entry::Vacant(e) => {
let pos = match self e.insert(node);
.nodes
.binary_search_by_key(&node.get_name(), |n| n.get_name())
{
Err(pos) => pos, // There is no node with this name; good!
Ok(_) => {
return Err(ValidateDirectoryError::DuplicateName(
node.get_name().to_vec(),
))
}
};
self.nodes.insert(pos, node);
Ok(()) Ok(())
} }
std::collections::btree_map::Entry::Occupied(occupied) => {
Err(DirectoryError::DuplicateName(occupied.key().to_vec()))
}
}
}
} }
fn checked_sum(iter: impl IntoIterator<Item = u64>) -> Option<u64> { fn checked_sum(iter: impl IntoIterator<Item = u64>) -> Option<u64> {
@ -133,49 +132,58 @@ fn checked_sum(iter: impl IntoIterator<Item = u64>) -> Option<u64> {
mod test { mod test {
use super::{Directory, DirectoryNode, FileNode, Node, SymlinkNode}; use super::{Directory, DirectoryNode, FileNode, Node, SymlinkNode};
use crate::fixtures::DUMMY_DIGEST; use crate::fixtures::DUMMY_DIGEST;
use crate::ValidateDirectoryError; use crate::DirectoryError;
#[test] #[test]
fn add_nodes_to_directory() { fn add_nodes_to_directory() {
let mut d = Directory::new(); let mut d = Directory::new();
d.add(Node::Directory( d.add(
DirectoryNode::new("b".into(), DUMMY_DIGEST.clone(), 1).unwrap(), "b".into(),
)) Node::Directory(DirectoryNode::new(DUMMY_DIGEST.clone(), 1)),
)
.unwrap(); .unwrap();
d.add(Node::Directory( d.add(
DirectoryNode::new("a".into(), DUMMY_DIGEST.clone(), 1).unwrap(), "a".into(),
)) Node::Directory(DirectoryNode::new(DUMMY_DIGEST.clone(), 1)),
)
.unwrap(); .unwrap();
d.add(Node::Directory( d.add(
DirectoryNode::new("z".into(), DUMMY_DIGEST.clone(), 1).unwrap(), "z".into(),
)) Node::Directory(DirectoryNode::new(DUMMY_DIGEST.clone(), 1)),
)
.unwrap(); .unwrap();
d.add(Node::File( d.add(
FileNode::new("f".into(), DUMMY_DIGEST.clone(), 1, true).unwrap(), "f".into(),
)) Node::File(FileNode::new(DUMMY_DIGEST.clone(), 1, true)),
)
.unwrap(); .unwrap();
d.add(Node::File( d.add(
FileNode::new("c".into(), DUMMY_DIGEST.clone(), 1, true).unwrap(), "c".into(),
)) Node::File(FileNode::new(DUMMY_DIGEST.clone(), 1, true)),
)
.unwrap(); .unwrap();
d.add(Node::File( d.add(
FileNode::new("g".into(), DUMMY_DIGEST.clone(), 1, true).unwrap(), "g".into(),
)) Node::File(FileNode::new(DUMMY_DIGEST.clone(), 1, true)),
)
.unwrap(); .unwrap();
d.add(Node::Symlink( d.add(
SymlinkNode::new("t".into(), "a".into()).unwrap(), "t".into(),
)) Node::Symlink(SymlinkNode::new("a".into()).unwrap()),
)
.unwrap(); .unwrap();
d.add(Node::Symlink( d.add(
SymlinkNode::new("o".into(), "a".into()).unwrap(), "o".into(),
)) Node::Symlink(SymlinkNode::new("a".into()).unwrap()),
)
.unwrap(); .unwrap();
d.add(Node::Symlink( d.add(
SymlinkNode::new("e".into(), "a".into()).unwrap(), "e".into(),
)) Node::Symlink(SymlinkNode::new("a".into()).unwrap()),
)
.unwrap(); .unwrap();
// Convert to proto struct and back to ensure we are not generating any invalid structures // Convert to proto struct and back to ensure we are not generating any invalid structures
@ -188,10 +196,11 @@ mod test {
let mut d = Directory::new(); let mut d = Directory::new();
assert_eq!( assert_eq!(
d.add(Node::Directory( d.add(
DirectoryNode::new("foo".into(), DUMMY_DIGEST.clone(), u64::MAX).unwrap(), "foo".into(),
)), Node::Directory(DirectoryNode::new(DUMMY_DIGEST.clone(), u64::MAX))
Err(ValidateDirectoryError::SizeOverflow) ),
Err(DirectoryError::SizeOverflow)
); );
} }
@ -199,16 +208,18 @@ mod test {
fn add_duplicate_node_to_directory() { fn add_duplicate_node_to_directory() {
let mut d = Directory::new(); let mut d = Directory::new();
d.add(Node::Directory( d.add(
DirectoryNode::new("a".into(), DUMMY_DIGEST.clone(), 1).unwrap(), "a".into(),
)) Node::Directory(DirectoryNode::new(DUMMY_DIGEST.clone(), 1)),
)
.unwrap(); .unwrap();
assert_eq!( assert_eq!(
format!( format!(
"{}", "{}",
d.add(Node::File( d.add(
FileNode::new("a".into(), DUMMY_DIGEST.clone(), 1, true).unwrap(), "a".into(),
)) Node::File(FileNode::new(DUMMY_DIGEST.clone(), 1, true))
)
.expect_err("adding duplicate dir entry must fail") .expect_err("adding duplicate dir entry must fail")
), ),
"\"a\" is a duplicate name" "\"a\" is a duplicate name"
@ -220,13 +231,10 @@ mod test {
async fn directory_reject_invalid_name() { async fn directory_reject_invalid_name() {
let mut dir = Directory::new(); let mut dir = Directory::new();
assert!( assert!(
dir.add(Node::Symlink( dir.add(
SymlinkNode::new(
"".into(), // wrong! can not be added to directory "".into(), // wrong! can not be added to directory
"doesntmatter".into(), Node::Symlink(SymlinkNode::new("doesntmatter".into(),).unwrap())
) )
.unwrap()
))
.is_err(), .is_err(),
"invalid symlink entry be rejected" "invalid symlink entry be rejected"
); );

37
tvix/castore/src/nodes/directory_node.rs
View file

@ -1,13 +1,11 @@
use crate::{B3Digest, NamedNode, ValidateNodeError}; use crate::B3Digest;
/// A DirectoryNode is a pointer to a [Directory], by its [Directory::digest]. /// A DirectoryNode is a pointer to a [Directory], by its [Directory::digest].
/// It also gives it a `name` and `size`. /// It also records a`size`.
/// Such a node is either an element in the [Directory] it itself is contained in, /// Such a node is either an element in the [Directory] it itself is contained in,
/// or a standalone root node./ /// or a standalone root node./
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct DirectoryNode { pub struct DirectoryNode {
/// The (base)name of the directory
name: bytes::Bytes,
/// The blake3 hash of a Directory message, serialized in protobuf canonical form. /// The blake3 hash of a Directory message, serialized in protobuf canonical form.
digest: B3Digest, digest: B3Digest,
/// Number of child elements in the Directory referred to by `digest`. /// Number of child elements in the Directory referred to by `digest`.
@ -23,8 +21,8 @@ pub struct DirectoryNode {
} }
impl DirectoryNode { impl DirectoryNode {
pub fn new(name: bytes::Bytes, digest: B3Digest, size: u64) -> Result<Self, ValidateNodeError> { pub fn new(digest: B3Digest, size: u64) -> Self {
Ok(Self { name, digest, size }) Self { digest, size }
} }
pub fn digest(&self) -> &B3Digest { pub fn digest(&self) -> &B3Digest {
@ -34,31 +32,4 @@ impl DirectoryNode {
pub fn size(&self) -> u64 { pub fn size(&self) -> u64 {
self.size self.size
} }
pub fn rename(self, name: bytes::Bytes) -> Self {
Self { name, ..self }
}
}
impl PartialOrd for DirectoryNode {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for DirectoryNode {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.get_name().cmp(other.get_name())
}
}
impl NamedNode for &DirectoryNode {
fn get_name(&self) -> &bytes::Bytes {
&self.name
}
}
impl NamedNode for DirectoryNode {
fn get_name(&self) -> &bytes::Bytes {
&self.name
}
} }

44
tvix/castore/src/nodes/file_node.rs
View file

@ -1,11 +1,8 @@
use crate::{B3Digest, NamedNode, ValidateNodeError}; use crate::B3Digest;
/// A FileNode represents a regular or executable file in a Directory or at the root. /// A FileNode represents a regular or executable file in a Directory or at the root.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct FileNode { pub struct FileNode {
/// The (base)name of the file
name: bytes::Bytes,
/// The blake3 digest of the file contents /// The blake3 digest of the file contents
digest: B3Digest, digest: B3Digest,
@ -17,18 +14,12 @@ pub struct FileNode {
} }
impl FileNode { impl FileNode {
pub fn new( pub fn new(digest: B3Digest, size: u64, executable: bool) -> Self {
name: bytes::Bytes, Self {
digest: B3Digest,
size: u64,
executable: bool,
) -> Result<Self, ValidateNodeError> {
Ok(Self {
name,
digest, digest,
size, size,
executable, executable,
}) }
} }
pub fn digest(&self) -> &B3Digest { pub fn digest(&self) -> &B3Digest {
@ -42,31 +33,4 @@ impl FileNode {
pub fn executable(&self) -> bool { pub fn executable(&self) -> bool {
self.executable self.executable
} }
pub fn rename(self, name: bytes::Bytes) -> Self {
Self { name, ..self }
}
}
impl PartialOrd for FileNode {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for FileNode {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.get_name().cmp(other.get_name())
}
}
impl NamedNode for &FileNode {
fn get_name(&self) -> &bytes::Bytes {
&self.name
}
}
impl NamedNode for FileNode {
fn get_name(&self) -> &bytes::Bytes {
&self.name
}
} }

53
tvix/castore/src/nodes/mod.rs
View file

@ -4,66 +4,17 @@ mod directory_node;
mod file_node; mod file_node;
mod symlink_node; mod symlink_node;
use bytes::Bytes;
pub use directory::Directory; pub use directory::Directory;
pub use directory_node::DirectoryNode; pub use directory_node::DirectoryNode;
pub use file_node::FileNode; pub use file_node::FileNode;
pub use symlink_node::SymlinkNode; pub use symlink_node::SymlinkNode;
/// A Node is either a [DirectoryNode], [FileNode] or [SymlinkNode]. /// A Node is either a [DirectoryNode], [FileNode] or [SymlinkNode].
/// While a Node by itself may have any name, only those matching specific requirements /// Nodes themselves don't have names, what gives them names is either them
/// can can be added as entries to a [Directory] (see the documentation on [Directory] for details). /// being inside a [Directory], or a root node with its own name attached to it.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub enum Node { pub enum Node {
Directory(DirectoryNode), Directory(DirectoryNode),
File(FileNode), File(FileNode),
Symlink(SymlinkNode), Symlink(SymlinkNode),
} }
impl Node {
/// Returns the node with a new name.
pub fn rename(self, name: Bytes) -> Self {
match self {
Node::Directory(n) => Node::Directory(n.rename(name)),
Node::File(n) => Node::File(n.rename(name)),
Node::Symlink(n) => Node::Symlink(n.rename(name)),
}
}
}
impl PartialOrd for Node {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for Node {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.get_name().cmp(other.get_name())
}
}
/// NamedNode is implemented for [FileNode], [DirectoryNode] and [SymlinkNode]
/// and [Node], so we can ask all of them for the name easily.
pub trait NamedNode {
fn get_name(&self) -> &Bytes;
}
impl NamedNode for &Node {
fn get_name(&self) -> &Bytes {
match self {
Node::File(node_file) => node_file.get_name(),
Node::Directory(node_directory) => node_directory.get_name(),
Node::Symlink(node_symlink) => node_symlink.get_name(),
}
}
}
impl NamedNode for Node {
fn get_name(&self) -> &Bytes {
match self {
Node::File(node_file) => node_file.get_name(),
Node::Directory(node_directory) => node_directory.get_name(),
Node::Symlink(node_symlink) => node_symlink.get_name(),
}
}
}

35
tvix/castore/src/nodes/symlink_node.rs
View file

@ -1,50 +1,21 @@
use crate::{NamedNode, ValidateNodeError}; use crate::ValidateNodeError;
/// A SymlinkNode represents a symbolic link in a Directory or at the root. /// A SymlinkNode represents a symbolic link in a Directory or at the root.
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq)]
pub struct SymlinkNode { pub struct SymlinkNode {
/// The (base)name of the symlink
name: bytes::Bytes,
/// The target of the symlink. /// The target of the symlink.
target: bytes::Bytes, target: bytes::Bytes,
} }
impl SymlinkNode { impl SymlinkNode {
pub fn new(name: bytes::Bytes, target: bytes::Bytes) -> Result<Self, ValidateNodeError> { pub fn new(target: bytes::Bytes) -> Result<Self, ValidateNodeError> {
if target.is_empty() || target.contains(&b'\0') { if target.is_empty() || target.contains(&b'\0') {
return Err(ValidateNodeError::InvalidSymlinkTarget(target)); return Err(ValidateNodeError::InvalidSymlinkTarget(target));
} }
Ok(Self { name, target }) Ok(Self { target })
} }
pub fn target(&self) -> &bytes::Bytes { pub fn target(&self) -> &bytes::Bytes {
&self.target &self.target
} }
pub(crate) fn rename(self, name: bytes::Bytes) -> Self {
Self { name, ..self }
}
}
impl PartialOrd for SymlinkNode {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl Ord for SymlinkNode {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.get_name().cmp(other.get_name())
}
}
impl NamedNode for &SymlinkNode {
fn get_name(&self) -> &bytes::Bytes {
&self.name
}
}
impl NamedNode for SymlinkNode {
fn get_name(&self) -> &bytes::Bytes {
&self.name
}
} }

8
tvix/castore/src/path.rs
View file

@ -38,7 +38,9 @@ impl Path {
if !bytes.is_empty() { if !bytes.is_empty() {
// Ensure all components are valid castore node names. // Ensure all components are valid castore node names.
for component in bytes.split_str(b"/") { for component in bytes.split_str(b"/") {
Directory::validate_node_name(component).ok()?; if !Directory::is_valid_name(component) {
return None;
}
} }
} }
@ -211,7 +213,9 @@ impl PathBuf {
/// Adjoins `name` to self. /// Adjoins `name` to self.
pub fn try_push(&mut self, name: &[u8]) -> Result<(), std::io::Error> { pub fn try_push(&mut self, name: &[u8]) -> Result<(), std::io::Error> {
Directory::validate_node_name(name).map_err(|_| std::io::ErrorKind::InvalidData)?; if !Directory::is_valid_name(name) {
return Err(std::io::ErrorKind::InvalidData.into());
}
if !self.inner.is_empty() { if !self.inner.is_empty() {
self.inner.push(b'/'); self.inner.push(b'/');

4
tvix/castore/src/proto/grpc_directoryservice_wrapper.rs
View file

@ -1,5 +1,5 @@
use crate::directoryservice::{DirectoryGraph, DirectoryService, LeavesToRootValidator}; use crate::directoryservice::{DirectoryGraph, DirectoryService, LeavesToRootValidator};
use crate::{proto, B3Digest, ValidateDirectoryError}; use crate::{proto, B3Digest, DirectoryError};
use futures::stream::BoxStream; use futures::stream::BoxStream;
use futures::TryStreamExt; use futures::TryStreamExt;
use std::ops::Deref; use std::ops::Deref;
@ -84,7 +84,7 @@ where
let mut validator = DirectoryGraph::<LeavesToRootValidator>::default(); let mut validator = DirectoryGraph::<LeavesToRootValidator>::default();
while let Some(directory) = req_inner.message().await? { while let Some(directory) = req_inner.message().await? {
validator validator
.add(directory.try_into().map_err(|e: ValidateDirectoryError| { .add(directory.try_into().map_err(|e: DirectoryError| {
tonic::Status::new(tonic::Code::Internal, e.to_string()) tonic::Status::new(tonic::Code::Internal, e.to_string())
})?) })?)
.map_err(|e| tonic::Status::new(tonic::Code::Internal, e.to_string()))?; .map_err(|e| tonic::Status::new(tonic::Code::Internal, e.to_string()))?;

255
tvix/castore/src/proto/mod.rs
View file

@ -5,12 +5,10 @@ use prost::Message;
mod grpc_blobservice_wrapper; mod grpc_blobservice_wrapper;
mod grpc_directoryservice_wrapper; mod grpc_directoryservice_wrapper;
use crate::{B3Digest, DirectoryError};
pub use grpc_blobservice_wrapper::GRPCBlobServiceWrapper; pub use grpc_blobservice_wrapper::GRPCBlobServiceWrapper;
pub use grpc_directoryservice_wrapper::GRPCDirectoryServiceWrapper; pub use grpc_directoryservice_wrapper::GRPCDirectoryServiceWrapper;
use crate::NamedNode;
use crate::{B3Digest, ValidateDirectoryError, ValidateNodeError};
tonic::include_proto!("tvix.castore.v1"); tonic::include_proto!("tvix.castore.v1");
#[cfg(feature = "tonic-reflection")] #[cfg(feature = "tonic-reflection")]
@ -71,168 +69,83 @@ impl Directory {
/// Accepts a name, and a mutable reference to the previous name. /// Accepts a name, and a mutable reference to the previous name.
/// If the passed name is larger than the previous one, the reference is updated. /// If the passed name is larger than the previous one, the reference is updated.
/// If it's not, an error is returned. /// If it's not, an error is returned.
fn update_if_lt_prev<'n>( fn update_if_lt_prev<'n>(prev_name: &mut &'n [u8], name: &'n [u8]) -> Result<(), DirectoryError> {
prev_name: &mut &'n [u8],
name: &'n [u8],
) -> Result<(), ValidateDirectoryError> {
if *name < **prev_name { if *name < **prev_name {
return Err(ValidateDirectoryError::WrongSorting(name.to_vec())); return Err(DirectoryError::WrongSorting(bytes::Bytes::copy_from_slice(
name,
)));
} }
*prev_name = name; *prev_name = name;
Ok(()) Ok(())
} }
impl TryFrom<&node::Node> for crate::Node { // TODO: add a proper owned version here that moves various fields
type Error = ValidateNodeError;
fn try_from(node: &node::Node) -> Result<crate::Node, ValidateNodeError> {
Ok(match node {
node::Node::Directory(n) => crate::Node::Directory(n.try_into()?),
node::Node::File(n) => crate::Node::File(n.try_into()?),
node::Node::Symlink(n) => crate::Node::Symlink(n.try_into()?),
})
}
}
impl TryFrom<&Node> for crate::Node {
type Error = ValidateNodeError;
fn try_from(node: &Node) -> Result<crate::Node, ValidateNodeError> {
match node {
Node { node: None } => Err(ValidateNodeError::NoNodeSet),
Node { node: Some(node) } => node.try_into(),
}
}
}
impl TryFrom<&DirectoryNode> for crate::DirectoryNode {
type Error = ValidateNodeError;
fn try_from(node: &DirectoryNode) -> Result<crate::DirectoryNode, ValidateNodeError> {
crate::DirectoryNode::new(
node.name.clone(),
node.digest.clone().try_into()?,
node.size,
)
}
}
impl TryFrom<&SymlinkNode> for crate::SymlinkNode {
type Error = ValidateNodeError;
fn try_from(node: &SymlinkNode) -> Result<crate::SymlinkNode, ValidateNodeError> {
crate::SymlinkNode::new(node.name.clone(), node.target.clone())
}
}
impl TryFrom<&FileNode> for crate::FileNode {
type Error = ValidateNodeError;
fn try_from(node: &FileNode) -> Result<crate::FileNode, ValidateNodeError> {
crate::FileNode::new(
node.name.clone(),
node.digest.clone().try_into()?,
node.size,
node.executable,
)
}
}
impl TryFrom<Directory> for crate::Directory { impl TryFrom<Directory> for crate::Directory {
type Error = ValidateDirectoryError; type Error = DirectoryError;
fn try_from(directory: Directory) -> Result<crate::Directory, ValidateDirectoryError> { fn try_from(value: Directory) -> Result<Self, Self::Error> {
(&directory).try_into() (&value).try_into()
} }
} }
impl TryFrom<&Directory> for crate::Directory { impl TryFrom<&Directory> for crate::Directory {
type Error = ValidateDirectoryError; type Error = DirectoryError;
fn try_from(directory: &Directory) -> Result<crate::Directory, ValidateDirectoryError> { fn try_from(directory: &Directory) -> Result<crate::Directory, DirectoryError> {
let mut dir = crate::Directory::new(); let mut dir = crate::Directory::new();
let mut last_file_name: &[u8] = b""; let mut last_file_name: &[u8] = b"";
// TODO: this currently loops over all three types separately, rather
// than peeking and picking from where would be the next.
for file in directory.files.iter().map(move |file| { for file in directory.files.iter().map(move |file| {
update_if_lt_prev(&mut last_file_name, &file.name).map(|()| file.clone()) update_if_lt_prev(&mut last_file_name, &file.name).map(|()| file.clone())
}) { }) {
let file = file?; let file = file?;
dir.add(crate::Node::File((&file).try_into().map_err(|e| {
ValidateDirectoryError::InvalidNode(file.name.into(), e) let (name, node) = Node {
})?))?; node: Some(node::Node::File(file)),
}
.into_name_and_node()?;
dir.add(name, node)?;
} }
let mut last_directory_name: &[u8] = b""; let mut last_directory_name: &[u8] = b"";
for directory in directory.directories.iter().map(move |directory| { for directory in directory.directories.iter().map(move |directory| {
update_if_lt_prev(&mut last_directory_name, &directory.name).map(|()| directory.clone()) update_if_lt_prev(&mut last_directory_name, &directory.name).map(|()| directory.clone())
}) { }) {
let directory = directory?; let directory = directory?;
dir.add(crate::Node::Directory((&directory).try_into().map_err(
|e| ValidateDirectoryError::InvalidNode(directory.name.into(), e), let (name, node) = Node {
)?))?; node: Some(node::Node::Directory(directory)),
}
.into_name_and_node()?;
dir.add(name, node)?;
} }
let mut last_symlink_name: &[u8] = b""; let mut last_symlink_name: &[u8] = b"";
for symlink in directory.symlinks.iter().map(move |symlink| { for symlink in directory.symlinks.iter().map(move |symlink| {
update_if_lt_prev(&mut last_symlink_name, &symlink.name).map(|()| symlink.clone()) update_if_lt_prev(&mut last_symlink_name, &symlink.name).map(|()| symlink.clone())
}) { }) {
let symlink = symlink?; let symlink = symlink?;
dir.add(crate::Node::Symlink((&symlink).try_into().map_err(
|e| ValidateDirectoryError::InvalidNode(symlink.name.into(), e), let (name, node) = Node {
)?))?; node: Some(node::Node::Symlink(symlink)),
} }
.into_name_and_node()?;
dir.add(name, node)?;
}
Ok(dir) Ok(dir)
} }
} }
impl From<&crate::Node> for node::Node { // TODO: add a proper owned version here that moves various fields
fn from(node: &crate::Node) -> node::Node {
match node {
crate::Node::Directory(n) => node::Node::Directory(n.into()),
crate::Node::File(n) => node::Node::File(n.into()),
crate::Node::Symlink(n) => node::Node::Symlink(n.into()),
}
}
}
impl From<&crate::Node> for Node {
fn from(node: &crate::Node) -> Node {
Node {
node: Some(node.into()),
}
}
}
impl From<&crate::DirectoryNode> for DirectoryNode {
fn from(node: &crate::DirectoryNode) -> DirectoryNode {
DirectoryNode {
digest: node.digest().clone().into(),
size: node.size(),
name: node.get_name().clone(),
}
}
}
impl From<&crate::FileNode> for FileNode {
fn from(node: &crate::FileNode) -> FileNode {
FileNode {
digest: node.digest().clone().into(),
size: node.size(),
name: node.get_name().clone(),
executable: node.executable(),
}
}
}
impl From<&crate::SymlinkNode> for SymlinkNode {
fn from(node: &crate::SymlinkNode) -> SymlinkNode {
SymlinkNode {
name: node.get_name().clone(),
target: node.target().clone(),
}
}
}
impl From<crate::Directory> for Directory { impl From<crate::Directory> for Directory {
fn from(directory: crate::Directory) -> Directory { fn from(value: crate::Directory) -> Self {
(&directory).into() (&value).into()
} }
} }
@ -241,16 +154,25 @@ impl From<&crate::Directory> for Directory {
let mut directories = vec![]; let mut directories = vec![];
let mut files = vec![]; let mut files = vec![];
let mut symlinks = vec![]; let mut symlinks = vec![];
for node in directory.nodes() {
for (name, node) in directory.nodes() {
match node { match node {
crate::Node::File(n) => { crate::Node::File(n) => files.push(FileNode {
files.push(n.into()); name: name.clone(),
} digest: n.digest().to_owned().into(),
crate::Node::Directory(n) => { size: n.size(),
directories.push(n.into()); executable: n.executable(),
} }),
crate::Node::Directory(n) => directories.push(DirectoryNode {
name: name.clone(),
digest: n.digest().to_owned().into(),
size: n.size(),
}),
crate::Node::Symlink(n) => { crate::Node::Symlink(n) => {
symlinks.push(n.into()); symlinks.push(SymlinkNode {
name: name.clone(),
target: n.target().to_owned(),
});
} }
} }
} }
@ -262,6 +184,67 @@ impl From<&crate::Directory> for Directory {
} }
} }
impl Node {
/// Converts a proto [Node] to a [crate::Node], and splits off the name.
pub fn into_name_and_node(self) -> Result<(bytes::Bytes, crate::Node), DirectoryError> {
match self.node.ok_or_else(|| DirectoryError::NoNodeSet)? {
node::Node::Directory(n) => {
let digest = B3Digest::try_from(n.digest)
.map_err(|e| DirectoryError::InvalidNode(n.name.to_owned(), e.into()))?;
let node = crate::Node::Directory(crate::DirectoryNode::new(digest, n.size));
Ok((n.name, node))
}
node::Node::File(n) => {
let digest = B3Digest::try_from(n.digest)
.map_err(|e| DirectoryError::InvalidNode(n.name.to_owned(), e.into()))?;
let node = crate::Node::File(crate::FileNode::new(digest, n.size, n.executable));
Ok((n.name, node))
}
node::Node::Symlink(n) => {
let node = crate::Node::Symlink(
crate::SymlinkNode::new(n.target)
.map_err(|e| DirectoryError::InvalidNode(n.name.to_owned(), e))?,
);
Ok((n.name, node))
}
}
}
/// Construsts a [Node] from a name and [crate::Node].
pub fn from_name_and_node(name: bytes::Bytes, n: crate::Node) -> Self {
// TODO: make these pub(crate) so we can avoid cloning?
match n {
crate::Node::Directory(directory_node) => Self {
node: Some(node::Node::Directory(DirectoryNode {
name,
digest: directory_node.digest().to_owned().into(),
size: directory_node.size(),
})),
},
crate::Node::File(file_node) => Self {
node: Some(node::Node::File(FileNode {
name,
digest: file_node.digest().to_owned().into(),
size: file_node.size(),
executable: file_node.executable(),
})),
},
crate::Node::Symlink(symlink_node) => Self {
node: Some(node::Node::Symlink(SymlinkNode {
name,
target: symlink_node.target().to_owned(),
})),
},
}
}
}
impl StatBlobResponse { impl StatBlobResponse {
/// Validates a StatBlobResponse. All chunks must have valid blake3 digests. /// Validates a StatBlobResponse. All chunks must have valid blake3 digests.
/// It is allowed to send an empty list, if no more granular chunking is /// It is allowed to send an empty list, if no more granular chunking is

30
tvix/castore/src/proto/tests/directory.rs
View file

@ -1,4 +1,4 @@
use crate::proto::{Directory, DirectoryNode, FileNode, SymlinkNode, ValidateDirectoryError}; use crate::proto::{Directory, DirectoryError, DirectoryNode, FileNode, SymlinkNode};
use crate::ValidateNodeError; use crate::ValidateNodeError;
use hex_literal::hex; use hex_literal::hex;
@ -162,8 +162,8 @@ fn validate_invalid_names() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::InvalidNode(n, ValidateNodeError::InvalidName(_)) => { DirectoryError::InvalidName(n) => {
assert_eq!(n, b"") assert_eq!(n.as_ref(), b"")
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
}; };
@ -179,8 +179,8 @@ fn validate_invalid_names() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::InvalidNode(n, ValidateNodeError::InvalidName(_)) => { DirectoryError::InvalidName(n) => {
assert_eq!(n, b".") assert_eq!(n.as_ref(), b".")
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
}; };
@ -197,8 +197,8 @@ fn validate_invalid_names() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::InvalidNode(n, ValidateNodeError::InvalidName(_)) => { DirectoryError::InvalidName(n) => {
assert_eq!(n, b"..") assert_eq!(n.as_ref(), b"..")
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
}; };
@ -213,8 +213,8 @@ fn validate_invalid_names() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::InvalidNode(n, ValidateNodeError::InvalidName(_)) => { DirectoryError::InvalidName(n) => {
assert_eq!(n, b"\x00") assert_eq!(n.as_ref(), b"\x00")
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
}; };
@ -229,8 +229,8 @@ fn validate_invalid_names() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::InvalidNode(n, ValidateNodeError::InvalidName(_)) => { DirectoryError::InvalidName(n) => {
assert_eq!(n, b"foo/bar") assert_eq!(n.as_ref(), b"foo/bar")
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
}; };
@ -248,7 +248,7 @@ fn validate_invalid_digest() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::InvalidNode(_, ValidateNodeError::InvalidDigestLen(n)) => { DirectoryError::InvalidNode(_, ValidateNodeError::InvalidDigestLen(n)) => {
assert_eq!(n, 2) assert_eq!(n, 2)
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
@ -275,8 +275,8 @@ fn validate_sorting() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::WrongSorting(s) => { DirectoryError::WrongSorting(s) => {
assert_eq!(s, b"a"); assert_eq!(s.as_ref(), b"a");
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),
} }
@ -300,7 +300,7 @@ fn validate_sorting() {
..Default::default() ..Default::default()
}; };
match crate::Directory::try_from(d).expect_err("must fail") { match crate::Directory::try_from(d).expect_err("must fail") {
ValidateDirectoryError::DuplicateName(s) => { DirectoryError::DuplicateName(s) => {
assert_eq!(s, b"a"); assert_eq!(s, b"a");
} }
_ => panic!("unexpected error"), _ => panic!("unexpected error"),

35
tvix/castore/src/tests/import.rs
View file

@ -2,14 +2,10 @@ use crate::blobservice::{self, BlobService};
use crate::directoryservice; use crate::directoryservice;
use crate::fixtures::*; use crate::fixtures::*;
use crate::import::fs::ingest_path; use crate::import::fs::ingest_path;
use crate::proto;
use crate::{DirectoryNode, Node, SymlinkNode}; use crate::{DirectoryNode, Node, SymlinkNode};
use tempfile::TempDir; use tempfile::TempDir;
#[cfg(target_family = "unix")]
use std::os::unix::ffi::OsStrExt;
#[cfg(target_family = "unix")] #[cfg(target_family = "unix")]
#[tokio::test] #[tokio::test]
async fn symlink() { async fn symlink() {
@ -34,9 +30,7 @@ async fn symlink() {
.expect("must succeed"); .expect("must succeed");
assert_eq!( assert_eq!(
Node::Symlink( Node::Symlink(SymlinkNode::new("/nix/store/somewhereelse".into(),).unwrap()),
SymlinkNode::new("doesntmatter".into(), "/nix/store/somewhereelse".into(),).unwrap()
),
root_node, root_node,
) )
} }
@ -59,13 +53,12 @@ async fn single_file() {
.expect("must succeed"); .expect("must succeed");
assert_eq!( assert_eq!(
proto::node::Node::File(proto::FileNode { Node::File(crate::FileNode::new(
name: "root".into(), HELLOWORLD_BLOB_DIGEST.clone(),
digest: HELLOWORLD_BLOB_DIGEST.clone().into(), HELLOWORLD_BLOB_CONTENTS.len() as u64,
size: HELLOWORLD_BLOB_CONTENTS.len() as u64, false,
executable: false, )),
}), root_node,
(&root_node).into(),
); );
// ensure the blob has been uploaded // ensure the blob has been uploaded
@ -95,20 +88,10 @@ async fn complicated() {
// ensure root_node matched expectations // ensure root_node matched expectations
assert_eq!( assert_eq!(
Node::Directory( Node::Directory(DirectoryNode::new(
DirectoryNode::new(
tmpdir
.path()
.file_name()
.unwrap()
.as_bytes()
.to_owned()
.into(),
DIRECTORY_COMPLICATED.digest().clone(), DIRECTORY_COMPLICATED.digest().clone(),
DIRECTORY_COMPLICATED.size(), DIRECTORY_COMPLICATED.size(),
) )),
.unwrap()
),
root_node, root_node,
); );

10
tvix/glue/src/builtins/derivation.rs
View file

@ -577,10 +577,7 @@ pub(crate) mod derivation_builtins {
}) })
.map_err(DerivationError::InvalidDerivation)?; .map_err(DerivationError::InvalidDerivation)?;
let root_node = Node::File( let root_node = Node::File(FileNode::new(blob_digest, blob_size, false));
FileNode::new(store_path.to_string().into(), blob_digest, blob_size, false)
.map_err(|e| ErrorKind::TvixError(Rc::new(e)))?,
);
// calculate the nar hash // calculate the nar hash
let (nar_size, nar_sha256) = state let (nar_size, nar_sha256) = state
@ -600,7 +597,10 @@ pub(crate) mod derivation_builtins {
state state
.path_info_service .path_info_service
.put(PathInfo { .put(PathInfo {
node: Some((&root_node).into()), node: Some(tvix_castore::proto::Node::from_name_and_node(
store_path.to_string().into(),
root_node,
)),
references: reference_paths references: reference_paths
.iter() .iter()
.map(|x| bytes::Bytes::copy_from_slice(x.digest())) .map(|x| bytes::Bytes::copy_from_slice(x.digest()))

11
tvix/glue/src/builtins/import.rs
View file

@ -213,16 +213,7 @@ mod import_builtins {
.tokio_handle .tokio_handle
.block_on(async { blob_writer.close().await })?; .block_on(async { blob_writer.close().await })?;
let root_node = Node::File( let root_node = Node::File(FileNode::new(blob_digest, blob_size, false));
FileNode::new(
// The name gets set further down, while constructing the PathInfo.
"".into(),
blob_digest,
blob_size,
false,
)
.map_err(|e| tvix_eval::ErrorKind::TvixError(Rc::new(e)))?,
);
let ca_hash = if recursive_ingestion { let ca_hash = if recursive_ingestion {
let (_nar_size, nar_sha256) = state let (_nar_size, nar_sha256) = state

37
tvix/glue/src/fetchers/mod.rs
View file

@ -10,9 +10,7 @@ use tokio::io::{AsyncBufRead, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
use tokio_util::io::{InspectReader, InspectWriter}; use tokio_util::io::{InspectReader, InspectWriter};
use tracing::{instrument, warn, Span}; use tracing::{instrument, warn, Span};
use tracing_indicatif::span_ext::IndicatifSpanExt; use tracing_indicatif::span_ext::IndicatifSpanExt;
use tvix_castore::{ use tvix_castore::{blobservice::BlobService, directoryservice::DirectoryService, FileNode, Node};
blobservice::BlobService, directoryservice::DirectoryService, FileNode, Node, ValidateNodeError,
};
use tvix_store::{nar::NarCalculationService, pathinfoservice::PathInfoService, proto::PathInfo}; use tvix_store::{nar::NarCalculationService, pathinfoservice::PathInfoService, proto::PathInfo};
use url::Url; use url::Url;
@ -329,10 +327,7 @@ where
// Construct and return the FileNode describing the downloaded contents. // Construct and return the FileNode describing the downloaded contents.
Ok(( Ok((
Node::File( Node::File(FileNode::new(blob_writer.close().await?, blob_size, false)),
FileNode::new(vec![].into(), blob_writer.close().await?, blob_size, false)
.map_err(|e| FetcherError::Io(std::io::Error::other(e.to_string())))?,
),
CAHash::Flat(actual_hash), CAHash::Flat(actual_hash),
blob_size, blob_size,
)) ))
@ -527,13 +522,7 @@ where
// Construct and return the FileNode describing the downloaded contents, // Construct and return the FileNode describing the downloaded contents,
// make it executable. // make it executable.
let root_node = Node::File( let root_node = Node::File(FileNode::new(blob_digest, file_size, true));
FileNode::new(vec![].into(), blob_digest, file_size, true).map_err(
|e: ValidateNodeError| {
FetcherError::Io(std::io::Error::other(e.to_string()))
},
)?,
);
Ok((root_node, CAHash::Nar(actual_hash), file_size)) Ok((root_node, CAHash::Nar(actual_hash), file_size))
} }
@ -557,9 +546,6 @@ where
// Calculate the store path to return, by calculating from ca_hash. // Calculate the store path to return, by calculating from ca_hash.
let store_path = build_ca_path(name, &ca_hash, Vec::<String>::new(), false)?; let store_path = build_ca_path(name, &ca_hash, Vec::<String>::new(), false)?;
// Rename the node name to match the Store Path.
let node = node.rename(store_path.to_string().into());
// If the resulting hash is not a CAHash::Nar, we also need to invoke // If the resulting hash is not a CAHash::Nar, we also need to invoke
// `calculate_nar` to calculate this representation, as it's required in // `calculate_nar` to calculate this representation, as it's required in
// the [PathInfo]. // the [PathInfo].
@ -577,7 +563,10 @@ where
// Construct the PathInfo and persist it. // Construct the PathInfo and persist it.
let path_info = PathInfo { let path_info = PathInfo {
node: Some((&node).into()), node: Some(tvix_castore::proto::Node::from_name_and_node(
store_path.to_string().into(),
node.clone(),
)),
references: vec![], references: vec![],
narinfo: Some(tvix_store::proto::NarInfo { narinfo: Some(tvix_store::proto::NarInfo {
nar_size, nar_size,
@ -589,20 +578,12 @@ where
}), }),
}; };
let path_info = self self.path_info_service
.path_info_service
.put(path_info) .put(path_info)
.await .await
.map_err(|e| FetcherError::Io(e.into()))?; .map_err(|e| FetcherError::Io(e.into()))?;
Ok(( Ok((store_path, node))
store_path,
(&path_info.node.unwrap().node.unwrap())
.try_into()
.map_err(|e: ValidateNodeError| {
FetcherError::Io(std::io::Error::other(e.to_string()))
})?,
))
} }
} }

38
tvix/glue/src/tvix_build.rs
View file

@ -1,7 +1,7 @@
//! This module contains glue code translating from //! This module contains glue code translating from
//! [nix_compat::derivation::Derivation] to [tvix_build::proto::BuildRequest]. //! [nix_compat::derivation::Derivation] to [tvix_build::proto::BuildRequest].
use std::collections::{BTreeMap, BTreeSet}; use std::collections::BTreeMap;
use bytes::Bytes; use bytes::Bytes;
use nix_compat::{derivation::Derivation, nixbase32}; use nix_compat::{derivation::Derivation, nixbase32};
@ -39,7 +39,7 @@ const NIX_ENVIRONMENT_VARS: [(&str, &str); 12] = [
#[allow(clippy::mutable_key_type)] #[allow(clippy::mutable_key_type)]
pub(crate) fn derivation_to_build_request( pub(crate) fn derivation_to_build_request(
derivation: &Derivation, derivation: &Derivation,
inputs: BTreeSet<Node>, inputs: BTreeMap<bytes::Bytes, Node>,
) -> std::io::Result<BuildRequest> { ) -> std::io::Result<BuildRequest> {
debug_assert!(derivation.validate(true).is_ok(), "drv must validate"); debug_assert!(derivation.validate(true).is_ok(), "drv must validate");
@ -109,7 +109,10 @@ pub(crate) fn derivation_to_build_request(
.into_iter() .into_iter()
.map(|(key, value)| EnvVar { key, value }) .map(|(key, value)| EnvVar { key, value })
.collect(), .collect(),
inputs: inputs.iter().map(Into::into).collect(), inputs: inputs
.into_iter()
.map(|(name, node)| tvix_castore::proto::Node::from_name_and_node(name, node))
.collect(),
inputs_dir: nix_compat::store_path::STORE_DIR[1..].into(), inputs_dir: nix_compat::store_path::STORE_DIR[1..].into(),
constraints, constraints,
working_dir: "build".into(), working_dir: "build".into(),
@ -189,10 +192,9 @@ fn calculate_pass_as_file_env(k: &str) -> (String, String) {
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use std::collections::BTreeSet;
use bytes::Bytes; use bytes::Bytes;
use nix_compat::derivation::Derivation; use nix_compat::derivation::Derivation;
use std::collections::BTreeMap;
use tvix_build::proto::{ use tvix_build::proto::{
build_request::{AdditionalFile, BuildConstraints, EnvVar}, build_request::{AdditionalFile, BuildConstraints, EnvVar},
BuildRequest, BuildRequest,
@ -206,14 +208,9 @@ mod test {
use lazy_static::lazy_static; use lazy_static::lazy_static;
lazy_static! { lazy_static! {
static ref INPUT_NODE_FOO: Node = Node::Directory( static ref INPUT_NODE_FOO_NAME: Bytes = "mp57d33657rf34lzvlbpfa1gjfv5gmpg-bar".into();
DirectoryNode::new( static ref INPUT_NODE_FOO: Node =
Bytes::from("mp57d33657rf34lzvlbpfa1gjfv5gmpg-bar"), Node::Directory(DirectoryNode::new(DUMMY_DIGEST.clone(), 42,));
DUMMY_DIGEST.clone(),
42,
)
.unwrap()
);
} }
#[test] #[test]
@ -222,8 +219,10 @@ mod test {
let derivation = Derivation::from_aterm_bytes(aterm_bytes).expect("must parse"); let derivation = Derivation::from_aterm_bytes(aterm_bytes).expect("must parse");
let build_request = let build_request = derivation_to_build_request(
derivation_to_build_request(&derivation, BTreeSet::from([INPUT_NODE_FOO.clone()])) &derivation,
BTreeMap::from([(INPUT_NODE_FOO_NAME.clone(), INPUT_NODE_FOO.clone())]),
)
.expect("must succeed"); .expect("must succeed");
let mut expected_environment_vars = vec![ let mut expected_environment_vars = vec![
@ -261,7 +260,10 @@ mod test {
command_args: vec![":".into()], command_args: vec![":".into()],
outputs: vec!["nix/store/fhaj6gmwns62s6ypkcldbaj2ybvkhx3p-foo".into()], outputs: vec!["nix/store/fhaj6gmwns62s6ypkcldbaj2ybvkhx3p-foo".into()],
environment_vars: expected_environment_vars, environment_vars: expected_environment_vars,
inputs: vec![(&*INPUT_NODE_FOO).into()], inputs: vec![tvix_castore::proto::Node::from_name_and_node(
INPUT_NODE_FOO_NAME.clone(),
INPUT_NODE_FOO.clone()
)],
inputs_dir: "nix/store".into(), inputs_dir: "nix/store".into(),
constraints: Some(BuildConstraints { constraints: Some(BuildConstraints {
system: derivation.system.clone(), system: derivation.system.clone(),
@ -285,7 +287,7 @@ mod test {
let derivation = Derivation::from_aterm_bytes(aterm_bytes).expect("must parse"); let derivation = Derivation::from_aterm_bytes(aterm_bytes).expect("must parse");
let build_request = let build_request =
derivation_to_build_request(&derivation, BTreeSet::from([])).expect("must succeed"); derivation_to_build_request(&derivation, BTreeMap::from([])).expect("must succeed");
let mut expected_environment_vars = vec![ let mut expected_environment_vars = vec![
EnvVar { EnvVar {
@ -355,7 +357,7 @@ mod test {
let derivation = Derivation::from_aterm_bytes(aterm_bytes).expect("must parse"); let derivation = Derivation::from_aterm_bytes(aterm_bytes).expect("must parse");
let build_request = let build_request =
derivation_to_build_request(&derivation, BTreeSet::from([])).expect("must succeed"); derivation_to_build_request(&derivation, BTreeMap::from([])).expect("must succeed");
let mut expected_environment_vars = vec![ let mut expected_environment_vars = vec![
// Note how bar and baz are not present in the env anymore, // Note how bar and baz are not present in the env anymore,

89
tvix/glue/src/tvix_store_io.rs
View file

@ -4,9 +4,9 @@ use futures::{StreamExt, TryStreamExt};
use nix_compat::nixhash::NixHash; use nix_compat::nixhash::NixHash;
use nix_compat::store_path::StorePathRef; use nix_compat::store_path::StorePathRef;
use nix_compat::{nixhash::CAHash, store_path::StorePath}; use nix_compat::{nixhash::CAHash, store_path::StorePath};
use std::collections::BTreeMap;
use std::{ use std::{
cell::RefCell, cell::RefCell,
collections::BTreeSet,
io, io,
path::{Path, PathBuf}, path::{Path, PathBuf},
sync::Arc, sync::Arc,
@ -21,7 +21,7 @@ use tvix_store::nar::NarCalculationService;
use tvix_castore::{ use tvix_castore::{
blobservice::BlobService, blobservice::BlobService,
directoryservice::{self, DirectoryService}, directoryservice::{self, DirectoryService},
{NamedNode, Node}, Node,
}; };
use tvix_store::{pathinfoservice::PathInfoService, proto::PathInfo}; use tvix_store::{pathinfoservice::PathInfoService, proto::PathInfo};
@ -120,12 +120,22 @@ impl TvixStoreIO {
.await? .await?
{ {
// if we have a PathInfo, we know there will be a root_node (due to validation) // if we have a PathInfo, we know there will be a root_node (due to validation)
Some(path_info) => path_info // TODO: use stricter typed BuildRequest here.
Some(path_info) => {
let (name, node) = path_info
.node .node
.as_ref()
.expect("no node") .expect("no node")
.try_into() .into_name_and_node()
.expect("invalid node"), .expect("invalid node");
assert_eq!(
store_path.to_string().as_bytes(),
name.as_ref(),
"returned node basename must match requested store path"
);
node
}
// If there's no PathInfo found, this normally means we have to // If there's no PathInfo found, this normally means we have to
// trigger the build (and insert into PathInfoService, after // trigger the build (and insert into PathInfoService, after
// reference scanning). // reference scanning).
@ -189,7 +199,7 @@ impl TvixStoreIO {
// Provide them, which means, here is where we recursively build // Provide them, which means, here is where we recursively build
// all dependencies. // all dependencies.
#[allow(clippy::mutable_key_type)] #[allow(clippy::mutable_key_type)]
let mut input_nodes: BTreeSet<Node> = let mut inputs: BTreeMap<Bytes, Node> =
futures::stream::iter(drv.input_derivations.iter()) futures::stream::iter(drv.input_derivations.iter())
.map(|(input_drv_path, output_names)| { .map(|(input_drv_path, output_names)| {
// look up the derivation object // look up the derivation object
@ -217,6 +227,7 @@ impl TvixStoreIO {
.clone() .clone()
}) })
.collect(); .collect();
// For each output, ask for the castore node. // For each output, ask for the castore node.
// We're in a per-derivation context, so if they're // We're in a per-derivation context, so if they're
// not built yet they'll all get built together. // not built yet they'll all get built together.
@ -231,7 +242,7 @@ impl TvixStoreIO {
.await?; .await?;
if let Some(node) = node { if let Some(node) = node {
Ok(node) Ok((output_path.to_string().into(), node))
} else { } else {
Err(io::Error::other("no node produced")) Err(io::Error::other("no node produced"))
} }
@ -245,26 +256,30 @@ impl TvixStoreIO {
.try_collect() .try_collect()
.await?; .await?;
// add input sources
// FUTUREWORK: merge these who things together // FUTUREWORK: merge these who things together
#[allow(clippy::mutable_key_type)] #[allow(clippy::mutable_key_type)]
let input_nodes_input_sources: BTreeSet<Node> = // add input sources
let input_sources: BTreeMap<_, _> =
futures::stream::iter(drv.input_sources.iter()) futures::stream::iter(drv.input_sources.iter())
.then(|input_source| { .then(|input_source| {
Box::pin(async { Box::pin({
let input_source = input_source.clone();
async move {
let node = self let node = self
.store_path_to_node(input_source, Path::new("")) .store_path_to_node(&input_source, Path::new(""))
.await?; .await?;
if let Some(node) = node { if let Some(node) = node {
Ok(node) Ok((input_source.to_string().into(), node))
} else { } else {
Err(io::Error::other("no node produced")) Err(io::Error::other("no node produced"))
} }
}
}) })
}) })
.try_collect() .try_collect()
.await?; .await?;
input_nodes.extend(input_nodes_input_sources);
inputs.extend(input_sources);
span.pb_set_message(&format!("🔨Building {}", &store_path)); span.pb_set_message(&format!("🔨Building {}", &store_path));
@ -273,7 +288,7 @@ impl TvixStoreIO {
// operations, so dealt with in the Some(…) match arm // operations, so dealt with in the Some(…) match arm
// synthesize the build request. // synthesize the build request.
let build_request = derivation_to_build_request(&drv, input_nodes)?; let build_request = derivation_to_build_request(&drv, inputs)?;
// create a build // create a build
let build_result = self let build_result = self
@ -287,17 +302,21 @@ impl TvixStoreIO {
// For each output, insert a PathInfo. // For each output, insert a PathInfo.
for output in &build_result.outputs { for output in &build_result.outputs {
let root_node = output.try_into().expect("invalid root node"); let (output_name, output_node) =
output.clone().into_name_and_node().expect("invalid node");
// calculate the nar representation // calculate the nar representation
let (nar_size, nar_sha256) = self let (nar_size, nar_sha256) = self
.nar_calculation_service .nar_calculation_service
.calculate_nar(&root_node) .calculate_nar(&output_node)
.await?; .await?;
// assemble the PathInfo to persist // assemble the PathInfo to persist
let path_info = PathInfo { let path_info = PathInfo {
node: Some((&root_node).into()), node: Some(tvix_castore::proto::Node::from_name_and_node(
output_name,
output_node,
)),
references: vec![], // TODO: refscan references: vec![], // TODO: refscan
narinfo: Some(tvix_store::proto::NarInfo { narinfo: Some(tvix_store::proto::NarInfo {
nar_size, nar_size,
@ -330,14 +349,15 @@ impl TvixStoreIO {
} }
// find the output for the store path requested // find the output for the store path requested
let s = store_path.to_string();
build_result build_result
.outputs .outputs
.into_iter() .into_iter()
.map(|output_node| Node::try_from(&output_node).expect("invalid node")) .map(|e| e.into_name_and_node().expect("invalid node"))
.find(|output_node| { .find(|(output_name, _output_node)| output_name == s.as_bytes())
output_node.get_name() == store_path.to_string().as_bytes()
})
.expect("build didn't produce the store path") .expect("build didn't produce the store path")
.1
} }
} }
} }
@ -380,12 +400,16 @@ impl TvixStoreIO {
}, },
)?; )?;
// assemble a new root_node with a name that is derived from the nar hash. tvix_store::import::log_node(name.as_bytes(), &root_node, path);
let root_node = root_node.rename(output_path.to_string().into_bytes().into());
tvix_store::import::log_node(&root_node, path);
let path_info = // construct a PathInfo
tvix_store::import::derive_nar_ca_path_info(nar_size, nar_sha256, Some(ca), root_node); let path_info = tvix_store::import::derive_nar_ca_path_info(
nar_size,
nar_sha256,
Some(ca),
output_path.to_string().into(),
root_node,
);
Ok(( Ok((
path_info, path_info,
@ -540,13 +564,12 @@ impl EvalIO for TvixStoreIO {
self.directory_service.as_ref().get(&digest).await self.directory_service.as_ref().get(&digest).await
})? { })? {
let mut children: Vec<(bytes::Bytes, FileType)> = Vec::new(); let mut children: Vec<(bytes::Bytes, FileType)> = Vec::new();
for node in directory.nodes() { // TODO: into_nodes() to avoid cloning
for (name, node) in directory.nodes() {
children.push(match node { children.push(match node {
Node::Directory(e) => { Node::Directory(_) => (name.clone(), FileType::Directory),
(e.get_name().clone(), FileType::Directory) Node::File(_) => (name.clone(), FileType::Regular),
} Node::Symlink(_) => (name.clone(), FileType::Symlink),
Node::File(e) => (e.get_name().clone(), FileType::Regular),
Node::Symlink(e) => (e.get_name().clone(), FileType::Symlink),
}) })
} }
Ok(children) Ok(children)

2
tvix/nar-bridge/src/lib.rs
View file

@ -7,7 +7,7 @@ use std::num::NonZeroUsize;
use std::sync::Arc; use std::sync::Arc;
use tvix_castore::blobservice::BlobService; use tvix_castore::blobservice::BlobService;
use tvix_castore::directoryservice::DirectoryService; use tvix_castore::directoryservice::DirectoryService;
use tvix_castore::proto::node::Node; use tvix_castore::Node;
use tvix_store::pathinfoservice::PathInfoService; use tvix_store::pathinfoservice::PathInfoService;
mod nar; mod nar;

11
tvix/nar-bridge/src/nar.rs
View file

@ -52,14 +52,15 @@ pub async fn get(
StatusCode::NOT_FOUND StatusCode::NOT_FOUND
})?; })?;
let root_node: tvix_castore::Node = (&root_node).try_into().map_err(|e| { let (root_name, root_node) = root_node.into_name_and_node().map_err(|e| {
warn!(err=%e, "root node validation failed"); warn!(err=%e, "root node validation failed");
StatusCode::BAD_REQUEST StatusCode::BAD_REQUEST
})?; })?;
// validate the node, but add a dummy node name, as we only send unnamed if !root_name.is_empty() {
// nodes warn!("root node has name, which it shouldn't");
let root_node = root_node.rename("00000000000000000000000000000000-dummy".into()); return Err(StatusCode::BAD_REQUEST);
}
let (w, r) = tokio::io::duplex(1024 * 8); let (w, r) = tokio::io::duplex(1024 * 8);
@ -125,7 +126,7 @@ pub async fn put(
// store mapping of narhash to root node into root_nodes. // store mapping of narhash to root node into root_nodes.
// we need it later to populate the root node when accepting the PathInfo. // we need it later to populate the root node when accepting the PathInfo.
root_nodes.write().put(nar_hash_actual, (&root_node).into()); root_nodes.write().put(nar_hash_actual, root_node);
Ok("") Ok("")
} }

36
tvix/nar-bridge/src/narinfo.rs
View file

@ -1,8 +1,9 @@
use axum::http::StatusCode; use axum::http::StatusCode;
use bytes::Bytes; use bytes::Bytes;
use nix_compat::{narinfo::NarInfo, nixbase32}; use nix_compat::{narinfo::NarInfo, nixbase32};
use prost::Message;
use tracing::{instrument, warn, Span}; use tracing::{instrument, warn, Span};
use tvix_castore::proto::{self as castorepb, node::Node}; use tvix_castore::proto::{self as castorepb};
use tvix_store::proto::PathInfo; use tvix_store::proto::PathInfo;
use crate::AppState; use crate::AppState;
@ -67,17 +68,21 @@ pub async fn get(
})?; })?;
// encode the (unnamed) root node in the NAR url itself. // encode the (unnamed) root node in the NAR url itself.
let root_node = // We strip the name from the proto node before sending it out.
tvix_castore::Node::try_from(path_info.node.as_ref().expect("root node must not be none")) // It's not needed to render the NAR, it'll make the URL shorter, and it
.unwrap() // PathInfo is validated // will make caching these requests easier.
.rename("".into()); let (_, root_node) = path_info
.node
let mut buf = Vec::new(); .as_ref()
Node::encode(&(&root_node).into(), &mut buf); .expect("invalid pathinfo")
.to_owned()
.into_name_and_node()
.expect("invalid pathinfo");
let url = format!( let url = format!(
"nar/tvix-castore/{}?narsize={}", "nar/tvix-castore/{}?narsize={}",
data_encoding::BASE64URL_NOPAD.encode(&buf), data_encoding::BASE64URL_NOPAD
.encode(&castorepb::Node::from_name_and_node("".into(), root_node).encode_to_vec()),
narinfo.nar_size, narinfo.nar_size,
); );
@ -125,19 +130,18 @@ pub async fn put(
// Lookup root node with peek, as we don't want to update the LRU list. // Lookup root node with peek, as we don't want to update the LRU list.
// We need to be careful to not hold the RwLock across the await point. // We need to be careful to not hold the RwLock across the await point.
let maybe_root_node: Option<tvix_castore::Node> = root_nodes let maybe_root_node: Option<tvix_castore::Node> =
.read() root_nodes.read().peek(&narinfo.nar_hash).cloned();
.peek(&narinfo.nar_hash)
.and_then(|v| v.try_into().ok());
match maybe_root_node { match maybe_root_node {
Some(root_node) => { Some(root_node) => {
// Set the root node from the lookup. // Set the root node from the lookup.
// We need to rename the node to the narinfo storepath basename, as // We need to rename the node to the narinfo storepath basename, as
// that's where it's stored in PathInfo. // that's where it's stored in PathInfo.
pathinfo.node = Some(castorepb::Node { pathinfo.node = Some(castorepb::Node::from_name_and_node(
node: Some((&root_node.rename(narinfo.store_path.to_string().into())).into()), narinfo.store_path.to_string().into(),
}); root_node,
));
// Persist the PathInfo. // Persist the PathInfo.
path_info_service.put(pathinfo).await.map_err(|e| { path_info_service.put(pathinfo).await.map_err(|e| {

7
tvix/store/src/bin/tvix-store.rs
View file

@ -351,9 +351,10 @@ async fn run_cli(cli: Cli) -> Result<(), Box<dyn std::error::Error + Send + Sync
// Create and upload a PathInfo pointing to the root_node, // Create and upload a PathInfo pointing to the root_node,
// annotated with information we have from the reference graph. // annotated with information we have from the reference graph.
let path_info = PathInfo { let path_info = PathInfo {
node: Some(tvix_castore::proto::Node { node: Some(tvix_castore::proto::Node::from_name_and_node(
node: Some((&root_node).into()), elem.path.to_string().into(),
}), root_node,
)),
references: Vec::from_iter( references: Vec::from_iter(
elem.references.iter().map(|e| e.digest().to_vec().into()), elem.references.iter().map(|e| e.digest().to_vec().into()),
), ),

27
tvix/store/src/import.rs
View file

@ -1,10 +1,8 @@
use bstr::ByteSlice;
use std::path::Path; use std::path::Path;
use tracing::{debug, instrument}; use tracing::{debug, instrument};
use tvix_castore::{ use tvix_castore::{
blobservice::BlobService, blobservice::BlobService, directoryservice::DirectoryService, import::fs::ingest_path, Node,
directoryservice::DirectoryService,
import::fs::ingest_path,
{NamedNode, Node},
}; };
use nix_compat::{ use nix_compat::{
@ -29,12 +27,12 @@ impl From<CAHash> for nar_info::Ca {
} }
} }
pub fn log_node(node: &Node, path: &Path) { pub fn log_node(name: &[u8], node: &Node, path: &Path) {
match node { match node {
Node::Directory(directory_node) => { Node::Directory(directory_node) => {
debug!( debug!(
path = ?path, path = ?path,
name = ?directory_node.get_name(), name = %name.as_bstr(),
digest = %directory_node.digest(), digest = %directory_node.digest(),
"import successful", "import successful",
) )
@ -42,7 +40,7 @@ pub fn log_node(node: &Node, path: &Path) {
Node::File(file_node) => { Node::File(file_node) => {
debug!( debug!(
path = ?path, path = ?path,
name = ?file_node.get_name(), name = %name.as_bstr(),
digest = %file_node.digest(), digest = %file_node.digest(),
"import successful" "import successful"
) )
@ -50,7 +48,7 @@ pub fn log_node(node: &Node, path: &Path) {
Node::Symlink(symlink_node) => { Node::Symlink(symlink_node) => {
debug!( debug!(
path = ?path, path = ?path,
name = ?symlink_node.get_name(), name = %name.as_bstr(),
target = ?symlink_node.target(), target = ?symlink_node.target(),
"import successful" "import successful"
) )
@ -84,13 +82,14 @@ pub fn derive_nar_ca_path_info(
nar_size: u64, nar_size: u64,
nar_sha256: [u8; 32], nar_sha256: [u8; 32],
ca: Option<&CAHash>, ca: Option<&CAHash>,
name: bytes::Bytes,
root_node: Node, root_node: Node,
) -> PathInfo { ) -> PathInfo {
// assemble the [crate::proto::PathInfo] object. // assemble the [crate::proto::PathInfo] object.
PathInfo { PathInfo {
node: Some(tvix_castore::proto::Node { node: Some(tvix_castore::proto::Node::from_name_and_node(
node: Some((&root_node).into()), name, root_node,
}), )),
// There's no reference scanning on path contents ingested like this. // There's no reference scanning on path contents ingested like this.
references: vec![], references: vec![],
narinfo: Some(NarInfo { narinfo: Some(NarInfo {
@ -140,14 +139,14 @@ where
) )
})?; })?;
// rename the root node to match the calculated output path. let name = bytes::Bytes::from(output_path.to_string());
let root_node = root_node.rename(output_path.to_string().into_bytes().into()); log_node(name.as_ref(), &root_node, path.as_ref());
log_node(&root_node, path.as_ref());
let path_info = derive_nar_ca_path_info( let path_info = derive_nar_ca_path_info(
nar_size, nar_size,
nar_sha256, nar_sha256,
Some(&CAHash::Nar(NixHash::Sha256(nar_sha256))), Some(&CAHash::Nar(NixHash::Sha256(nar_sha256))),
output_path.to_string().into_bytes().into(),
root_node, root_node,
); );

30
tvix/store/src/nar/import.rs
View file

@ -12,7 +12,7 @@ use tvix_castore::{
blobs::{self, ConcurrentBlobUploader}, blobs::{self, ConcurrentBlobUploader},
ingest_entries, IngestionEntry, IngestionError, ingest_entries, IngestionEntry, IngestionError,
}, },
PathBuf, {NamedNode, Node}, Node, PathBuf,
}; };
/// Ingests the contents from a [AsyncRead] providing NAR into the tvix store, /// Ingests the contents from a [AsyncRead] providing NAR into the tvix store,
@ -97,9 +97,7 @@ where
let (_, node) = try_join!(produce, consume)?; let (_, node) = try_join!(produce, consume)?;
// remove the fake "root" name again Ok(node)
debug_assert_eq!(&node.get_name()[..], b"root");
Ok(node.rename("".into()))
} }
async fn produce_nar_inner<BS>( async fn produce_nar_inner<BS>(
@ -198,13 +196,7 @@ mod test {
.expect("must parse"); .expect("must parse");
assert_eq!( assert_eq!(
Node::Symlink( Node::Symlink(SymlinkNode::new("/nix/store/somewhereelse".into(),).unwrap()),
SymlinkNode::new(
"".into(), // name must be empty
"/nix/store/somewhereelse".into(),
)
.unwrap()
),
root_node root_node
); );
} }
@ -224,15 +216,11 @@ mod test {
.expect("must parse"); .expect("must parse");
assert_eq!( assert_eq!(
Node::File( Node::File(FileNode::new(
FileNode::new(
"".into(), // name must be empty
HELLOWORLD_BLOB_DIGEST.clone(), HELLOWORLD_BLOB_DIGEST.clone(),
HELLOWORLD_BLOB_CONTENTS.len() as u64, HELLOWORLD_BLOB_CONTENTS.len() as u64,
false, false,
) )),
.unwrap()
),
root_node root_node
); );
@ -255,14 +243,10 @@ mod test {
.expect("must parse"); .expect("must parse");
assert_eq!( assert_eq!(
Node::Directory( Node::Directory(DirectoryNode::new(
DirectoryNode::new(
"".into(), // name must be empty
DIRECTORY_COMPLICATED.digest(), DIRECTORY_COMPLICATED.digest(),
DIRECTORY_COMPLICATED.size(), DIRECTORY_COMPLICATED.size(),
) )),
.unwrap()
),
root_node, root_node,
); );

6
tvix/store/src/nar/mod.rs
View file

@ -37,13 +37,13 @@ pub enum RenderError {
#[error("failure talking to a backing store client: {0}")] #[error("failure talking to a backing store client: {0}")]
StoreError(#[source] std::io::Error), StoreError(#[source] std::io::Error),
#[error("unable to find directory {}, referred from {:?}", .0, .1)] #[error("unable to find directory {0}, referred from {1:?}")]
DirectoryNotFound(B3Digest, bytes::Bytes), DirectoryNotFound(B3Digest, bytes::Bytes),
#[error("unable to find blob {}, referred from {:?}", .0, .1)] #[error("unable to find blob {0}, referred from {1:?}")]
BlobNotFound(B3Digest, bytes::Bytes), BlobNotFound(B3Digest, bytes::Bytes),
#[error("unexpected size in metadata for blob {}, referred from {:?} returned, expected {}, got {}", .0, .1, .2, .3)] #[error("unexpected size in metadata for blob {0}, referred from {1:?} returned, expected {2}, got {3}")]
UnexpectedBlobMeta(B3Digest, bytes::Bytes, u32, u32), UnexpectedBlobMeta(B3Digest, bytes::Bytes, u32, u32),
#[error("failure using the NAR writer: {0}")] #[error("failure using the NAR writer: {0}")]

33
tvix/store/src/nar/renderer.rs
View file

@ -8,11 +8,7 @@ use tokio::io::{self, AsyncWrite, BufReader};
use tonic::async_trait; use tonic::async_trait;
use tracing::{instrument, Span}; use tracing::{instrument, Span};
use tracing_indicatif::span_ext::IndicatifSpanExt; use tracing_indicatif::span_ext::IndicatifSpanExt;
use tvix_castore::{ use tvix_castore::{blobservice::BlobService, directoryservice::DirectoryService, Node};
blobservice::BlobService,
directoryservice::DirectoryService,
{NamedNode, Node},
};
pub struct SimpleRenderer<BS, DS> { pub struct SimpleRenderer<BS, DS> {
blob_service: BS, blob_service: BS,
@ -103,6 +99,7 @@ where
walk_node( walk_node(
nar_root_node, nar_root_node,
proto_root_node, proto_root_node,
b"",
blob_service, blob_service,
directory_service, directory_service,
) )
@ -115,7 +112,8 @@ where
/// This consumes the node. /// This consumes the node.
async fn walk_node<BS, DS>( async fn walk_node<BS, DS>(
nar_node: nar_writer::Node<'_, '_>, nar_node: nar_writer::Node<'_, '_>,
proto_node: &Node, castore_node: &Node,
name: &[u8],
blob_service: BS, blob_service: BS,
directory_service: DS, directory_service: DS,
) -> Result<(BS, DS), RenderError> ) -> Result<(BS, DS), RenderError>
@ -123,10 +121,10 @@ where
BS: BlobService + Send, BS: BlobService + Send,
DS: DirectoryService + Send, DS: DirectoryService + Send,
{ {
match proto_node { match castore_node {
Node::Symlink(proto_symlink_node) => { Node::Symlink(symlink_node) => {
nar_node nar_node
.symlink(proto_symlink_node.target()) .symlink(symlink_node.target())
.await .await
.map_err(RenderError::NARWriterError)?; .map_err(RenderError::NARWriterError)?;
} }
@ -154,19 +152,19 @@ where
.await .await
.map_err(RenderError::NARWriterError)?; .map_err(RenderError::NARWriterError)?;
} }
Node::Directory(proto_directory_node) => { Node::Directory(directory_node) => {
// look it up with the directory service // look it up with the directory service
match directory_service match directory_service
.get(proto_directory_node.digest()) .get(directory_node.digest())
.await .await
.map_err(|e| RenderError::StoreError(e.into()))? .map_err(|e| RenderError::StoreError(e.into()))?
{ {
// if it's None, that's an error! // if it's None, that's an error!
None => Err(RenderError::DirectoryNotFound( None => Err(RenderError::DirectoryNotFound(
proto_directory_node.digest().clone(), directory_node.digest().clone(),
proto_directory_node.get_name().clone(), bytes::Bytes::copy_from_slice(name),
))?, ))?,
Some(proto_directory) => { Some(directory) => {
// start a directory node // start a directory node
let mut nar_node_directory = nar_node let mut nar_node_directory = nar_node
.directory() .directory()
@ -180,15 +178,16 @@ where
// for each node in the directory, create a new entry with its name, // for each node in the directory, create a new entry with its name,
// and then recurse on that entry. // and then recurse on that entry.
for proto_node in proto_directory.nodes() { for (name, node) in directory.nodes() {
let child_node = nar_node_directory let child_node = nar_node_directory
.entry(proto_node.get_name()) .entry(name)
.await .await
.map_err(RenderError::NARWriterError)?; .map_err(RenderError::NARWriterError)?;
(blob_service, directory_service) = Box::pin(walk_node( (blob_service, directory_service) = Box::pin(walk_node(
child_node, child_node,
proto_node, node,
name.as_ref(),
blob_service, blob_service,
directory_service, directory_service,
)) ))

22
tvix/store/src/pathinfoservice/fs/mod.rs
View file

@ -3,7 +3,7 @@ use futures::StreamExt;
use tonic::async_trait; use tonic::async_trait;
use tvix_castore::fs::{RootNodes, TvixStoreFs}; use tvix_castore::fs::{RootNodes, TvixStoreFs};
use tvix_castore::{blobservice::BlobService, directoryservice::DirectoryService}; use tvix_castore::{blobservice::BlobService, directoryservice::DirectoryService};
use tvix_castore::{Error, Node, ValidateNodeError}; use tvix_castore::{Error, Node};
use super::PathInfoService; use super::PathInfoService;
@ -58,25 +58,31 @@ where
.get(*store_path.digest()) .get(*store_path.digest())
.await? .await?
.map(|path_info| { .map(|path_info| {
path_info let node = path_info
.node .node
.as_ref() .as_ref()
.expect("missing root node") .expect("missing root node")
.try_into() .to_owned();
.map_err(|e: ValidateNodeError| Error::StorageError(e.to_string()))
match node.into_name_and_node() {
Ok((_name, node)) => Ok(node),
Err(e) => Err(Error::StorageError(e.to_string())),
}
}) })
.transpose()?) .transpose()?)
} }
fn list(&self) -> BoxStream<Result<Node, Error>> { fn list(&self) -> BoxStream<Result<(bytes::Bytes, Node), Error>> {
Box::pin(self.0.as_ref().list().map(|result| { Box::pin(self.0.as_ref().list().map(|result| {
result.and_then(|path_info| { result.and_then(|path_info| {
path_info let node = path_info
.node .node
.as_ref() .as_ref()
.expect("missing root node") .expect("missing root node")
.try_into() .to_owned();
.map_err(|e: ValidateNodeError| Error::StorageError(e.to_string()))
node.into_name_and_node()
.map_err(|e| Error::StorageError(e.to_string()))
}) })
})) }))
} }

7
tvix/store/src/pathinfoservice/grpc.rs
View file

@ -132,9 +132,10 @@ where
let path_info = self let path_info = self
.grpc_client .grpc_client
.clone() .clone()
.calculate_nar(tvix_castore::proto::Node { .calculate_nar(tvix_castore::proto::Node::from_name_and_node(
node: Some(root_node.into()), "".into(),
}) root_node.to_owned(),
))
.await .await
.map_err(|e| Error::StorageError(e.to_string()))? .map_err(|e| Error::StorageError(e.to_string()))?
.into_inner(); .into_inner();

16
tvix/store/src/pathinfoservice/lru.rs
View file

@ -108,11 +108,17 @@ mod test {
let mut p = PATHINFO_1.clone(); let mut p = PATHINFO_1.clone();
let root_node = p.node.as_mut().unwrap(); let root_node = p.node.as_mut().unwrap();
if let castorepb::Node { node: Some(node) } = root_node { if let castorepb::Node { node: Some(node) } = root_node {
let n = node.to_owned(); match node {
*node = (&tvix_castore::Node::try_from(&n) castorepb::node::Node::Directory(n) => {
.unwrap() n.name = "11111111111111111111111111111111-dummy2".into()
.rename("11111111111111111111111111111111-dummy2".into())) }
.into(); castorepb::node::Node::File(n) => {
n.name = "11111111111111111111111111111111-dummy2".into()
}
castorepb::node::Node::Symlink(n) => {
n.name = "11111111111111111111111111111111-dummy2".into()
}
}
} else { } else {
unreachable!() unreachable!()
} }

8
tvix/store/src/pathinfoservice/nix_http.rs
View file

@ -228,10 +228,10 @@ where
} }
Ok(Some(PathInfo { Ok(Some(PathInfo {
node: Some(castorepb::Node { node: Some(castorepb::Node::from_name_and_node(
// set the name of the root node to the digest-name of the store path. narinfo.store_path.to_string().into(),
node: Some((&root_node.rename(narinfo.store_path.to_string().into())).into()), root_node,
}), )),
references: pathinfo.references, references: pathinfo.references,
narinfo: pathinfo.narinfo, narinfo: pathinfo.narinfo,
})) }))

2
tvix/store/src/proto/grpc_pathinfoservice_wrapper.rs
View file

@ -74,7 +74,7 @@ where
&self, &self,
request: Request<castorepb::Node>, request: Request<castorepb::Node>,
) -> Result<Response<proto::CalculateNarResponse>> { ) -> Result<Response<proto::CalculateNarResponse>> {
let root_node = (&request.into_inner()).try_into().map_err(|e| { let (_, root_node) = request.into_inner().into_name_and_node().map_err(|e| {
warn!(err = %e, "invalid root node"); warn!(err = %e, "invalid root node");
Status::invalid_argument("invalid root node") Status::invalid_argument("invalid root node")
})?; })?;

13
tvix/store/src/proto/mod.rs
View file

@ -9,7 +9,7 @@ use nix_compat::{
store_path::{self, StorePathRef}, store_path::{self, StorePathRef},
}; };
use thiserror::Error; use thiserror::Error;
use tvix_castore::{NamedNode, ValidateNodeError}; use tvix_castore::DirectoryError;
mod grpc_pathinfoservice_wrapper; mod grpc_pathinfoservice_wrapper;
@ -39,7 +39,7 @@ pub enum ValidatePathInfoError {
/// Node fails validation /// Node fails validation
#[error("Invalid root node: {:?}", .0.to_string())] #[error("Invalid root node: {:?}", .0.to_string())]
InvalidRootNode(ValidateNodeError), InvalidRootNode(DirectoryError),
/// Invalid node name encountered. Root nodes in PathInfos have more strict name requirements /// Invalid node name encountered. Root nodes in PathInfos have more strict name requirements
#[error("Failed to parse {} as StorePath: {1}", .0.to_str_lossy())] #[error("Failed to parse {} as StorePath: {1}", .0.to_str_lossy())]
@ -160,12 +160,13 @@ impl PathInfo {
let root_nix_path = match &self.node { let root_nix_path = match &self.node {
None => Err(ValidatePathInfoError::NoNodePresent)?, None => Err(ValidatePathInfoError::NoNodePresent)?,
Some(node) => { Some(node) => {
// TODO save result somewhere let (name, _node) = node
let node: tvix_castore::Node = node .clone()
.try_into() .into_name_and_node()
.map_err(ValidatePathInfoError::InvalidRootNode)?; .map_err(ValidatePathInfoError::InvalidRootNode)?;
// parse the name of the node itself and return // parse the name of the node itself and return
parse_node_name_root(node.get_name(), ValidatePathInfoError::InvalidNodeName)? parse_node_name_root(name.as_ref(), ValidatePathInfoError::InvalidNodeName)?
.to_owned() .to_owned()
} }
}; };

28
tvix/store/src/proto/tests/pathinfo.rs
View file

@ -6,11 +6,11 @@ use nix_compat::nixbase32;
use nix_compat::store_path::{self, StorePath, StorePathRef}; use nix_compat::store_path::{self, StorePath, StorePathRef};
use rstest::rstest; use rstest::rstest;
use tvix_castore::proto as castorepb; use tvix_castore::proto as castorepb;
use tvix_castore::ValidateNodeError; use tvix_castore::{DirectoryError, ValidateNodeError};
#[rstest] #[rstest]
#[case::no_node(None, Err(ValidatePathInfoError::NoNodePresent))] #[case::no_node(None, Err(ValidatePathInfoError::NoNodePresent))]
#[case::no_node_2(Some(castorepb::Node { node: None}), Err(ValidatePathInfoError::InvalidRootNode(ValidateNodeError::NoNodeSet)))] #[case::no_node_2(Some(castorepb::Node { node: None}), Err(ValidatePathInfoError::InvalidRootNode(DirectoryError::NoNodeSet)))]
fn validate_pathinfo( fn validate_pathinfo(
#[case] node: Option<castorepb::Node>, #[case] node: Option<castorepb::Node>,
@ -35,7 +35,7 @@ fn validate_pathinfo(
name: DUMMY_PATH.into(), name: DUMMY_PATH.into(),
digest: Bytes::new(), digest: Bytes::new(),
size: 0, size: 0,
}, Err(ValidatePathInfoError::InvalidRootNode(tvix_castore::ValidateNodeError::InvalidDigestLen(0))))] }, Err(ValidatePathInfoError::InvalidRootNode(DirectoryError::InvalidNode(DUMMY_PATH.into(), ValidateNodeError::InvalidDigestLen(0)))))]
#[case::invalid_node_name_no_storepath(castorepb::DirectoryNode { #[case::invalid_node_name_no_storepath(castorepb::DirectoryNode {
name: "invalid".into(), name: "invalid".into(),
digest: DUMMY_DIGEST.clone().into(), digest: DUMMY_DIGEST.clone().into(),
@ -74,7 +74,7 @@ fn validate_directory(
digest: Bytes::new(), digest: Bytes::new(),
..Default::default() ..Default::default()
}, },
Err(ValidatePathInfoError::InvalidRootNode(tvix_castore::ValidateNodeError::InvalidDigestLen(0))) Err(ValidatePathInfoError::InvalidRootNode(DirectoryError::InvalidNode(DUMMY_PATH.into(), ValidateNodeError::InvalidDigestLen(0))))
)] )]
#[case::invalid_node_name( #[case::invalid_node_name(
castorepb::FileNode { castorepb::FileNode {
@ -226,24 +226,28 @@ fn validate_inconsistent_narinfo_reference_name_digest() {
/// Create a node with an empty symlink target, and ensure it fails validation. /// Create a node with an empty symlink target, and ensure it fails validation.
#[test] #[test]
fn validate_symlink_empty_target_invalid() { fn validate_symlink_empty_target_invalid() {
let node = castorepb::node::Node::Symlink(castorepb::SymlinkNode { castorepb::Node {
node: Some(castorepb::node::Node::Symlink(castorepb::SymlinkNode {
name: "foo".into(), name: "foo".into(),
target: "".into(), target: "".into(),
}); })),
}
tvix_castore::Node::try_from(&node).expect_err("must fail validation"); .into_name_and_node()
.expect_err("must fail validation");
} }
/// Create a node with a symlink target including null bytes, and ensure it /// Create a node with a symlink target including null bytes, and ensure it
/// fails validation. /// fails validation.
#[test] #[test]
fn validate_symlink_target_null_byte_invalid() { fn validate_symlink_target_null_byte_invalid() {
let node = castorepb::node::Node::Symlink(castorepb::SymlinkNode { castorepb::Node {
node: Some(castorepb::node::Node::Symlink(castorepb::SymlinkNode {
name: "foo".into(), name: "foo".into(),
target: "foo\0".into(), target: "foo\0".into(),
}); })),
}
tvix_castore::Node::try_from(&node).expect_err("must fail validation"); .into_name_and_node()
.expect_err("must fail validation");
} }
/// Create a PathInfo with a correct deriver field and ensure it succeeds. /// Create a PathInfo with a correct deriver field and ensure it succeeds.

52
tvix/store/src/tests/nar_renderer.rs
View file

@ -22,9 +22,7 @@ async fn single_symlink(
write_nar( write_nar(
&mut buf, &mut buf,
&Node::Symlink( &Node::Symlink(SymlinkNode::new("/nix/store/somewhereelse".into()).unwrap()),
SymlinkNode::new("doesntmatter".into(), "/nix/store/somewhereelse".into()).unwrap(),
),
// don't put anything in the stores, as we don't actually do any requests. // don't put anything in the stores, as we don't actually do any requests.
blob_service, blob_service,
directory_service, directory_service,
@ -44,15 +42,11 @@ async fn single_file_missing_blob(
) { ) {
let e = write_nar( let e = write_nar(
sink(), sink(),
&Node::File( &Node::File(FileNode::new(
FileNode::new(
"doesntmatter".into(),
HELLOWORLD_BLOB_DIGEST.clone(), HELLOWORLD_BLOB_DIGEST.clone(),
HELLOWORLD_BLOB_CONTENTS.len() as u64, HELLOWORLD_BLOB_CONTENTS.len() as u64,
false, false,
) )),
.unwrap(),
),
// the blobservice is empty intentionally, to provoke the error. // the blobservice is empty intentionally, to provoke the error.
blob_service, blob_service,
directory_service, directory_service,
@ -92,15 +86,11 @@ async fn single_file_wrong_blob_size(
// Test with a root FileNode of a too big size // Test with a root FileNode of a too big size
let e = write_nar( let e = write_nar(
sink(), sink(),
&Node::File( &Node::File(FileNode::new(
FileNode::new(
"doesntmatter".into(),
HELLOWORLD_BLOB_DIGEST.clone(), HELLOWORLD_BLOB_DIGEST.clone(),
42, // <- note the wrong size here! 42, // <- note the wrong size here!
false, false,
) )),
.unwrap(),
),
blob_service.clone(), blob_service.clone(),
directory_service.clone(), directory_service.clone(),
) )
@ -117,15 +107,11 @@ async fn single_file_wrong_blob_size(
// Test with a root FileNode of a too small size // Test with a root FileNode of a too small size
let e = write_nar( let e = write_nar(
sink(), sink(),
&Node::File( &Node::File(FileNode::new(
FileNode::new(
"doesntmatter".into(),
HELLOWORLD_BLOB_DIGEST.clone(), HELLOWORLD_BLOB_DIGEST.clone(),
2, // <- note the wrong size here! 2, // <- note the wrong size here!
false, false,
) )),
.unwrap(),
),
blob_service, blob_service,
directory_service, directory_service,
) )
@ -161,15 +147,11 @@ async fn single_file(
write_nar( write_nar(
&mut buf, &mut buf,
&Node::File( &Node::File(FileNode::new(
FileNode::new(
"doesntmatter".into(),
HELLOWORLD_BLOB_DIGEST.clone(), HELLOWORLD_BLOB_DIGEST.clone(),
HELLOWORLD_BLOB_CONTENTS.len() as u64, HELLOWORLD_BLOB_CONTENTS.len() as u64,
false, false,
) )),
.unwrap(),
),
blob_service, blob_service,
directory_service, directory_service,
) )
@ -207,14 +189,10 @@ async fn test_complicated(
write_nar( write_nar(
&mut buf, &mut buf,
&Node::Directory( &Node::Directory(DirectoryNode::new(
DirectoryNode::new(
"doesntmatter".into(),
DIRECTORY_COMPLICATED.digest(), DIRECTORY_COMPLICATED.digest(),
DIRECTORY_COMPLICATED.size(), DIRECTORY_COMPLICATED.size(),
) )),
.unwrap(),
),
blob_service.clone(), blob_service.clone(),
directory_service.clone(), directory_service.clone(),
) )
@ -225,14 +203,10 @@ async fn test_complicated(
// ensure calculate_nar does return the correct sha256 digest and sum. // ensure calculate_nar does return the correct sha256 digest and sum.
let (nar_size, nar_digest) = calculate_size_and_sha256( let (nar_size, nar_digest) = calculate_size_and_sha256(
&Node::Directory( &Node::Directory(DirectoryNode::new(
DirectoryNode::new(
"doesntmatter".into(),
DIRECTORY_COMPLICATED.digest(), DIRECTORY_COMPLICATED.digest(),
DIRECTORY_COMPLICATED.size(), DIRECTORY_COMPLICATED.size(),
) )),
.unwrap(),
),
blob_service, blob_service,
directory_service, directory_service,
) )