[glue] Remove Shared database ownership

examples/reshare/src/application.rs

@@ -10,8 +10,8 @@ use commonware_cryptography::{
 use commonware_glue::{
     dkg::reshare::Input as ReshareInput,
     stateful::{
-        Application, Input, Proposed,
-        db::{DatabaseSet, Merkleized as _, Unmerkleized as _},
+        Application, ExecutionError, Input, Proposed,
+        db::{DatabaseSet, Merkleized as _, MerkleizedOf, Unmerkleized as _, UnmerkleizedOf},
     },
 };
 use commonware_runtime::{BufferPooler, Clock, Metrics, Spawner, Storage};
@@ -36,13 +36,12 @@ impl App {
 
     async fn execute<E: Spawner + Metrics + Clock + Storage + BufferPooler>(
         height: Height,
-        batches: <Database<E> as DatabaseSet<E>>::Unmerkleized,
-    ) -> <Database<E> as DatabaseSet<E>>::Merkleized {
-        batches
+        batches: UnmerkleizedOf<Database<E>, E>,
+    ) -> Result<MerkleizedOf<Database<E>, E>, ExecutionError> {
+        Ok(batches
             .write(HEIGHT_KEY, Some(U64::new(height.get())))
             .merkleize()
-            .await
-            .expect("height write must merkleize")
+            .await?)
     }
 }
 
@@ -65,14 +64,16 @@ where
         &mut self,
         context: (E, Self::Context),
         mut ancestry: impl Ancestry<Self::Block>,
-        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
+        batches: UnmerkleizedOf<Self::Databases, E>,
         input: Input<Self::Input, Self::Provider>,
-    ) -> Option<Proposed<Self, E>> {
+    ) -> Result<Option<Proposed<Self, E>>, ExecutionError> {
         // The `reshare::Application` wrapper selected and fetched the payload.
         let payload = input.upstream.payload;
-        let parent = ancestry.next().await?;
+        let Some(parent) = ancestry.next().await else {
+            return Ok(None);
+        };
         let height = parent.height().next();
-        let merkleized = Self::execute(height, batches).await;
+        let merkleized = Self::execute(height, batches).await?;
         let bounds = merkleized.bounds();
         let block = Block {
             context: context.1,
@@ -82,31 +83,33 @@ where
             range: non_empty_range!(bounds.inactivity_floor, bounds.tip.size),
             payload,
         };
-        Some(Proposed { block, merkleized })
+        Ok(Some(Proposed { block, merkleized }))
     }
 
     async fn verify(
         &mut self,
         _context: (E, Self::Context),
         mut ancestry: impl Ancestry<Self::Block>,
-        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
-    ) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
+        batches: UnmerkleizedOf<Self::Databases, E>,
+    ) -> Result<Option<MerkleizedOf<Self::Databases, E>>, ExecutionError> {
         // Validation from higher layers:
         // - Epoch validation is handled by `Deferred`
         // - QMDB root / range validation is handled by `stateful::Application`
         // - Reshare `Payload` validation is handled by `reshare::Application`
 
-        let block = ancestry.next().await?;
-        let merkleized = Self::execute(block.height(), batches).await;
-        Some(merkleized)
+        let Some(block) = ancestry.next().await else {
+            return Ok(None);
+        };
+        let merkleized = Self::execute(block.height(), batches).await?;
+        Ok(Some(merkleized))
     }
 
     async fn apply(
         &mut self,
         _context: (E, Self::Context),
         block: &Self::Block,
-        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
-    ) -> <Self::Databases as DatabaseSet<E>>::Merkleized {
+        batches: UnmerkleizedOf<Self::Databases, E>,
+    ) -> Result<MerkleizedOf<Self::Databases, E>, ExecutionError> {
         Self::execute(block.height(), batches).await
     }
 

examples/reshare/src/types.rs

@@ -28,7 +28,7 @@ use commonware_cryptography::{
 use commonware_formatting::{from_hex, hex};
 use commonware_glue::{
     dkg::{self, ParticipantsProvider, Registrar as RegistrarTrait, ReshareBlock, types::Payload},
-    stateful::db::{Shared, SyncEngineConfig},
+    stateful::db::{Single, SyncEngineConfig},
 };
 use commonware_parallel::Sequential;
 use commonware_runtime::{Buf, BufMut, Quota, buffer::paged::CacheRef};
@@ -62,8 +62,8 @@ use tracing::info;
 pub type Scheme = simplex::scheme::bls12381_threshold::vrf::Scheme<ed25519::PublicKey, MinSig>;
 /// QMDB holding the application state.
 pub type Qmdb<E> = fixed::Db<mmr::Family, E, U64, U64, Sha256, TwoCap, Sequential>;
-/// Shared handle to the application QMDB.
-pub type Database<E> = Shared<Qmdb<E>>;
+/// Database set containing a single QMDB.
+pub type Database<E> = Single<Qmdb<E>>;
 /// Globally unique namespace for every message signed by this example.
 pub const NAMESPACE: &[u8] = b"_COMMONWARE_RESHARE_EXAMPLE";
 /// Number of blocks in each epoch.

glue/src/dkg/tests/reshare/harness.rs

@@ -20,11 +20,11 @@ use crate::{
         reporter::MonitorReporter,
     },
     stateful::{
-        Application, Config as StatefulConfig, Input, Proposed, Stateful as StatefulActor,
-        SyncPlan,
+        Application, Config as StatefulConfig, ExecutionError, Input, Proposed,
+        Stateful as StatefulActor, SyncPlan,
         db::{
-            DatabaseSet, Merkleized as _, Shared, SyncEngineConfig, Unmerkleized as _,
-            p2p as qmdb_resolver,
+            DatabaseSet, Merkleized as _, MerkleizedOf, Publisher, ReadersOf, Single,
+            SyncEngineConfig, Unmerkleized as _, UnmerkleizedOf, p2p as qmdb_resolver,
         },
     },
 };
@@ -97,7 +97,7 @@ use std::{
 
 type Qmdb<E> =
     fixed::Db<mmr::Family, E, sha256::Digest, sha256::Digest, Sha256, TwoCap, Sequential>;
-type Database<E> = Shared<Qmdb<E>>;
+type Database<E> = Single<Qmdb<E>>;
 type Scheme = simplex::scheme::bls12381_threshold::vrf::Scheme<ed25519::PublicKey, MinPk>;
 type MarshalVariant = Standard<Block>;
 type Marshal = MarshalMailbox<Scheme, MarshalVariant>;
@@ -362,11 +362,11 @@ struct App {
 impl App {
     async fn execute<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler>(
         height: Height,
-        mut batches: <Database<E> as DatabaseSet<E>>::Unmerkleized,
-    ) -> <Database<E> as DatabaseSet<E>>::Merkleized {
+        mut batches: UnmerkleizedOf<Database<E>, E>,
+    ) -> Result<MerkleizedOf<Database<E>, E>, ExecutionError> {
         let key = Sha256::hash(&[b"height"]);
         batches = batches.write(key, Some(u64_to_digest(height.get())));
-        batches.merkleize().await.unwrap()
+        Ok(batches.merkleize().await?)
     }
 }
 
@@ -386,14 +386,16 @@ impl<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler> Application<E>
         &mut self,
         context: (E, Self::Context),
         ancestry: impl Ancestry<Self::Block>,
-        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
+        batches: UnmerkleizedOf<Self::Databases, E>,
         input: Input<Self::Input, Self::Provider>,
-    ) -> Option<Proposed<Self, E>> {
-        let parent = ancestry.peek()?.clone();
+    ) -> Result<Option<Proposed<Self, E>>, ExecutionError> {
+        let Some(parent) = ancestry.peek().cloned() else {
+            return Ok(None);
+        };
         let height = Height::new(parent.height().get() + 1);
         // The reshare::Application wrapper selected and fetched the payload.
         let payload = input.upstream.payload;
-        let merkleized = Self::execute(height, batches).await;
+        let merkleized = Self::execute(height, batches).await?;
         let bounds = merkleized.bounds();
         let block = Block {
             context: context.1,
@@ -403,36 +405,38 @@ impl<E: Rng + Spawner + Metrics + Clock + Storage + BufferPooler> Application<E>
             range: non_empty_range!(bounds.inactivity_floor, bounds.tip.size),
             payload,
         };
-        Some(Proposed { block, merkleized })
+        Ok(Some(Proposed { block, merkleized }))
     }
 
     async fn verify(
         &mut self,
         _context: (E, Self::Context),
         ancestry: impl Ancestry<Self::Block>,
-        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
-    ) -> Option<<Self::Databases as DatabaseSet<E>>::Merkleized> {
+        batches: UnmerkleizedOf<Self::Databases, E>,
+    ) -> Result<Option<MerkleizedOf<Self::Databases, E>>, ExecutionError> {
         // Reshare final-block payload validation is enforced by the surrounding
         // reshare::Application wrapper; this inner app only executes state.
-        let tip = ancestry.peek()?.clone();
-        let merkleized = Self::execute(tip.height(), batches).await;
-        Some(merkleized)
+        let Some(tip) = ancestry.peek().cloned() else {
+            return Ok(None);
+        };
+        let merkleized = Self::execute(tip.height(), batches).await?;
+        Ok(Some(merkleized))
     }
 
     async fn apply(
         &mut self,
         _context: (E, Self::Context),
         block: &Self::Block,
-        batches: <Self::Databases as DatabaseSet<E>>::Unmerkleized,
-    ) -> <Self::Databases as DatabaseSet<E>>::Merkleized {
+        batches: UnmerkleizedOf<Self::Databases, E>,
+    ) -> Result<MerkleizedOf<Self::Databases, E>, ExecutionError> {
         Self::execute(block.height(), batches).await
     }
 
     async fn finalized(
         &mut self,
         context: (E, Self::Context),
         block: &Self::Block,
-        _readers: <Self::Databases as DatabaseSet<E>>::Readers,
+        _readers: ReadersOf<Self::Databases, E>,
     ) {
         self.processed
             .lock()
@@ -1019,8 +1023,7 @@ impl EngineDefinition for ReshareEngine {
         };
 
         let publication_context = context.child("publication");
-        let (snapshot_publisher, snapshot_subscriber) =
-            crate::stateful::db::Publisher::new(&publication_context);
+        let (snapshot_publisher, snapshot_subscriber) = Publisher::new(&publication_context);
         let (qmdb_resolver_actor, qmdb_sync_resolver) = qmdb_resolver::Actor::new(
             context.child("qmdb_resolver"),
             qmdb_resolver::Config {

glue/src/stateful/actor/core/mailbox.rs

@@ -22,8 +22,6 @@ use rand_core::Rng;
 use std::{collections::VecDeque, sync::Arc};
 use tracing::{Span, info_span};
 
-type RetryMailbox<E, A> = Arc<dyn Fn(Message<E, A>) + Send + Sync>;
-
 /// A verification is scoped to its caller.
 pub(in crate::stateful::actor) struct Verification {
     response: oneshot::Sender<bool>,
@@ -71,15 +69,6 @@ where
         span: Span,
         block: Arc<A::Block>,
         acknowledgement: Exact,
-        retry_mailbox: RetryMailbox<E, A>,
-    },
-
-    /// Requests the database set.
-    ///
-    /// The actor replies once startup handoff has produced the database set,
-    /// or immediately if that has already happened.
-    SubscribeDatabases {
-        response: oneshot::Sender<A::Databases>,
     },
 }
 
@@ -92,7 +81,6 @@ where
         match self {
             Self::Propose { response, .. } => response.is_closed(),
             Self::Verify { verification, .. } => verification.is_cancelled(),
-            Self::SubscribeDatabases { response } => response.is_closed(),
             Self::Finalized { .. } => false,
         }
     }
@@ -164,7 +152,6 @@ where
     A: Application<E>,
 {
     sender: Sender<Message<E, A>>,
-    retry_mailbox: RetryMailbox<E, A>,
 }
 
 impl<E, A> Clone for Mailbox<E, A>
@@ -175,7 +162,6 @@ where
     fn clone(&self) -> Self {
         Self {
             sender: self.sender.clone(),
-            retry_mailbox: self.retry_mailbox.clone(),
         }
     }
 }
@@ -186,44 +172,8 @@ where
     A: Application<E>,
 {
     /// Create a mailbox from the send half of the actor's message channel.
-    pub(super) fn new(sender: Sender<Message<E, A>>) -> Self {
-        let retry_sender = sender.clone();
-        let retry_mailbox = Arc::new(move |message| {
-            let _ = retry_sender.enqueue(message);
-        });
-        Self {
-            sender,
-            retry_mailbox,
-        }
-    }
-}
-
-impl<E, A> Mailbox<E, A>
-where
-    E: Rng + Spawner + Metrics + Clock,
-    A: Application<E>,
-{
-    /// Wait for the database set.
-    ///
-    /// This resolves once startup handoff has produced the database set. Late
-    /// callers receive the current database set immediately.
-    ///
-    /// ## Safety
-    ///
-    /// Holders must never manually prune these databases. Stateful uses
-    /// [`Config::prune_config`](crate::stateful::Config::prune_config) to
-    /// schedule safe pruning without pruning past the rewind window needed for
-    /// crash reconciliation. With pruning enabled, glue keeps a
-    /// `max_pending_acks + 1` finalized-target window plus the configured
-    /// extra block windows before pruning.
-    pub async fn subscribe_databases(&self) -> A::Databases {
-        let (response, receiver) = oneshot::channel();
-        let _ = self
-            .sender
-            .enqueue(Message::SubscribeDatabases { response });
-        receiver
-            .await
-            .expect("stateful actor dropped during subscribe_databases")
+    pub(super) const fn new(sender: Sender<Message<E, A>>) -> Self {
+        Self { sender }
     }
 }
 
@@ -277,9 +227,12 @@ where
             ancestry: BoxedAncestry::new(ancestry),
             verification: Verification { response },
         });
-        receiver
-            .await
-            .expect("stateful actor dropped during verify")
+        match receiver.await {
+            Ok(valid) => valid,
+            // The actor exited without answering. Never fabricate a verdict.
+            // Park until this future is dropped.
+            Err(_) => std::future::pending().await,
+        }
     }
 }
 
@@ -305,7 +258,6 @@ where
                     span,
                     block,
                     acknowledgement,
-                    retry_mailbox: self.retry_mailbox.clone(),
                 }
             }
         };

glue/src/stateful/actor/core/mod.rs

@@ -250,7 +250,6 @@ where
             sync_metadata,
             syncer: syncer_mailbox,
             deferred_verifications: Vec::new(),
-            database_subscribers: Vec::new(),
             artifact: None,
             snapshot_publisher: self.snapshot_publisher,
             sync_completed,
@@ -278,22 +277,19 @@ where
         let metrics = StatefulMetrics::new(self.context.as_present());
         let _ = metrics.sync_done.try_set(1);
         let processor = Processor::new(self.application, databases, anchor, metrics, self.pruning);
-
-        // The recovered state alone must publish before the loop starts, so
-        // serving begins before the next finalization.
+        // Publish the recovered state before the loop starts, so serving begins
+        // before the next finalization.
         let mut snapshot_publisher = self.snapshot_publisher;
-        processor.publish_snapshot(&mut snapshot_publisher).await;
+        let processor = processor.publish_snapshot(&mut snapshot_publisher).await;
         Processing {
             context: self.context,
             mailbox: self.mailbox,
             provider: self.provider,
             marshal,
-            processor,
             snapshot_publisher,
-            deferred_verifications: Vec::new(),
             skip_finalized_until,
         }
-        .start()
+        .start(processor, Vec::new())
         .await
     }
 }