[runtime] Integrate atomic blob storage backends

runtime/src/deterministic.rs

@@ -1117,6 +1117,14 @@ impl Context {
         self.executor().auditor.clone()
     }
 
+    fn fill_random(&self, dest: &mut [u8]) {
+        let executor = self.executor();
+        executor.auditor.event(b"rand", |hasher| {
+            hasher.update(b"fill_bytes");
+        });
+        executor.rng.lock().fill_bytes(dest);
+    }
+
     /// Compute a [Sha256] digest of all storage contents.
     pub fn storage_audit(&self) -> Digest {
         self.storage.inner().inner().inner().audit()
@@ -1593,11 +1601,7 @@ impl TryRng for Context {
     }
 
     fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), Self::Error> {
-        let executor = self.executor();
-        executor.auditor.event(b"rand", |hasher| {
-            hasher.update(b"fill_bytes");
-        });
-        executor.rng.lock().fill_bytes(dest);
+        self.fill_random(dest);
         Ok(())
     }
 }
@@ -1625,6 +1629,32 @@ impl crate::Storage for Context {
     }
 }
 
+impl crate::atomic::Backend for Context {
+    type Worker = Self;
+
+    fn atomic_worker(&self) -> Self {
+        <Self as crate::Supervisor>::child(self, "atomic_storage")
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.storage.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        let mut identifier = [0; 16];
+        self.fill_random(&mut identifier);
+        identifier
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        self.storage.open_atomic_existing(partition, name).await
+    }
+}
+
 impl crate::BufferPooler for Context {
     fn network_buffer_pool(&self) -> &crate::BufferPool {
         &self.network_buffer_pool
@@ -1641,8 +1671,9 @@ mod tests {
     #[cfg(feature = "external")]
     use crate::FutureExt;
     use crate::{
-        Blob, Metrics as _, Resolver, Runner as _, Spawner as _, Storage, Strategizer,
-        Supervisor as _, WriteOptions, deterministic, reschedule,
+        AtomicBlob as _, AtomicStorage as _, BatchOperation, Blob, Metrics as _, Resolver,
+        Runner as _, Spawner as _, Storage, Strategizer, Supervisor as _, WriteOptions,
+        deterministic, reschedule,
     };
     use commonware_macros::test_traced;
     use commonware_parallel::Strategy;
@@ -1700,6 +1731,23 @@ mod tests {
         assert_ne!(state_a, state_b);
     }
 
+    #[test]
+    fn test_try_fill_bytes_is_deterministic_and_audited() {
+        fn run(seed: u64) -> ([u8; 32], String) {
+            deterministic::Runner::seeded(seed).start(|mut context| async move {
+                let before = context.auditor().state();
+                let mut bytes = [0; 32];
+                context.try_fill_bytes(&mut bytes).unwrap();
+                let after = context.auditor().state();
+                assert_ne!(after, before);
+                (bytes, after)
+            })
+        }
+
+        assert_eq!(run(7), run(7));
+        assert_ne!(run(7), run(8));
+    }
+
     #[test]
     fn test_same_seed_same_order() {
         // Generate initial outputs
@@ -1914,6 +1962,46 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_recover_escaped_atomic_blob_cannot_publish_crash_lost_bytes() {
+        let (stale, checkpoint) =
+            deterministic::Runner::default().start_and_recover(|context| async move {
+                let (blob, _) = context.open_atomic("stale", b"blob").await.unwrap();
+                blob.append(b"lost").await.unwrap();
+                blob
+            });
+
+        deterministic::Runner::from(checkpoint).start(|context| async move {
+            assert!(stale.sync().await.is_err());
+            drop(stale);
+
+            let (_, len) = context.open_atomic("stale", b"blob").await.unwrap();
+            assert_eq!(len, 0);
+        });
+    }
+
+    #[test]
+    fn test_recover_rejects_escaped_atomic_batch_participant() {
+        let (stale, checkpoint) =
+            deterministic::Runner::default().start_and_recover(|context| async move {
+                let (blob, _) = context.open_atomic("stale_batch", b"blob").await.unwrap();
+                blob.append(b"lost").await.unwrap();
+                blob
+            });
+
+        deterministic::Runner::from(checkpoint).start(|context| async move {
+            assert!(
+                context
+                    .apply(vec![BatchOperation::Publish(stale)])
+                    .await
+                    .is_err()
+            );
+
+            let (_, len) = context.open_atomic("stale_batch", b"blob").await.unwrap();
+            assert_eq!(len, 0);
+        });
+    }
+
     #[test]
     fn test_recover_snapshots_fault_configuration() {
         let (stale_config, checkpoint) =

runtime/src/storage/audited.rs

@@ -72,6 +72,52 @@ impl<S: crate::Storage> crate::Storage for Storage<S> {
     }
 }
 
+impl<S: crate::atomic::Backend> crate::atomic::Backend for Storage<S> {
+    type Worker = Storage<S::Worker>;
+
+    fn atomic_worker(&self) -> Self::Worker {
+        Storage {
+            inner: self.inner.atomic_worker(),
+            auditor: self.auditor.clone(),
+        }
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.inner.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        self.inner.new_atomic_identifier()
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        self.auditor.event(b"open_atomic_existing", |hasher| {
+            hasher.update(partition.as_bytes());
+            hasher.update(name);
+        });
+        self.inner
+            .open_atomic_existing(partition, name)
+            .await
+            .map(|opened| {
+                opened.map(|(blob, len)| {
+                    (
+                        Blob {
+                            auditor: self.auditor.clone(),
+                            inner: blob,
+                            partition: partition.into(),
+                            name: name.to_vec(),
+                        },
+                        len,
+                    )
+                })
+            })
+    }
+}
+
 #[derive(Clone)]
 pub struct Blob<B: crate::Blob> {
     auditor: Arc<Auditor>,
@@ -156,8 +202,9 @@ mod tests {
         Storage as _, WriteOptions,
         deterministic::Auditor,
         storage::{
-            audited::Storage as AuditedStorage, memory::Storage as MemStorage,
-            tests::run_storage_tests,
+            audited::Storage as AuditedStorage,
+            memory::Storage as MemStorage,
+            tests::{RecordingAtomicBackend, run_storage_tests},
         },
         telemetry::metrics::Registry,
     };
@@ -178,6 +225,35 @@ mod tests {
         run_storage_tests(storage).await;
     }
 
+    #[tokio::test]
+    async fn test_audited_atomic_backend_forwards_worker_and_identifier() {
+        let auditor = Arc::new(Auditor::default());
+        let inner = RecordingAtomicBackend::new(MemStorage::new(test_pool()));
+        let storage = AuditedStorage::new(inner, auditor);
+
+        let worker = crate::atomic::Backend::atomic_worker(&storage);
+        assert!(Arc::ptr_eq(&storage.auditor, &worker.auditor));
+        let resources = crate::atomic::Backend::atomic_resources(&storage);
+        let worker_resources = crate::atomic::Backend::atomic_resources(&worker);
+        assert!(Arc::ptr_eq(
+            &resources.exclusion,
+            &worker_resources.exclusion
+        ));
+        assert!(Arc::ptr_eq(
+            &resources.payload_budget,
+            &worker_resources.payload_budget
+        ));
+        let mut expected_identifier = [2; 16];
+        expected_identifier[8..].fill(0);
+        assert_eq!(
+            crate::atomic::Backend::new_atomic_identifier(&worker),
+            expected_identifier
+        );
+
+        assert_eq!(storage.inner().worker_calls(), 1);
+        assert_eq!(storage.inner().identifier_tags(), vec![2]);
+    }
+
     #[tokio::test]
     async fn test_audited_storage_separates_partition_and_blob_names() {
         let auditor1 = Arc::new(Auditor::default());

runtime/src/storage/faulty.rs

@@ -571,6 +571,39 @@ impl<S: crate::Storage> crate::Storage for Storage<S> {
     }
 }
 
+impl<S: crate::atomic::Backend> crate::atomic::Backend for Storage<S> {
+    type Worker = Storage<S::Worker>;
+
+    fn atomic_worker(&self) -> Self::Worker {
+        Storage {
+            inner: self.inner.atomic_worker(),
+            ctx: self.ctx.clone(),
+            pending: self.pending.clone(),
+            generations: self.generations.clone(),
+        }
+    }
+
+    fn atomic_resources(&self) -> crate::atomic::AtomicResources {
+        self.inner.atomic_resources()
+    }
+
+    fn new_atomic_identifier(&self) -> [u8; 16] {
+        self.inner.new_atomic_identifier()
+    }
+
+    async fn open_atomic_existing(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> Result<Option<(Self::Blob, u64)>, Error> {
+        if self.ctx.should_fail(Op::Open) {
+            return Err(injected_io_error().into());
+        }
+        let opened = self.inner.open_atomic_existing(partition, name).await?;
+        Ok(opened.map(|(blob, len)| (self.wrap_blob(partition, name, blob, len), len)))
+    }
+}
+
 /// A blob wrapper that injects deterministic faults based on configuration.
 #[derive(Clone)]
 pub struct Blob<B: crate::Blob> {