[glue] Add glue crate

Author: clabby

.github/workflows/publish.yml

@@ -115,6 +115,11 @@ jobs:
       continue-on-error: true
       env:
         CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
+    - name: Publish glue
+      run: cargo publish --manifest-path glue/Cargo.toml
+      continue-on-error: true
+      env:
+        CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
     - name: Publish chat
       run: cargo publish --manifest-path examples/chat/Cargo.toml
       continue-on-error: true

Cargo.lock

@@ -9,6 +9,7 @@ members = [
     "consensus",
     "cryptography",
     "deployer",
+    "glue",
     "macros",
     "macros/impl",
     "math",
@@ -96,6 +97,7 @@ commonware-conformance-macros = { version = "2026.3.0", path = "conformance/macr
 commonware-consensus = { version = "2026.3.0", path = "consensus" }
 commonware-cryptography = { version = "2026.3.0", path = "cryptography", default-features = false }
 commonware-deployer = { version = "2026.3.0", path = "deployer", default-features = false }
+commonware-glue = { version = "2026.3.0", path = "glue" }
 commonware-invariants = { version = "2026.3.0", path = "invariants" }
 commonware-macros = { version = "2026.3.0", path = "macros", default-features = false }
 commonware-macros-impl = { version = "2026.3.0", path = "macros/impl" }

Cargo.toml

@@ -20,6 +20,7 @@ _Primitives are designed for deployment in adversarial environments. If you find
 * [consensus](./consensus/README.md): Order opaque messages in a Byzantine environment.
 * [cryptography](./cryptography/README.md): Generate keys, sign arbitrary messages, and deterministically verify signatures.
 * [deployer](./deployer/README.md): Deploy infrastructure across cloud providers.
+* [glue](./glue/README.md): Bootstrap applications with commonware primitive compositions.
 * [math](./math/README.md): Create and manipulate mathematical objects.
 * [p2p](./p2p/README.md): Communicate with authenticated peers over encrypted connections.
 * [parallel](./parallel/README.md): Parallelize fold operations with pluggable execution strategies.

README.md

@@ -2,6 +2,7 @@
 
 use crate::{types::Height, Block, Heightable};
 use commonware_cryptography::Digestible;
+use commonware_utils::sync::Mutex;
 use futures::{
     future::{BoxFuture, OptionFuture},
     FutureExt, Stream,
@@ -10,6 +11,7 @@ use pin_project::pin_project;
 use std::{
     future::Future,
     pin::Pin,
+    sync::Arc,
     task::{Context, Poll},
 };
 
@@ -34,6 +36,45 @@ pub trait BlockProvider: Clone + Send + 'static {
     ) -> impl Future<Output = Option<Self::Block>> + Send;
 }
 
+/// A type-erased [`BlockProvider`] that wraps any concrete provider behind
+/// a shared closure.
+///
+/// Used by actor mailboxes and other channel-based patterns where a generic
+/// `BlockProvider` type parameter must be eliminated before sending across
+/// a channel.
+#[derive(Clone)]
+pub struct ErasedBlockProvider<B: Block> {
+    fetch: Arc<dyn Fn(<B as Digestible>::Digest) -> BoxFuture<'static, Option<B>> + Send + Sync>,
+}
+
+impl<B: Block> ErasedBlockProvider<B> {
+    /// Erase a concrete [`BlockProvider`] into a type-erased provider.
+    ///
+    /// The provider is wrapped in a `Mutex` so that the resulting closure
+    /// is `Sync` (required by `Arc<dyn Fn + Send + Sync>`). The lock is
+    /// held only for the duration of `clone()`, never across an await.
+    pub fn new<M: BlockProvider<Block = B>>(provider: M) -> Self {
+        let provider = Arc::new(Mutex::new(provider));
+        Self {
+            fetch: Arc::new(move |digest| {
+                let p = provider.lock().clone();
+                Box::pin(async move { p.fetch_block(digest).await })
+            }),
+        }
+    }
+}
+
+impl<B: Block> BlockProvider for ErasedBlockProvider<B> {
+    type Block = B;
+
+    fn fetch_block(
+        self,
+        digest: <B as Digestible>::Digest,
+    ) -> impl Future<Output = Option<B>> + Send {
+        (self.fetch)(digest)
+    }
+}
+
 /// Yields the ancestors of a block while prefetching parents, _not_ including the genesis block.
 ///
 // TODO(<https://github.com/commonwarexyz/monorepo/issues/2982>): Once marshal can also yield the genesis block,
@@ -47,6 +88,31 @@ pub struct AncestorStream<M, B: Block> {
 }
 
 impl<M, B: Block> AncestorStream<M, B> {
+    /// Returns a reference to the next block that will be yielded by the
+    /// stream, without consuming it.
+    ///
+    /// Returns `None` if the buffer is empty and the next block is being
+    /// fetched asynchronously.
+    pub fn peek(&self) -> Option<&B> {
+        self.buffered.last()
+    }
+
+    /// Erase the block provider type, producing an
+    /// `AncestorStream<ErasedBlockProvider<B>, B>`.
+    ///
+    /// The returned stream behaves identically but can be sent across
+    /// channels that require a concrete type.
+    pub fn erase(self) -> AncestorStream<ErasedBlockProvider<B>, B>
+    where
+        M: BlockProvider<Block = B>,
+    {
+        AncestorStream {
+            buffered: self.buffered,
+            marshal: ErasedBlockProvider::new(self.marshal),
+            pending: self.pending,
+        }
+    }
+
     /// Creates a new [AncestorStream] starting from the given ancestry.
     ///
     /// # Panics

consensus/src/marshal/ancestry.rs

@@ -157,6 +157,11 @@ mod tests {
         harness::prune_finalized_archives::<CodingHarness>();
     }
 
+    #[test_traced("WARN")]
+    fn test_coding_set_floor_without_pruning_preserves_archives() {
+        harness::set_floor_without_pruning_preserves_archives::<CodingHarness>();
+    }
+
     #[test_traced("WARN")]
     fn test_coding_rejects_block_delivery_below_floor() {
         harness::reject_stale_block_delivery_after_floor_update::<CodingHarness>();

consensus/src/marshal/coding/mod.rs

@@ -625,6 +625,9 @@ where
                         let finalization = self.get_finalization_by_height(height).await;
                         response.send_lossy(finalization);
                     }
+                    Message::GetProcessedHeight { response } => {
+                        response.send_lossy(self.last_processed_height);
+                    }
                     Message::HintFinalized { height, targets } => {
                         // Skip if height is at or below the floor
                         if height <= self.last_processed_height {
@@ -670,7 +673,10 @@ where
                         )
                         .await;
                     }
-                    Message::SetFloor { height } => {
+                    Message::SetFloor {
+                        height,
+                        prune_archives,
+                    } => {
                         if self.last_processed_height >= height {
                             warn!(
                                 %height,
@@ -692,10 +698,12 @@ where
                         // updating `last_processed_height`.
                         self.pending_acks.clear();
 
-                        // Prune data in the finalized archives below the new floor.
-                        if let Err(err) = self.prune_finalized_archives(height).await {
-                            error!(?err, %height, "failed to prune finalized archives");
-                            return;
+                        if prune_archives {
+                            // Prune data in the finalized archives below the new floor.
+                            if let Err(err) = self.prune_finalized_archives(height).await {
+                                error!(?err, %height, "failed to prune finalized archives");
+                                return;
+                            }
                         }
 
                         // Intentionally keep existing block subscriptions alive. Canceling

consensus/src/marshal/core/actor.rs

@@ -45,6 +45,11 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
         /// A channel to send the retrieved finalization.
         response: oneshot::Sender<Option<Finalization<S, V::Commitment>>>,
     },
+    /// A request to retrieve the latest processed height acknowledged by the application.
+    GetProcessedHeight {
+        /// A channel to send the latest processed height.
+        response: oneshot::Sender<Height>,
+    },
     /// A hint that a finalized block may be available at a given height.
     ///
     /// This triggers a network fetch if the finalization is not available locally.
@@ -111,7 +116,7 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
     /// Sets the sync starting point (advances if higher than current).
     ///
     /// Marshal will sync and deliver blocks starting at `floor + 1`. Data below
-    /// the floor is pruned.
+    /// the floor is pruned when `prune_archives` is `true`.
     ///
     /// To prune data without affecting the sync starting point (say at some trailing depth
     /// from tip), use [Message::Prune] instead.
@@ -120,6 +125,9 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
     SetFloor {
         /// The candidate floor height.
         height: Height,
+
+        /// Whether to prune finalized archives below the new floor.
+        prune_archives: bool,
     },
     /// Prunes finalized blocks and certificates below the given height.
     ///
@@ -194,6 +202,13 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
             .flatten()
     }
 
+    /// Retrieve the latest processed height acknowledged by the application.
+    pub async fn get_processed_height(&self) -> Option<Height> {
+        self.sender
+            .request(|response| Message::GetProcessedHeight { response })
+            .await
+    }
+
     /// Hints that a finalized block may be available at the given height.
     ///
     /// This method will request the finalization from the network via the resolver
@@ -300,14 +315,19 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
     /// Sets the sync starting point (advances if higher than current).
     ///
     /// Marshal will sync and deliver blocks starting at `floor + 1`. Data below
-    /// the floor is pruned.
+    /// the floor is pruned when `prune_archives` is `true`.
     ///
     /// To prune data without affecting the sync starting point (say at some trailing depth
     /// from tip), use [Self::prune] instead.
     ///
     /// The default floor is 0.
-    pub async fn set_floor(&self, height: Height) {
-        self.sender.send_lossy(Message::SetFloor { height }).await;
+    pub async fn set_floor(&self, height: Height, prune_archives: bool) {
+        self.sender
+            .send_lossy(Message::SetFloor {
+                height,
+                prune_archives,
+            })
+            .await;
     }
 
     /// Prunes finalized blocks and certificates below the given height.