refactor(tvix/castore/import): move process_entry to the end of the file
This makes it easier to understand the code. Change-Id: I0a9047433000551a6ba1f50a8c5c93527bc86216 Reviewed-on: https://cl.tvl.fyi/c/depot/+/11413 Tested-by: BuildkiteCI Autosubmit: flokli <flokli@flokli.de> Reviewed-by: Connor Brewster <cbrewster@hey.com>
This commit is contained in:
Florian Klink
clbot
parent
c19bc95b8a
commit
d47bd4f4bc
1 changed files with 95 additions and 95 deletions
|
|
@ -66,101 +66,6 @@ impl From<Error> for std::io::Error {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// This processes a given [walkdir::DirEntry] and returns a
|
|
||||||
/// proto::node::Node, depending on the type of the entry.
|
|
||||||
///
|
|
||||||
/// If the entry is a file, its contents are uploaded.
|
|
||||||
/// If the entry is a directory, the Directory is uploaded as well.
|
|
||||||
/// For this to work, it relies on the caller to provide the directory object
|
|
||||||
/// with the previously returned (child) nodes.
|
|
||||||
///
|
|
||||||
/// It assumes to be called only if all children of it have already been processed. If the entry is
|
|
||||||
/// indeed a directory, it'll also upload that directory to the store. For this, the
|
|
||||||
/// so-far-assembled Directory object for this path needs to be passed in.
|
|
||||||
///
|
|
||||||
/// It assumes the caller adds returned nodes to the directories it assembles.
|
|
||||||
#[instrument(skip_all, fields(entry.file_type=?&entry.file_type(),entry.path=?entry.path()))]
|
|
||||||
async fn process_entry<'a, BS>(
|
|
||||||
blob_service: BS,
|
|
||||||
directory_putter: &'a mut Box<dyn DirectoryPutter>,
|
|
||||||
entry: &'a walkdir::DirEntry,
|
|
||||||
maybe_directory: Option<Directory>,
|
|
||||||
) -> Result<Node, Error>
|
|
||||||
where
|
|
||||||
BS: AsRef<dyn BlobService> + Clone,
|
|
||||||
{
|
|
||||||
let file_type = entry.file_type();
|
|
||||||
|
|
||||||
if file_type.is_dir() {
|
|
||||||
let directory = maybe_directory
|
|
||||||
.expect("tvix bug: must be called with some directory in the case of directory");
|
|
||||||
let directory_digest = directory.digest();
|
|
||||||
let directory_size = directory.size();
|
|
||||||
|
|
||||||
// upload this directory
|
|
||||||
directory_putter
|
|
||||||
.put(directory)
|
|
||||||
.await
|
|
||||||
.map_err(|e| Error::UploadDirectoryError(entry.path().to_path_buf(), e))?;
|
|
||||||
|
|
||||||
return Ok(Node::Directory(DirectoryNode {
|
|
||||||
name: entry.file_name().as_bytes().to_owned().into(),
|
|
||||||
digest: directory_digest.into(),
|
|
||||||
size: directory_size,
|
|
||||||
}));
|
|
||||||
}
|
|
||||||
|
|
||||||
if file_type.is_symlink() {
|
|
||||||
let target: bytes::Bytes = std::fs::read_link(entry.path())
|
|
||||||
.map_err(|e| Error::UnableToStat(entry.path().to_path_buf(), e))?
|
|
||||||
.as_os_str()
|
|
||||||
.as_bytes()
|
|
||||||
.to_owned()
|
|
||||||
.into();
|
|
||||||
|
|
||||||
return Ok(Node::Symlink(SymlinkNode {
|
|
||||||
name: entry.file_name().as_bytes().to_owned().into(),
|
|
||||||
target,
|
|
||||||
}));
|
|
||||||
}
|
|
||||||
|
|
||||||
if file_type.is_file() {
|
|
||||||
let metadata = entry
|
|
||||||
.metadata()
|
|
||||||
.map_err(|e| Error::UnableToStat(entry.path().to_path_buf(), e.into()))?;
|
|
||||||
|
|
||||||
let mut file = tokio::fs::File::open(entry.path())
|
|
||||||
.await
|
|
||||||
.map_err(|e| Error::UnableToOpen(entry.path().to_path_buf(), e))?;
|
|
||||||
|
|
||||||
let mut writer = blob_service.as_ref().open_write().await;
|
|
||||||
|
|
||||||
if let Err(e) = tokio::io::copy(&mut file, &mut writer).await {
|
|
||||||
return Err(Error::UnableToRead(entry.path().to_path_buf(), e));
|
|
||||||
};
|
|
||||||
|
|
||||||
let digest = writer
|
|
||||||
.close()
|
|
||||||
.await
|
|
||||||
.map_err(|e| Error::UnableToRead(entry.path().to_path_buf(), e))?;
|
|
||||||
|
|
||||||
return Ok(Node::File(FileNode {
|
|
||||||
name: entry.file_name().as_bytes().to_vec().into(),
|
|
||||||
digest: digest.into(),
|
|
||||||
size: metadata.len(),
|
|
||||||
// If it's executable by the user, it'll become executable.
|
|
||||||
// This matches nix's dump() function behaviour.
|
|
||||||
executable: metadata.permissions().mode() & 64 != 0,
|
|
||||||
}));
|
|
||||||
}
|
|
||||||
|
|
||||||
// Nix says things like: error: file '/home/raito/dev/code.tvl.fyi/tvix/glue/src/tests/import_fixtures/a_devnode' has an unsupported type
|
|
||||||
Err(Error::UnsupportedFileType(
|
|
||||||
entry.path().to_path_buf(),
|
|
||||||
file_type,
|
|
||||||
))
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Walk the filesystem at a given path and returns a level-keyed list of directory entries.
|
/// Walk the filesystem at a given path and returns a level-keyed list of directory entries.
|
||||||
///
|
///
|
||||||
/// This is how [`ingest_path`] assembles the set of entries to pass on [`ingest_entries`].
|
/// This is how [`ingest_path`] assembles the set of entries to pass on [`ingest_entries`].
|
||||||
|
|
@ -377,3 +282,98 @@ where
|
||||||
// unreachable, we already bailed out before if root doesn't exist.
|
// unreachable, we already bailed out before if root doesn't exist.
|
||||||
unreachable!("Tvix bug: no root node emitted during ingestion")
|
unreachable!("Tvix bug: no root node emitted during ingestion")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// This processes a given [walkdir::DirEntry] and returns a
|
||||||
|
/// proto::node::Node, depending on the type of the entry.
|
||||||
|
///
|
||||||
|
/// If the entry is a file, its contents are uploaded.
|
||||||
|
/// If the entry is a directory, the Directory is uploaded as well.
|
||||||
|
/// For this to work, it relies on the caller to provide the directory object
|
||||||
|
/// with the previously returned (child) nodes.
|
||||||
|
///
|
||||||
|
/// It assumes to be called only if all children of it have already been processed. If the entry is
|
||||||
|
/// indeed a directory, it'll also upload that directory to the store. For this, the
|
||||||
|
/// so-far-assembled Directory object for this path needs to be passed in.
|
||||||
|
///
|
||||||
|
/// It assumes the caller adds returned nodes to the directories it assembles.
|
||||||
|
#[instrument(skip_all, fields(entry.file_type=?&entry.file_type(),entry.path=?entry.path()))]
|
||||||
|
async fn process_entry<'a, BS>(
|
||||||
|
blob_service: BS,
|
||||||
|
directory_putter: &'a mut Box<dyn DirectoryPutter>,
|
||||||
|
entry: &'a walkdir::DirEntry,
|
||||||
|
maybe_directory: Option<Directory>,
|
||||||
|
) -> Result<Node, Error>
|
||||||
|
where
|
||||||
|
BS: AsRef<dyn BlobService> + Clone,
|
||||||
|
{
|
||||||
|
let file_type = entry.file_type();
|
||||||
|
|
||||||
|
if file_type.is_dir() {
|
||||||
|
let directory = maybe_directory
|
||||||
|
.expect("tvix bug: must be called with some directory in the case of directory");
|
||||||
|
let directory_digest = directory.digest();
|
||||||
|
let directory_size = directory.size();
|
||||||
|
|
||||||
|
// upload this directory
|
||||||
|
directory_putter
|
||||||
|
.put(directory)
|
||||||
|
.await
|
||||||
|
.map_err(|e| Error::UploadDirectoryError(entry.path().to_path_buf(), e))?;
|
||||||
|
|
||||||
|
return Ok(Node::Directory(DirectoryNode {
|
||||||
|
name: entry.file_name().as_bytes().to_owned().into(),
|
||||||
|
digest: directory_digest.into(),
|
||||||
|
size: directory_size,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
if file_type.is_symlink() {
|
||||||
|
let target: bytes::Bytes = std::fs::read_link(entry.path())
|
||||||
|
.map_err(|e| Error::UnableToStat(entry.path().to_path_buf(), e))?
|
||||||
|
.as_os_str()
|
||||||
|
.as_bytes()
|
||||||
|
.to_owned()
|
||||||
|
.into();
|
||||||
|
|
||||||
|
return Ok(Node::Symlink(SymlinkNode {
|
||||||
|
name: entry.file_name().as_bytes().to_owned().into(),
|
||||||
|
target,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
if file_type.is_file() {
|
||||||
|
let metadata = entry
|
||||||
|
.metadata()
|
||||||
|
.map_err(|e| Error::UnableToStat(entry.path().to_path_buf(), e.into()))?;
|
||||||
|
|
||||||
|
let mut file = tokio::fs::File::open(entry.path())
|
||||||
|
.await
|
||||||
|
.map_err(|e| Error::UnableToOpen(entry.path().to_path_buf(), e))?;
|
||||||
|
|
||||||
|
let mut writer = blob_service.as_ref().open_write().await;
|
||||||
|
|
||||||
|
if let Err(e) = tokio::io::copy(&mut file, &mut writer).await {
|
||||||
|
return Err(Error::UnableToRead(entry.path().to_path_buf(), e));
|
||||||
|
};
|
||||||
|
|
||||||
|
let digest = writer
|
||||||
|
.close()
|
||||||
|
.await
|
||||||
|
.map_err(|e| Error::UnableToRead(entry.path().to_path_buf(), e))?;
|
||||||
|
|
||||||
|
return Ok(Node::File(FileNode {
|
||||||
|
name: entry.file_name().as_bytes().to_vec().into(),
|
||||||
|
digest: digest.into(),
|
||||||
|
size: metadata.len(),
|
||||||
|
// If it's executable by the user, it'll become executable.
|
||||||
|
// This matches nix's dump() function behaviour.
|
||||||
|
executable: metadata.permissions().mode() & 64 != 0,
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
// Nix says things like: error: file '/home/raito/dev/code.tvl.fyi/tvix/glue/src/tests/import_fixtures/a_devnode' has an unsupported type
|
||||||
|
Err(Error::UnsupportedFileType(
|
||||||
|
entry.path().to_path_buf(),
|
||||||
|
file_type,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue