Remove unsafe from write_data

With blob files gone there is no longer an invariant to maintain when
writing payloads.

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

Author: jrray

crates/spfs/src/server/payload.rs

@@ -180,11 +180,7 @@ async fn handle_uncompressed_upload(
     repo: Arc<storage::RepositoryHandle>,
     reader: Pin<Box<dyn crate::tracking::BlobRead>>,
 ) -> crate::Result<hyper::Response<ResponseBody>> {
-    // Safety: it is unsafe to create a payload without its corresponding
-    // blob, but this payload http server is part of a larger repository
-    // and does not intend to be responsible for ensuring the integrity
-    // of the object graph - only the up/down of payload data
-    let result = unsafe { repo.write_data(reader).await };
+    let result = repo.write_data(reader).await;
     let (digest, size) = result.map_err(|err| {
         crate::Error::String(format!(
             "An error occurred while spawning a thread for this operation: {err:?}"

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

@@ -284,13 +284,8 @@ impl PayloadStorage for FallbackProxy {
         self.primary.iter_payload_digests()
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
-        // Safety: we are wrapping the same underlying unsafe function and
-        // so the same safety holds for our callers
-        let res = unsafe { self.primary.write_data(reader).await? };
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
+        let res = self.primary.write_data(reader).await?;
         Ok(res)
     }
 

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

@@ -28,14 +28,9 @@ impl crate::storage::PayloadStorage for MaybeOpenFsRepository {
             .boxed()
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
         let opened = self.opened().await?;
-        // Safety: we are simply deferring this function to the inner
-        // one and so the same safety rules apply to our caller
-        unsafe { opened.write_data(reader).await }
+        opened.write_data(reader).await
     }
 
     async fn open_payload(
@@ -61,10 +56,7 @@ impl crate::storage::PayloadStorage for OpenFsRepository {
         Box::pin(self.payloads.iter())
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
         self.payloads.write_data(reader).await
     }
 

crates/spfs/src/storage/handle.rs

@@ -261,13 +261,8 @@ impl PayloadStorage for RepositoryHandle {
         each_variant!(self, repo, { repo.iter_payload_digests() })
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
-        // Safety: we are wrapping the same underlying unsafe function and
-        // so the same safety holds for our callers
-        unsafe { each_variant!(self, repo, { repo.write_data(reader).await }) }
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
+        each_variant!(self, repo, { repo.write_data(reader).await })
     }
 
     async fn open_payload(
@@ -436,13 +431,8 @@ impl PayloadStorage for Arc<RepositoryHandle> {
         each_variant!(&**self, repo, { repo.iter_payload_digests() })
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
-        // Safety: we are wrapping the same underlying unsafe function and
-        // so the same safety holds for our callers
-        unsafe { each_variant!(&**self, repo, { repo.write_data(reader).await }) }
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
+        each_variant!(&**self, repo, { repo.write_data(reader).await })
     }
 
     async fn open_payload(

crates/spfs/src/storage/payload.rs

@@ -23,16 +23,7 @@ pub trait PayloadStorage: Sync + Send {
     async fn has_payload(&self, digest: encoding::Digest) -> bool;
 
     /// Store the contents of the given stream, returning its digest and size
-    ///
-    /// # Safety
-    ///
-    /// It is unsafe to write payload data without also creating a blob
-    /// to track that payload in the database. Usually, its better to
-    /// call [`super::RepositoryExt::commit_payload`] instead.
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)>;
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)>;
 
     /// Return a handle and filename to the full content of a payload.
     ///
@@ -60,13 +51,8 @@ impl<T: PayloadStorage> PayloadStorage for &T {
         PayloadStorage::iter_payload_digests(&**self)
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
-        // Safety: we are wrapping the same underlying unsafe function and
-        // so the same safety holds for our callers
-        unsafe { PayloadStorage::write_data(&**self, reader).await }
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
+        PayloadStorage::write_data(&**self, reader).await
     }
 
     async fn open_payload(

crates/spfs/src/storage/payload_test.rs

@@ -23,13 +23,10 @@ async fn test_payload_io(
     let bytes = "simple string data".as_bytes();
     let reader = Box::pin(bytes);
 
-    // Safety: we are intentionally calling this function to test it
-    let (digest, size) = unsafe {
-        tmprepo
-            .write_data(reader)
-            .await
-            .expect("failed to write payload data")
-    };
+    let (digest, size) = tmprepo
+        .write_data(reader)
+        .await
+        .expect("failed to write payload data");
     assert_eq!(size, bytes.len() as u64);
 
     let mut actual = String::new();
@@ -58,13 +55,10 @@ async fn test_payload_existence(
     let bytes = "simple string data".as_bytes();
     let reader = Box::pin(bytes);
 
-    // Safety: we are intentionally calling this unsafe function to test it
-    let (digest, size) = unsafe {
-        tmprepo
-            .write_data(reader)
-            .await
-            .expect("failed to write payload data")
-    };
+    let (digest, size) = tmprepo
+        .write_data(reader)
+        .await
+        .expect("failed to write payload data");
     assert_eq!(size, bytes.len() as u64);
 
     let actual = tmprepo.has_payload(digest).await;

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

@@ -131,17 +131,12 @@ where
         self.inner.iter_payload_digests()
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
         // payloads are stored by digest, not time, and so can still
         // be safely written to a past repository view. In practice,
         // this allows some recovery and sync operations to still function
         // on pinned repositories
-
-        // Safety: we are simply calling the same inner unsafe function
-        unsafe { self.inner.write_data(reader).await }
+        self.inner.write_data(reader).await
     }
 
     async fn open_payload(

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

@@ -220,14 +220,8 @@ impl PayloadStorage for ProxyRepository {
         self.primary.iter_payload_digests()
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
-        // Safety: we are wrapping the same underlying unsafe function and
-        // so the same safety holds for our callers
-        let res = unsafe { self.primary.write_data(reader).await? };
-        Ok(res)
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
+        self.primary.write_data(reader).await
     }
 
     async fn open_payload(

crates/spfs/src/storage/repository.rs

@@ -120,9 +120,7 @@ impl<T> Repository for T where
 pub trait RepositoryExt: super::PayloadStorage + graph::DatabaseExt {
     /// Commit the data from 'reader' as a payload in this repository
     async fn commit_payload(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<encoding::Digest> {
-        // Safety: it is unsafe to write data without also creating a blob
-        // to track that payload, which is exactly what this function is doing
-        let (digest, size) = unsafe { self.write_data(reader).await? };
+        let (digest, size) = self.write_data(reader).await?;
         let blob = graph::Blob::new(digest, size);
         self.write_object(&blob).await?;
         Ok(digest)

crates/spfs/src/storage/rpc/payload.rs

@@ -39,10 +39,7 @@ impl storage::PayloadStorage for super::RpcRepository {
         Box::pin(stream)
     }
 
-    async unsafe fn write_data(
-        &self,
-        reader: Pin<Box<dyn BlobRead>>,
-    ) -> Result<(encoding::Digest, u64)> {
+    async fn write_data(&self, reader: Pin<Box<dyn BlobRead>>) -> Result<(encoding::Digest, u64)> {
         let request = proto::WritePayloadRequest {};
         let option = self
             .payload_client