Create an error type specific to payload operations

This error type is Clone and can be used in the upcoming change to
open_payload that can return a cached result.

Signed-off-by: J Robert Ray <[email protected]>

Author: jrray

crates/spfs-vfs/src/winfsp/mount.rs

@@ -311,7 +311,7 @@ impl winfsp::filesystem::FileSystemContext for Mount {
                             // TODO: are there attribute flags to identify this as a non-seekable file?
                             return;
                         }
-                        Err(spfs::Error::UnknownObject(_)) => continue,
+                        Err(spfs::PayloadError::UnknownPayload(_)) => continue,
                         Err(err) => err!(send, err),
                     },
                 }

crates/spfs/src/clean.rs

@@ -22,7 +22,7 @@ use crate::prelude::*;
 use crate::runtime::makedirs_with_perms;
 use crate::storage::fs::FsRepositoryOps;
 use crate::storage::{TagNamespace, TagNamespaceBuf};
-use crate::{Digest, Error, Result, encoding, graph, storage, tracking};
+use crate::{Digest, Error, PayloadError, Result, encoding, graph, storage, tracking};
 
 #[cfg(test)]
 #[path = "./clean_test.rs"]
@@ -544,6 +544,11 @@ where
     /// objects has completed successfully and with no errors. Otherwise, it may
     /// remove data that is still being used
     async unsafe fn remove_unvisited_objects_and_payloads(&self) -> Result<CleanResult> {
+        enum ErrorFlavor {
+            Spfs(Error),
+            Payload(PayloadError),
+        }
+
         let mut result = CleanResult::default();
         let mut stream = self
             .repo
@@ -571,17 +576,19 @@ where
                     graph::DatabaseItem::Object(digest, _flat_object) => self
                         .repo
                         .remove_object_if_older_than(self.must_be_older_than, *digest)
+                        .map_err(ErrorFlavor::Spfs)
                         .boxed(),
                     graph::DatabaseItem::Payload(digest) => self
                         .repo
                         .remove_payload_if_older_than(self.must_be_older_than, *digest)
+                        .map_err(ErrorFlavor::Payload)
                         .boxed(),
                 })
-                .map(|res| {
-                    if let Err(Error::UnknownObject(_)) = res {
-                        return Ok(true);
-                    }
-                    res
+                .map(|res| match res {
+                    Err(ErrorFlavor::Spfs(Error::UnknownObject(_))) => Ok(true),
+                    Err(ErrorFlavor::Spfs(err)) => Err(err),
+                    Err(ErrorFlavor::Payload(err)) => Err(Error::String(err.to_string())),
+                    Ok(removed) => Ok(removed),
                 })
                 .map_ok(move |removed| (item, removed))
                 .boxed();

crates/spfs/src/clean_test.rs

@@ -13,7 +13,7 @@ use tokio::time::sleep;
 use super::{Cleaner, TracingCleanReporter};
 use crate::encoding::prelude::*;
 use crate::fixtures::*;
-use crate::{Error, storage, tracking};
+use crate::{Error, PayloadError, storage, tracking};
 
 #[rstest]
 #[tokio::test]
@@ -150,7 +150,7 @@ async fn test_clean_untagged_objects(#[future] tmprepo: TempRepo, tmpdir: tempfi
             continue;
         }
         let res = tmprepo.open_payload(node.entry.object).await;
-        if let Err(Error::UnknownObject(_)) = res {
+        if let Err(PayloadError::UnknownPayload(_)) = res {
             continue;
         }
         if let Err(err) = res {
@@ -264,7 +264,7 @@ async fn test_clean_on_repo_with_tag_namespace(
             continue;
         }
         let res = namespaced_tmprepo.open_payload(node.entry.object).await;
-        if let Err(Error::UnknownObject(_)) = res {
+        if let Err(PayloadError::UnknownPayload(_)) = res {
             continue;
         }
         if let Err(err) = res {
@@ -378,7 +378,7 @@ async fn test_clean_on_repo_with_tag_namespace_set(
             continue;
         }
         let res = namespaced_tmprepo.open_payload(node.entry.object).await;
-        if let Err(Error::UnknownObject(_)) = res {
+        if let Err(PayloadError::UnknownPayload(_)) = res {
             continue;
         }
         if let Err(err) = res {

crates/spfs/src/commit.rs

@@ -53,7 +53,10 @@ pub struct WriteToRepositoryBlobHasher<'repo> {
 #[tonic::async_trait]
 impl BlobHasher for WriteToRepositoryBlobHasher<'_> {
     async fn hash_blob(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<encoding::Digest> {
-        self.repo.commit_payload(reader).await
+        self.repo
+            .commit_payload(reader)
+            .await
+            .map_err(|err| Error::String(format!("commit_payload failed: {err}")))
     }
 }
 
@@ -297,7 +300,7 @@ where
                             .into_bytes();
                         let reader =
                             Box::pin(tokio::io::BufReader::new(std::io::Cursor::new(content)));
-                        self.repo.commit_payload(reader).await?
+                        self.repo.commit_payload(reader).await.map_err(|err| Error::String(format!("commit_payload failed: {err}")))?
                     } else {
                         let file = tokio::fs::File::open(&local_path).await.map_err(|err| {
                             // TODO: add better message for file missing
@@ -308,7 +311,7 @@ where
                             )
                         })?;
                         let reader = Box::pin(tokio::io::BufReader::new(file));
-                        self.repo.commit_payload(reader).await?
+                        self.repo.commit_payload(reader).await.map_err(|err| Error::String(format!("commit_payload failed: {err}")))?
                     };
                     if created != entry.object {
                         return Err(Error::String(format!(

crates/spfs/src/error.rs

@@ -388,6 +388,75 @@ impl OsError for std::io::Error {
 
 pub type Result<T> = std::result::Result<T, Error>;
 
+#[derive(Diagnostic, Debug, Clone, Error)]
+#[diagnostic(
+    url(
+        "https://spkenv.dev/error_codes#{}",
+        self.code().unwrap_or_else(|| Box::new("spfs::generic"))
+    )
+)]
+pub enum PayloadError {
+    #[error("payload did not exist: {0}")]
+    UnknownPayload(encoding::Digest),
+    #[error("invalid digest: {0}")]
+    InvalidDigest(String),
+    #[error("storage read error from {0} at {1}: {2}")]
+    StorageReadError(&'static str, std::path::PathBuf, #[source] Arc<io::Error>),
+    #[error("storage write error from {0} at {1}: {2}")]
+    StorageWriteError(&'static str, std::path::PathBuf, #[source] Arc<io::Error>),
+    #[error("Cannot write to a repository which has been pinned in time")]
+    RepositoryIsPinned,
+    #[error("Error communicating with the server: {0:?}")]
+    Tonic(Box<tonic::Status>),
+    #[error("payload error: {0}")]
+    String(String),
+}
+
+impl PayloadError {
+    /// Return true if this error suggests the call might succeed if retried on
+    /// a secondary repository.
+    #[inline]
+    pub fn try_next_repo(&self) -> bool {
+        matches!(self, PayloadError::UnknownPayload(_))
+    }
+}
+
+impl OsError for PayloadError {
+    fn os_error(&self) -> Option<i32> {
+        match self {
+            #[cfg(unix)]
+            Self::UnknownPayload(_) => Some(libc::ENOENT),
+            #[cfg(windows)]
+            Self::UnknownPayload(_) => {
+                Some(windows::Win32::Foundation::ERROR_FILE_NOT_FOUND.0 as i32)
+            }
+            Self::StorageReadError(_, _, err) => err.os_error(),
+            Self::StorageWriteError(_, _, err) => err.os_error(),
+            _ => None,
+        }
+    }
+}
+
+impl From<String> for PayloadError {
+    fn from(err: String) -> Self {
+        Self::String(err)
+    }
+}
+
+impl From<&str> for PayloadError {
+    fn from(err: &str) -> Self {
+        Self::String(err.to_string())
+    }
+}
+
+impl From<tonic::Status> for PayloadError {
+    fn from(value: tonic::Status) -> Self {
+        PayloadError::Tonic(Box::new(value))
+    }
+}
+
+pub type PayloadResult<T> = std::result::Result<T, PayloadError>;
+
 #[derive(Diagnostic, Debug, Clone, Error)]
 #[diagnostic(
     url(
@@ -413,9 +482,9 @@ pub enum SyncError {
     #[error("manifest '{0}' could not be read: {1}")]
     ManifestReadError(Digest, Arc<Error>),
     #[error("payload '{0}' could not be read: {1}")]
-    PayloadReadError(Digest, Arc<Error>),
+    PayloadReadError(Digest, Arc<PayloadError>),
     #[error("payload '{0}' could not be written: {1}")]
-    PayloadWriteError(Digest, Arc<Error>),
+    PayloadWriteError(Digest, Arc<PayloadError>),
     #[error(
         "source repository provided payload that did not match the requested digest: wanted {0}, got {1}. wrote {2} bytes"
     )]

crates/spfs/src/lib.rs

@@ -50,7 +50,16 @@ pub use clean::Cleaner;
 pub use commit::Committer;
 pub use diff::{diff, diff_runtime_changes, runtime_active_changes};
 pub use encoding::Digest;
-pub use error::{Error, OsError, OsErrorExt, Result, SyncError, SyncResult};
+pub use error::{
+    Error,
+    OsError,
+    OsErrorExt,
+    PayloadError,
+    PayloadResult,
+    Result,
+    SyncError,
+    SyncResult,
+};
 pub use resolve::{
     RenderResult,
     compute_environment_manifest,

crates/spfs/src/proto/conversions.rs

@@ -5,7 +5,7 @@
 use std::convert::{TryFrom, TryInto};
 use std::ops::Not;
 
-use crate::{Error, Result, encoding, graph, storage, tracking};
+use crate::{Error, PayloadError, PayloadResult, Result, encoding, graph, storage, tracking};
 
 pub(crate) fn convert_to_datetime(
     source: Option<super::DateTime>,
@@ -33,6 +33,10 @@ pub fn convert_digest(source: Option<super::Digest>) -> Result<encoding::Digest>
     Ok(*convert_digest_ref(source.as_ref())?)
 }
 
+pub fn convert_payload_digest(source: Option<super::Digest>) -> PayloadResult<encoding::Digest> {
+    convert_digest(source).map_err(|err| PayloadError::InvalidDigest(err.to_string()))
+}
+
 impl TryFrom<super::Digest> for encoding::Digest {
     type Error = Error;
     fn try_from(source: super::Digest) -> Result<Self> {
@@ -180,6 +184,41 @@ impl From<super::Error> for Error {
     }
 }
 
+impl From<PayloadError> for super::PayloadError {
+    fn from(err: PayloadError) -> Self {
+        let kind = Some(match err {
+            PayloadError::UnknownPayload(digest) => {
+                super::payload_error::Kind::UnknownPayload(super::UnknownPayloadError {
+                    message: digest.to_string(),
+                })
+            }
+            PayloadError::String(message) => super::payload_error::Kind::Other(message),
+            other => super::payload_error::Kind::Other(other.to_string()),
+        });
+        Self { kind }
+    }
+}
+
+impl From<super::PayloadError> for PayloadError {
+    fn from(rpc: super::PayloadError) -> Self {
+        match rpc.kind {
+            Some(super::payload_error::Kind::UnknownPayload(rpc)) => {
+                match crate::encoding::Digest::parse(&rpc.message) {
+                    Ok(digest) => PayloadError::UnknownPayload(digest),
+                    Err(_) => PayloadError::String(
+                        "Server reported UnknownPayload but did not provide a valid digest"
+                            .to_string(),
+                    ),
+                }
+            }
+            Some(super::payload_error::Kind::Other(message)) => PayloadError::String(message),
+            None => {
+                PayloadError::String("Server did not provide a payload error message".to_string())
+            }
+        }
+    }
+}
+
 impl<T: graph::ObjectProto> From<&graph::FlatObject<T>> for super::Object {
     fn from(value: &graph::FlatObject<T>) -> Self {
         Self {

crates/spfs/src/proto/defs/database.proto

@@ -47,8 +47,9 @@ message FindDigestsResponse {
 message IterDigestsRequest{}
 message IterDigestsResponse{
     oneof result {
-        Error error = 1;
+        // 1 was used
         Digest ok = 2;
+        PayloadError error = 3;
     }
 }
 

crates/spfs/src/proto/defs/error.proto

@@ -19,6 +19,9 @@ message AmbiguousReferenceError {
 message InvalidReferenceError {
     string message = 1;
 }
+message UnknownPayloadError {
+    string message = 1;
+}
 
 message Error {
     oneof kind {
@@ -29,3 +32,10 @@ message Error {
         InvalidReferenceError InvalidReference = 5;
     }
 }
+
+message PayloadError {
+    oneof kind {
+        string other = 1;
+        UnknownPayloadError UnknownPayload = 2;
+    }
+}

crates/spfs/src/proto/defs/payload.proto

@@ -20,16 +20,17 @@ message HasPayloadResponse{
     //     Error error = 1;
     //     bool ok = 2;
     // }
-    bool exists =3;
+    bool exists = 3;
 }
 
 message PayloadSizeRequest{
     Digest digest = 1;
 }
 message PayloadSizeResponse{
     oneof result {
-        Error error = 1;
+        // 1 was used
         uint64 ok = 2;
+        PayloadError error = 3;
     }
 }
 
@@ -46,13 +47,15 @@ message WritePayloadResponse{
             uint64 size = 2;
         }
         oneof result {
-            Error error = 1;
+            // 1 was used
             UploadResult ok = 2;
+            PayloadError error = 3;
         }
     }
     oneof result {
-        Error error = 1;
+        // 1 was used
         UploadOption ok = 2;
+        PayloadError error = 3;
     }
 }
 
@@ -64,8 +67,9 @@ message OpenPayloadResponse{
         repeated string locations = 1;
     }
     oneof result {
-        Error error = 1;
+        // 1 was used
         DownloadOption ok = 2;
+        PayloadError error = 3;
     }
 }
 
@@ -74,8 +78,9 @@ message RemovePayloadRequest{
 }
 message RemovePayloadResponse{
     oneof result {
-        Error error = 1;
+        // 1 was used
         Ok ok = 2;
+        PayloadError error = 3;
     }
 }
 
@@ -85,8 +90,9 @@ message RemovePayloadIfOlderThanRequest{
 }
 message RemovePayloadIfOlderThanResponse{
     oneof result {
-        Error error = 1;
+        // 1 was used
         bool ok = 2;
+        PayloadError error = 3;
     }
 }