From 30d82efa774f72e6d33c2070b32d365121654c54 Mon Sep 17 00:00:00 2001 From: Florian Klink Date: Tue, 12 Dec 2023 15:48:01 +0200 Subject: [PATCH] refactor(tvix/castore/blobservice): use io::Result in trait For all these calls, the caller has enough context about what it did, so it should be fine to use io::Result here. We pretty much only constructed crate::Error::StorageError before anyways, so this conveys *more* information. Change-Id: I5cabb3769c9c2314bab926d34dda748fda9d3ccc Reviewed-on: https://cl.tvl.fyi/c/depot/+/10328 Reviewed-by: raitobezarius Tested-by: BuildkiteCI Autosubmit: flokli --- tvix/castore/src/blobservice/grpc.rs | 31 ++++++++++---------------- tvix/castore/src/blobservice/memory.rs | 20 +++++++++-------- tvix/castore/src/blobservice/mod.rs | 8 +++---- tvix/castore/src/blobservice/sled.rs | 15 +++++++------ tvix/castore/src/import.rs | 5 ++++- tvix/store/src/nar/mod.rs | 4 ++-- tvix/store/src/nar/renderer.rs | 25 +++++++++------------ 7 files changed, 52 insertions(+), 56 deletions(-) diff --git a/tvix/castore/src/blobservice/grpc.rs b/tvix/castore/src/blobservice/grpc.rs index 9cc4f340d..f90240d88 100644 --- a/tvix/castore/src/blobservice/grpc.rs +++ b/tvix/castore/src/blobservice/grpc.rs @@ -1,7 +1,6 @@ use super::{naive_seeker::NaiveSeeker, BlobReader, BlobService, BlobWriter}; use crate::{proto, B3Digest}; use futures::sink::SinkExt; -use futures::TryFutureExt; use std::{ collections::VecDeque, io::{self}, @@ -39,7 +38,7 @@ impl GRPCBlobService { #[async_trait] impl BlobService for GRPCBlobService { #[instrument(skip(self, digest), fields(blob.digest=%digest))] - async fn has(&self, digest: &B3Digest) -> Result { + async fn has(&self, digest: &B3Digest) -> io::Result { let mut grpc_client = self.grpc_client.clone(); let resp = grpc_client .stat(proto::StatBlobRequest { @@ -51,16 +50,13 @@ impl BlobService for GRPCBlobService { match resp { Ok(_blob_meta) => Ok(true), Err(e) if e.code() == Code::NotFound => Ok(false), - Err(e) => Err(crate::Error::StorageError(e.to_string())), + Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)), } } // On success, this returns a Ok(Some(io::Read)), which can be used to read // the contents of the Blob, identified by the digest. - async fn open_read( - &self, - digest: &B3Digest, - ) -> Result>, crate::Error> { + async fn open_read(&self, digest: &B3Digest) -> io::Result>> { // Get a stream of [proto::BlobChunk], or return an error if the blob // doesn't exist. let resp = self @@ -90,7 +86,7 @@ impl BlobService for GRPCBlobService { Ok(Some(Box::new(NaiveSeeker::new(data_reader)))) } Err(e) if e.code() == Code::NotFound => Ok(None), - Err(e) => Err(crate::Error::StorageError(e.to_string())), + Err(e) => Err(io::Error::new(io::ErrorKind::Other, e)), } } @@ -141,25 +137,20 @@ pub struct GRPCBlobWriter { #[async_trait] impl BlobWriter for GRPCBlobWriter { - async fn close(&mut self) -> Result { + async fn close(&mut self) -> io::Result { if self.task_and_writer.is_none() { // if we're already closed, return the b3 digest, which must exist. // If it doesn't, we already closed and failed once, and didn't handle the error. match &self.digest { Some(digest) => Ok(digest.clone()), - None => Err(crate::Error::StorageError( - "previously closed with error".to_string(), - )), + None => Err(io::Error::new(io::ErrorKind::BrokenPipe, "already closed")), } } else { let (task, mut writer) = self.task_and_writer.take().unwrap(); // invoke shutdown, so the inner writer closes its internal tx side of // the channel. - writer - .shutdown() - .map_err(|e| crate::Error::StorageError(e.to_string())) - .await?; + writer.shutdown().await?; // block on the RPC call to return. // This ensures all chunks are sent out, and have been received by the @@ -168,15 +159,17 @@ impl BlobWriter for GR match task.await? { Ok(resp) => { // return the digest from the response, and store it in self.digest for subsequent closes. + let digest_len = resp.digest.len(); let digest: B3Digest = resp.digest.try_into().map_err(|_| { - crate::Error::StorageError( - "invalid root digest length in response".to_string(), + io::Error::new( + io::ErrorKind::Other, + format!("invalid root digest length {} in response", digest_len), ) })?; self.digest = Some(digest.clone()); Ok(digest) } - Err(e) => Err(crate::Error::StorageError(e.to_string())), + Err(e) => Err(io::Error::new(io::ErrorKind::Other, e.to_string())), } } } diff --git a/tvix/castore/src/blobservice/memory.rs b/tvix/castore/src/blobservice/memory.rs index 0954f0adb..25eec334d 100644 --- a/tvix/castore/src/blobservice/memory.rs +++ b/tvix/castore/src/blobservice/memory.rs @@ -8,7 +8,7 @@ use tonic::async_trait; use tracing::instrument; use super::{BlobReader, BlobService, BlobWriter}; -use crate::{B3Digest, Error}; +use crate::B3Digest; #[derive(Clone, Default)] pub struct MemoryBlobService { @@ -18,13 +18,13 @@ pub struct MemoryBlobService { #[async_trait] impl BlobService for MemoryBlobService { #[instrument(skip_all, ret, err, fields(blob.digest=%digest))] - async fn has(&self, digest: &B3Digest) -> Result { + async fn has(&self, digest: &B3Digest) -> io::Result { let db = self.db.read().unwrap(); Ok(db.contains_key(digest)) } #[instrument(skip_all, err, fields(blob.digest=%digest))] - async fn open_read(&self, digest: &B3Digest) -> Result>, Error> { + async fn open_read(&self, digest: &B3Digest) -> io::Result>> { let db = self.db.read().unwrap(); match db.get(digest).map(|x| Cursor::new(x.clone())) { @@ -100,13 +100,11 @@ impl tokio::io::AsyncWrite for MemoryBlobWriter { #[async_trait] impl BlobWriter for MemoryBlobWriter { - async fn close(&mut self) -> Result { + async fn close(&mut self) -> io::Result { if self.writers.is_none() { match &self.digest { Some(digest) => Ok(digest.clone()), - None => Err(crate::Error::StorageError( - "previously closed with error".to_string(), - )), + None => Err(io::Error::new(io::ErrorKind::BrokenPipe, "already closed")), } } else { let (buf, hasher) = self.writers.take().unwrap(); @@ -115,13 +113,17 @@ impl BlobWriter for MemoryBlobWriter { let digest: B3Digest = hasher.finalize().as_bytes().into(); // Only insert if the blob doesn't already exist. - let db = self.db.read()?; + let db = self.db.read().map_err(|e| { + io::Error::new(io::ErrorKind::BrokenPipe, format!("RwLock poisoned: {}", e)) + })?; if !db.contains_key(&digest) { // drop the read lock, so we can open for writing. drop(db); // open the database for writing. - let mut db = self.db.write()?; + let mut db = self.db.write().map_err(|e| { + io::Error::new(io::ErrorKind::BrokenPipe, format!("RwLock poisoned: {}", e)) + })?; // and put buf in there. This will move buf out. db.insert(digest.clone(), buf); diff --git a/tvix/castore/src/blobservice/mod.rs b/tvix/castore/src/blobservice/mod.rs index 61d5583b9..faaf94f03 100644 --- a/tvix/castore/src/blobservice/mod.rs +++ b/tvix/castore/src/blobservice/mod.rs @@ -1,7 +1,7 @@ use std::io; use tonic::async_trait; -use crate::{B3Digest, Error}; +use crate::B3Digest; mod from_addr; mod grpc; @@ -25,10 +25,10 @@ pub use self::sled::SledBlobService; #[async_trait] pub trait BlobService: Send + Sync { /// Check if the service has the blob, by its content hash. - async fn has(&self, digest: &B3Digest) -> Result; + async fn has(&self, digest: &B3Digest) -> io::Result; /// Request a blob from the store, by its content hash. - async fn open_read(&self, digest: &B3Digest) -> Result>, Error>; + async fn open_read(&self, digest: &B3Digest) -> io::Result>>; /// Insert a new blob into the store. Returns a [BlobWriter], which /// implements [io::Write] and a [BlobWriter::close]. @@ -43,7 +43,7 @@ pub trait BlobWriter: tokio::io::AsyncWrite + Send + Sync + Unpin + 'static { /// contents written. /// /// Closing a already-closed BlobWriter is a no-op. - async fn close(&mut self) -> Result; + async fn close(&mut self) -> io::Result; } /// A [tokio::io::AsyncRead] that also allows seeking. diff --git a/tvix/castore/src/blobservice/sled.rs b/tvix/castore/src/blobservice/sled.rs index f7bf33e8c..3dd4bff7b 100644 --- a/tvix/castore/src/blobservice/sled.rs +++ b/tvix/castore/src/blobservice/sled.rs @@ -34,19 +34,19 @@ impl SledBlobService { #[async_trait] impl BlobService for SledBlobService { #[instrument(skip(self), fields(blob.digest=%digest))] - async fn has(&self, digest: &B3Digest) -> Result { + async fn has(&self, digest: &B3Digest) -> io::Result { match self.db.contains_key(digest.as_slice()) { Ok(has) => Ok(has), - Err(e) => Err(Error::StorageError(e.to_string())), + Err(e) => Err(io::Error::new(io::ErrorKind::Other, e.to_string())), } } #[instrument(skip(self), fields(blob.digest=%digest))] - async fn open_read(&self, digest: &B3Digest) -> Result>, Error> { + async fn open_read(&self, digest: &B3Digest) -> io::Result>> { match self.db.get(digest.as_slice()) { Ok(None) => Ok(None), Ok(Some(data)) => Ok(Some(Box::new(Cursor::new(data[..].to_vec())))), - Err(e) => Err(Error::StorageError(e.to_string())), + Err(e) => Err(io::Error::new(io::ErrorKind::Other, e.to_string())), } } @@ -118,12 +118,13 @@ impl tokio::io::AsyncWrite for SledBlobWriter { #[async_trait] impl BlobWriter for SledBlobWriter { - async fn close(&mut self) -> Result { + async fn close(&mut self) -> io::Result { if self.writers.is_none() { match &self.digest { Some(digest) => Ok(digest.clone()), - None => Err(crate::Error::StorageError( - "previously closed with error".to_string(), + None => Err(io::Error::new( + io::ErrorKind::NotConnected, + "already closed", )), } } else { diff --git a/tvix/castore/src/import.rs b/tvix/castore/src/import.rs index 675dd0ec8..a31bb22a6 100644 --- a/tvix/castore/src/import.rs +++ b/tvix/castore/src/import.rs @@ -117,7 +117,10 @@ async fn process_entry<'a>( return Err(Error::UnableToRead(entry.path().to_path_buf(), e)); }; - let digest = writer.close().await?; + 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(), diff --git a/tvix/store/src/nar/mod.rs b/tvix/store/src/nar/mod.rs index 25a9e1882..c1a7fc2a9 100644 --- a/tvix/store/src/nar/mod.rs +++ b/tvix/store/src/nar/mod.rs @@ -1,5 +1,5 @@ use data_encoding::BASE64; -use tvix_castore::{B3Digest, Error}; +use tvix_castore::B3Digest; mod import; mod renderer; @@ -11,7 +11,7 @@ pub use renderer::write_nar; #[derive(Debug, thiserror::Error)] pub enum RenderError { #[error("failure talking to a backing store client: {0}")] - StoreError(Error), + StoreError(#[source] std::io::Error), #[error("unable to find directory {}, referred from {:?}", .0, .1)] DirectoryNotFound(B3Digest, bytes::Bytes), diff --git a/tvix/store/src/nar/renderer.rs b/tvix/store/src/nar/renderer.rs index 2904f912f..f510a9c76 100644 --- a/tvix/store/src/nar/renderer.rs +++ b/tvix/store/src/nar/renderer.rs @@ -10,12 +10,10 @@ use std::{ }; use tokio::io::{self, AsyncWrite, BufReader}; use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; -use tracing::warn; use tvix_castore::{ blobservice::BlobService, directoryservice::DirectoryService, proto::{self as castorepb, NamedNode}, - Error, }; /// Invoke [write_nar], and return the size and sha256 digest of the produced @@ -110,14 +108,11 @@ async fn walk_node( .map_err(RenderError::NARWriterError)?; } castorepb::node::Node::File(proto_file_node) => { - let digest = proto_file_node.digest.clone().try_into().map_err(|_e| { - warn!( - file_node = ?proto_file_node, - "invalid digest length in file node", - ); - - RenderError::StoreError(Error::StorageError( - "invalid digest len in file node".to_string(), + let digest_len = proto_file_node.digest.len(); + let digest = proto_file_node.digest.clone().try_into().map_err(|_| { + RenderError::StoreError(io::Error::new( + io::ErrorKind::Other, + format!("invalid digest len {} in file node", digest_len), )) })?; @@ -143,13 +138,15 @@ async fn walk_node( .map_err(RenderError::NARWriterError)?; } castorepb::node::Node::Directory(proto_directory_node) => { + let digest_len = proto_directory_node.digest.len(); let digest = proto_directory_node .digest .clone() .try_into() - .map_err(|_e| { - RenderError::StoreError(Error::StorageError( - "invalid digest len in directory node".to_string(), + .map_err(|_| { + RenderError::StoreError(io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid digest len {} in directory node", digest_len), )) })?; @@ -157,7 +154,7 @@ async fn walk_node( match directory_service .get(&digest) .await - .map_err(RenderError::StoreError)? + .map_err(|e| RenderError::StoreError(e.into()))? { // if it's None, that's an error! None => {