Add a payload_size operation

Without a blob object to read, if something wants to know the size of a
payload it is possible with this to do it without having to read the
payload data.

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

Author: jrray

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

@@ -23,6 +23,16 @@ message HasPayloadResponse{
     bool exists =3;
 }
 
+message PayloadSizeRequest{
+    Digest digest = 1;
+}
+message PayloadSizeResponse{
+    oneof result {
+        Error error = 1;
+        uint64 ok = 2;
+    }
+}
+
 message WritePayloadRequest{}
 message WritePayloadResponse{
     message UploadOption {
@@ -72,6 +82,7 @@ message RemovePayloadResponse{
 service PayloadService {
     rpc IterDigests(IterDigestsRequest) returns (stream IterDigestsResponse);
     rpc HasPayload(HasPayloadRequest) returns (HasPayloadResponse);
+    rpc PayloadSize(PayloadSizeRequest) returns (PayloadSizeResponse);
     rpc WritePayload(WritePayloadRequest) returns (WritePayloadResponse);
     rpc OpenPayload(OpenPayloadRequest) returns (OpenPayloadResponse);
     rpc RemovePayload(RemovePayloadRequest) returns (RemovePayloadResponse);

crates/spfs/src/proto/result.rs

@@ -127,6 +127,11 @@ rpc_result!(
     bool
 );
 
+rpc_result!(
+    g::PayloadSizeResponse,
+    g::payload_size_response::Result,
+    u64
+);
 rpc_result!(
     g::WritePayloadResponse,
     g::write_payload_response::Result,

crates/spfs/src/server/payload.rs

@@ -70,6 +70,18 @@ impl proto::payload_service_server::PayloadService for PayloadService {
         Ok(Response::new(result))
     }
 
+    async fn payload_size(
+        &self,
+        request: Request<proto::PayloadSizeRequest>,
+    ) -> Result<Response<proto::PayloadSizeResponse>, Status> {
+        let request = request.into_inner();
+        let digest = convert_digest(request.digest)
+            .map_err(|err| Status::invalid_argument(err.to_string()))?;
+        let size = proto::handle_error!(self.repo.payload_size(digest).await);
+        let result = proto::PayloadSizeResponse::ok(size);
+        Ok(Response::new(result))
+    }
+
     async fn open_payload(
         &self,
         request: Request<proto::OpenPayloadRequest>,

crates/spfs/src/storage/fallback/repository.rs

@@ -280,6 +280,10 @@ impl PayloadStorage for FallbackProxy {
         false
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        crate::storage::proxy::payload_size(self, digest).await
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         self.primary.iter_payload_digests()
     }

crates/spfs/src/storage/fs/payloads.rs

@@ -21,6 +21,11 @@ impl crate::storage::PayloadStorage for MaybeOpenFsRepository {
         opened.has_payload(digest).await
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        let opened = self.opened().await?;
+        opened.payload_size(digest).await
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         self.opened()
             .and_then(|opened| ready(Ok(opened.iter_payload_digests())))
@@ -52,6 +57,17 @@ impl crate::storage::PayloadStorage for OpenFsRepository {
         tokio::fs::symlink_metadata(path).await.is_ok()
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        let path = self.payloads.build_digest_path(&digest);
+        tokio::fs::symlink_metadata(&path)
+            .await
+            .map(|meta| meta.len())
+            .map_err(|err| match err.kind() {
+                ErrorKind::NotFound => Error::UnknownObject(digest),
+                _ => Error::StorageReadError("symlink_metadata on payload", path, err),
+            })
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         Box::pin(self.payloads.iter())
     }

crates/spfs/src/storage/handle.rs

@@ -257,6 +257,10 @@ impl PayloadStorage for RepositoryHandle {
         each_variant!(self, repo, { repo.has_payload(digest).await })
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        each_variant!(self, repo, { repo.payload_size(digest).await })
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         each_variant!(self, repo, { repo.iter_payload_digests() })
     }
@@ -427,6 +431,10 @@ impl PayloadStorage for Arc<RepositoryHandle> {
         each_variant!(&**self, repo, { repo.has_payload(digest).await })
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        each_variant!(&**self, repo, { repo.payload_size(digest).await })
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         each_variant!(&**self, repo, { repo.iter_payload_digests() })
     }

crates/spfs/src/storage/payload.rs

@@ -22,6 +22,9 @@ pub trait PayloadStorage: Sync + Send {
     /// Return true if the identified payload exists.
     async fn has_payload(&self, digest: encoding::Digest) -> bool;
 
+    /// Return the payload size if the identified payload exists.
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64>;
+
     /// Store the contents of the given stream, returning its digest and size
     async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)>;
 
@@ -47,6 +50,10 @@ impl<T: PayloadStorage> PayloadStorage for &T {
         PayloadStorage::has_payload(&**self, digest).await
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        PayloadStorage::payload_size(&**self, digest).await
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         PayloadStorage::iter_payload_digests(&**self)
     }

crates/spfs/src/storage/pinned/repository.rs

@@ -127,6 +127,10 @@ where
         self.inner.has_payload(digest).await
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        self.inner.payload_size(digest).await
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         self.inner.iter_payload_digests()
     }

crates/spfs/src/storage/proxy/mod.rs

@@ -12,6 +12,7 @@ pub(crate) use repository::{
     find_tags_in_namespace,
     iter_tag_streams_in_namespace,
     ls_tags_in_namespace,
+    payload_size,
     read_tag_in_namespace,
 };
 

crates/spfs/src/storage/proxy/repository.rs

@@ -202,6 +202,21 @@ impl graph::DatabaseExt for ProxyRepository {
     }
 }
 
+pub(crate) async fn payload_size<R>(repo: R, digest: encoding::Digest) -> Result<u64>
+where
+    R: ProxyRepositoryExt,
+{
+    if let Ok(size) = repo.primary().payload_size(digest).await {
+        return Ok(size);
+    }
+    for secondary in repo.secondary().iter() {
+        if let Ok(size) = secondary.payload_size(digest).await {
+            return Ok(size);
+        }
+    }
+    Err(crate::Error::UnknownObject(digest))
+}
+
 #[async_trait::async_trait]
 impl PayloadStorage for ProxyRepository {
     async fn has_payload(&self, digest: encoding::Digest) -> bool {
@@ -216,6 +231,10 @@ impl PayloadStorage for ProxyRepository {
         false
     }
 
+    async fn payload_size(&self, digest: encoding::Digest) -> Result<u64> {
+        payload_size(self, digest).await
+    }
+
     fn iter_payload_digests(&self) -> Pin<Box<dyn Stream<Item = Result<encoding::Digest>> + Send>> {
         self.primary.iter_payload_digests()
     }