[runtime/storage] remove Blob::len(); add blob length to Storage::open return type (#809)

Co-authored-by: Patrick O'Grady <[email protected]>

Author: danlaine

runtime/src/deterministic.rs

@@ -255,16 +255,6 @@ impl Auditor {
         *hash = hasher.finalize().to_vec();
     }
 
-    pub(crate) fn len(&self, partition: &str, name: &[u8]) {
-        let mut hash = self.hash.lock().unwrap();
-        let mut hasher = Sha256::new();
-        hasher.update(&*hash);
-        hasher.update(b"len");
-        hasher.update(partition.as_bytes());
-        hasher.update(name);
-        *hash = hasher.finalize().to_vec();
-    }
-
     pub(crate) fn read_at(&self, partition: &str, name: &[u8], buf: usize, offset: u64) {
         let mut hash = self.hash.lock().unwrap();
         let mut hasher = Sha256::new();
@@ -1338,7 +1328,7 @@ impl CryptoRng for Context {}
 impl crate::Storage for Context {
     type Blob = <MeteredStorage<AuditedStorage<MemStorage>> as crate::Storage>::Blob;
 
-    async fn open(&self, partition: &str, name: &[u8]) -> Result<Self::Blob, Error> {
+    async fn open(&self, partition: &str, name: &[u8]) -> Result<(Self::Blob, u64), Error> {
         self.storage.open(partition, name).await
     }
 
@@ -1461,7 +1451,7 @@ mod tests {
 
         // Run some tasks, sync storage, and recover the runtime
         let (context, state) = executor1.start(|context| async move {
-            let blob = context.open(partition, name).await.unwrap();
+            let (blob, _) = context.open(partition, name).await.unwrap();
             blob.write_at(data, 0).await.unwrap();
             blob.sync().await.unwrap();
             let state = context.auditor().state();
@@ -1475,10 +1465,9 @@ mod tests {
         // Check that synced storage persists after recovery
         let executor = Runner::from(recovered_context);
         executor.start(|context| async move {
-            let blob = context.open(partition, name).await.unwrap();
-            let len = blob.len().await.unwrap();
+            let (blob, len) = context.open(partition, name).await.unwrap();
             assert_eq!(len, data.len() as u64);
-            let mut buf = vec![0; len as usize];
+            let mut buf = vec![0; data.len()];
             blob.read_at(&mut buf, 0).await.unwrap();
             assert_eq!(buf, data);
         });
@@ -1495,7 +1484,7 @@ mod tests {
         // Run some tasks without syncing storage
         let context = executor.start(|context| async move {
             let context = context.clone();
-            let blob = context.open(partition, name).await.unwrap();
+            let (blob, _) = context.open(partition, name).await.unwrap();
             blob.write_at(&data, 0).await.unwrap();
             // Intentionally do not call sync() here
             context
@@ -1507,8 +1496,7 @@ mod tests {
 
         // Check that unsynced storage does not persist after recovery
         executor.start(|context| async move {
-            let blob = context.open(partition, name).await.unwrap();
-            let len = blob.len().await.unwrap();
+            let (_, len) = context.open(partition, name).await.unwrap();
             assert_eq!(len, 0);
         });
     }

runtime/src/lib.rs

@@ -270,15 +270,16 @@ pub trait Storage: Clone + Send + Sync + 'static {
     /// The readable/writeable storage buffer that can be opened by this Storage.
     type Blob: Blob;
 
-    /// Open an existing blob in a given partition or create a new one.
+    /// Open an existing blob in a given partition or create a new one, returning
+    /// the blob and its length.
     ///
     /// Multiple instances of the same blob can be opened concurrently, however,
     /// writing to the same blob concurrently may lead to undefined behavior.
     fn open(
         &self,
         partition: &str,
         name: &[u8],
-    ) -> impl Future<Output = Result<Self::Blob, Error>> + Send;
+    ) -> impl Future<Output = Result<(Self::Blob, u64), Error>> + Send;
 
     /// Remove a blob from a given partition.
     ///
@@ -305,9 +306,6 @@ pub trait Storage: Clone + Send + Sync + 'static {
 /// and writing to both is undefined behavior.
 #[allow(clippy::len_without_is_empty)]
 pub trait Blob: Clone + Send + Sync + 'static {
-    /// Get the length of the blob.
-    fn len(&self) -> impl Future<Output = Result<u64, Error>> + Send;
-
     /// Read from the blob at the given offset.
     ///
     /// `read_at` does not return the number of bytes read because it
@@ -521,7 +519,7 @@ mod tests {
             let name = b"test_blob";
 
             // Open a new blob
-            let blob = context
+            let (blob, _) = context
                 .open(partition, name)
                 .await
                 .expect("Failed to open blob");
@@ -542,10 +540,6 @@ mod tests {
                 .expect("Failed to read from blob");
             assert_eq!(&buffer, data);
 
-            // Get blob length
-            let length = blob.len().await.expect("Failed to get blob length");
-            assert_eq!(length, data.len() as u64);
-
             // Close the blob
             blob.close().await.expect("Failed to close blob");
 
@@ -557,10 +551,11 @@ mod tests {
             assert!(blobs.contains(&name.to_vec()));
 
             // Reopen the blob
-            let blob = context
+            let (blob, len) = context
                 .open(partition, name)
                 .await
                 .expect("Failed to reopen blob");
+            assert_eq!(len, data.len() as u64);
 
             // Read data part of message back
             let mut buffer = vec![0u8; 7];
@@ -606,7 +601,7 @@ mod tests {
             let name = b"test_blob_rw";
 
             // Open a new blob
-            let blob = context
+            let (blob, _) = context
                 .open(partition, name)
                 .await
                 .expect("Failed to open blob");
@@ -621,10 +616,6 @@ mod tests {
                 .await
                 .expect("Failed to write data2");
 
-            // Assert that length tracks pending data
-            let length = blob.len().await.expect("Failed to get blob length");
-            assert_eq!(length, 10);
-
             // Read data back
             let mut buffer = vec![0u8; 10];
             blob.read_at(&mut buffer, 0)
@@ -638,13 +629,9 @@ mod tests {
             blob.write_at(data3, 5)
                 .await
                 .expect("Failed to write data3");
-            let length = blob.len().await.expect("Failed to get blob length");
-            assert_eq!(length, 10);
 
             // Truncate the blob
             blob.truncate(5).await.expect("Failed to truncate blob");
-            let length = blob.len().await.expect("Failed to get blob length");
-            assert_eq!(length, 5);
             let mut buffer = vec![0u8; 5];
             blob.read_at(&mut buffer, 0)
                 .await
@@ -654,7 +641,7 @@ mod tests {
             // Full read after truncation
             let mut buffer = vec![0u8; 10];
             let result = blob.read_at(&mut buffer, 0).await;
-            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
+            assert!(result.is_err());
 
             // Close the blob
             blob.close().await.expect("Failed to close blob");
@@ -668,17 +655,17 @@ mod tests {
         runner.start(|context| async move {
             let partitions = ["partition1", "partition2", "partition3"];
             let name = b"test_blob_rw";
+            let data1 = b"Hello";
+            let data2 = b"World";
 
             for (additional, partition) in partitions.iter().enumerate() {
                 // Open a new blob
-                let blob = context
+                let (blob, _) = context
                     .open(partition, name)
                     .await
                     .expect("Failed to open blob");
 
                 // Write data at different offsets
-                let data1 = b"Hello";
-                let data2 = b"World";
                 blob.write_at(data1, 0)
                     .await
                     .expect("Failed to write data1");
@@ -692,10 +679,11 @@ mod tests {
 
             for (additional, partition) in partitions.iter().enumerate() {
                 // Open a new blob
-                let blob = context
+                let (blob, len) = context
                     .open(partition, name)
                     .await
                     .expect("Failed to open blob");
+                assert_eq!(len, (data1.len() + data2.len() + additional) as u64);
 
                 // Read data back
                 let mut buffer = vec![0u8; 10 + additional];
@@ -720,15 +708,15 @@ mod tests {
             let name = b"test_blob_rw";
 
             // Open a new blob
-            let blob = context
+            let (blob, _) = context
                 .open(partition, name)
                 .await
                 .expect("Failed to open blob");
 
             // Read data past file length (empty file)
             let mut buffer = vec![0u8; 10];
             let result = blob.read_at(&mut buffer, 0).await;
-            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
+            assert!(result.is_err());
 
             // Write data to the blob
             let data = b"Hello, Storage!";
@@ -739,7 +727,7 @@ mod tests {
             // Read data past file length (non-empty file)
             let mut buffer = vec![0u8; 20];
             let result = blob.read_at(&mut buffer, 0).await;
-            assert!(matches!(result, Err(Error::BlobInsufficientLength)));
+            assert!(result.is_err());
         })
     }
 
@@ -752,7 +740,7 @@ mod tests {
             let name = b"test_blob_rw";
 
             // Open a new blob
-            let blob = context
+            let (blob, _) = context
                 .open(partition, name)
                 .await
                 .expect("Failed to open blob");
@@ -800,10 +788,6 @@ mod tests {
                 .expect("Failed to read from blob");
             assert_eq!(&buffer, data);
 
-            // Get blob length
-            let length = blob.len().await.expect("Failed to get blob length");
-            assert_eq!(length, data.len() as u64);
-
             // Close the blob
             blob.close().await.expect("Failed to close blob");
 

runtime/src/storage/audited.rs

@@ -16,13 +16,18 @@ impl<S: crate::Storage> Storage<S> {
 impl<S: crate::Storage> crate::Storage for Storage<S> {
     type Blob = Blob<S::Blob>;
 
-    async fn open(&self, partition: &str, name: &[u8]) -> Result<Self::Blob, Error> {
+    async fn open(&self, partition: &str, name: &[u8]) -> Result<(Self::Blob, u64), Error> {
         self.auditor.open(partition, name);
-        self.inner.open(partition, name).await.map(|blob| Blob {
-            auditor: self.auditor.clone(),
-            inner: blob,
-            partition: partition.into(),
-            name: name.to_vec(),
+        self.inner.open(partition, name).await.map(|(blob, len)| {
+            (
+                Blob {
+                    auditor: self.auditor.clone(),
+                    inner: blob,
+                    partition: partition.into(),
+                    name: name.to_vec(),
+                },
+                len,
+            )
         })
     }
 
@@ -46,11 +51,6 @@ pub struct Blob<B: crate::Blob> {
 }
 
 impl<B: crate::Blob> crate::Blob for Blob<B> {
-    async fn len(&self) -> Result<u64, Error> {
-        self.auditor.len(&self.partition, &self.name);
-        self.inner.len().await
-    }
-
     async fn read_at(&self, buf: &mut [u8], offset: u64) -> Result<(), Error> {
         self.auditor
             .read_at(&self.partition, &self.name, buf.len(), offset);
@@ -114,8 +114,8 @@ mod tests {
         let storage2 = AuditedStorage::new(inner2, auditor2.clone());
 
         // Perform a sequence of operations on both storages simultaneously
-        let blob1 = storage1.open("partition", b"test_blob").await.unwrap();
-        let blob2 = storage2.open("partition", b"test_blob").await.unwrap();
+        let (blob1, _) = storage1.open("partition", b"test_blob").await.unwrap();
+        let (blob2, _) = storage2.open("partition", b"test_blob").await.unwrap();
 
         // Write data to the blobs
         blob1.write_at(b"hello world", 0).await.unwrap();
@@ -142,10 +142,6 @@ mod tests {
         // Truncate the blobs
         blob1.truncate(5).await.unwrap();
         blob2.truncate(5).await.unwrap();
-        let len1 = blob1.len().await.unwrap();
-        let len2 = blob2.len().await.unwrap();
-        assert_eq!(len1, 5, "Blob1 length after truncation is incorrect");
-        assert_eq!(len2, 5, "Blob2 length after truncation is incorrect");
         assert_eq!(
             auditor1.state(),
             auditor2.state(),

runtime/src/storage/memory.rs

@@ -1,8 +1,7 @@
+use commonware_utils::hex;
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex, RwLock};
 
-use commonware_utils::hex;
-
 /// In-memory storage implementation for the commonware runtime.
 #[derive(Clone)]
 pub struct Storage {
@@ -20,15 +19,18 @@ impl Default for Storage {
 impl crate::Storage for Storage {
     type Blob = Blob;
 
-    async fn open(&self, partition: &str, name: &[u8]) -> Result<Self::Blob, crate::Error> {
+    async fn open(&self, partition: &str, name: &[u8]) -> Result<(Self::Blob, u64), crate::Error> {
         let mut partitions = self.partitions.lock().unwrap();
         let partition_entry = partitions.entry(partition.into()).or_default();
         let content = partition_entry.entry(name.into()).or_default();
-        Ok(Blob::new(
-            self.partitions.clone(),
-            partition.into(),
-            name,
-            content.clone(),
+        Ok((
+            Blob::new(
+                self.partitions.clone(),
+                partition.into(),
+                name,
+                content.clone(),
+            ),
+            content.len() as u64,
         ))
     }
 
@@ -92,11 +94,6 @@ impl Blob {
 }
 
 impl crate::Blob for Blob {
-    async fn len(&self) -> Result<u64, crate::Error> {
-        let content = self.content.read().unwrap();
-        Ok(content.len() as u64)
-    }
-
     async fn read_at(&self, buf: &mut [u8], offset: u64) -> Result<(), crate::Error> {
         let offset = offset
             .try_into()