[glue] DKG+reshare

consensus/src/lib.rs

@@ -274,15 +274,22 @@ stability_scope!(ALPHA, cfg(not(target_arch = "wasm32")) {
         /// The block type produced by the application's builder.
         type Block: Block;
 
+        /// Per-proposal input handed to [`propose`](Self::propose). Applications
+        /// that need no input set this to `()`.
+        type Input: Send;
+
         /// Build a new block on top of the provided parent ancestry. If the build job fails,
         /// or the proposer's slot should be skipped, the implementor should return [None].
         ///
+        /// `input` is the per-proposal input for this build.
+        ///
         /// This future may be cancelled before it completes. Implementations must be
         /// cancellation-safe.
         fn propose(
             &mut self,
             context: (E, Self::Context),
             ancestry: impl Ancestry<Self::Block>,
+            input: Self::Input,
         ) -> impl Future<Output = Option<Self::Block>> + Send;
 
         /// Verify a block produced by the application's proposer, relative to its ancestry.

consensus/src/marshal/ancestry.rs

@@ -17,7 +17,7 @@ use std::{
 };
 
 /// A stream of blocks used by application propose and verify calls.
-pub trait Ancestry<B: Block>: Stream<Item = Arc<B>> + Send + Unpin + 'static {
+pub trait Ancestry<B: Block>: Stream<Item = Arc<B>> + Clone + Send + Unpin + 'static {
     /// Peeks at the latest block in the stream without consuming it. Returns [None]
     /// if the stream does not yet have a block available or has been exhausted.
     fn peek(&self) -> Option<&B>;
@@ -27,15 +27,79 @@ pub trait Ancestry<B: Block>: Stream<Item = Arc<B>> + Send + Unpin + 'static {
 ///
 /// Blocks are yielded in iterator order and no parent fetching is performed. This is useful when
 /// the caller wants to bound the ancestry available to the application.
-pub fn from_iter<B>(blocks: impl IntoIterator<Item = Arc<B>>) -> impl Ancestry<B>
+pub fn from_iter<B: Block>(blocks: impl IntoIterator<Item = Arc<B>>) -> impl Ancestry<B> {
+    BoundedAncestry {
+        blocks: blocks.into_iter().collect(),
+    }
+}
+
+/// Prepends a fixed sequence of blocks to an existing ancestry stream.
+///
+/// Blocks are yielded in iterator order before the tail is polled.
+pub fn with_prefix<B, S>(blocks: impl IntoIterator<Item = Arc<B>>, tail: S) -> impl Ancestry<B>
 where
     B: Block,
+    S: Ancestry<B>,
 {
-    BoundedAncestry {
+    PrefixedAncestry {
         blocks: blocks.into_iter().collect(),
+        tail,
+    }
+}
+
+/// Type-erased ancestry stream that preserves cloneability.
+pub struct BoxedAncestry<B: Block>(Box<dyn ErasedAncestry<B>>);
+
+impl<B: Block> BoxedAncestry<B> {
+    /// Erases the concrete ancestry stream type.
+    pub fn new(ancestry: impl Ancestry<B>) -> Self {
+        Self(Box::new(ancestry))
+    }
+}
+
+impl<B: Block> Clone for BoxedAncestry<B> {
+    fn clone(&self) -> Self {
+        Self(self.0.clone_box())
+    }
+}
+
+impl<B: Block> Unpin for BoxedAncestry<B> {}
+
+impl<B: Block> Ancestry<B> for BoxedAncestry<B> {
+    fn peek(&self) -> Option<&B> {
+        self.0.peek_erased()
     }
 }
 
+impl<B: Block> Stream for BoxedAncestry<B> {
+    type Item = Arc<B>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut *self.0).poll_next(cx)
+    }
+}
+
+trait ErasedAncestry<B: Block>: Stream<Item = Arc<B>> + Send + Unpin + 'static {
+    fn peek_erased(&self) -> Option<&B>;
+
+    fn clone_box(&self) -> Box<dyn ErasedAncestry<B>>;
+}
+
+impl<B, A> ErasedAncestry<B> for A
+where
+    B: Block,
+    A: Ancestry<B>,
+{
+    fn peek_erased(&self) -> Option<&B> {
+        Ancestry::peek(self)
+    }
+
+    fn clone_box(&self) -> Box<dyn ErasedAncestry<B>> {
+        Box::new(self.clone())
+    }
+}
+
+#[derive(Clone)]
 struct BoundedAncestry<B: Block> {
     blocks: VecDeque<Arc<B>>,
 }
@@ -56,6 +120,39 @@ impl<B: Block> Stream for BoundedAncestry<B> {
     }
 }
 
+#[derive(Clone)]
+struct PrefixedAncestry<B: Block, S> {
+    blocks: VecDeque<Arc<B>>,
+    tail: S,
+}
+
+impl<B: Block, S> Unpin for PrefixedAncestry<B, S> {}
+
+impl<B, S> Ancestry<B> for PrefixedAncestry<B, S>
+where
+    B: Block,
+    S: Ancestry<B>,
+{
+    fn peek(&self) -> Option<&B> {
+        self.blocks.front().map(Arc::as_ref)
+    }
+}
+
+impl<B, S> Stream for PrefixedAncestry<B, S>
+where
+    B: Block,
+    S: Ancestry<B>,
+{
+    type Item = Arc<B>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        if let Some(block) = self.blocks.pop_front() {
+            return Poll::Ready(Some(block));
+        }
+        Pin::new(&mut self.tail).poll_next(cx)
+    }
+}
+
 /// An interface for providing parent blocks.
 pub trait BlockProvider: Send + 'static {
     /// The block type the provider walks.
@@ -192,9 +289,25 @@ impl<M: BlockProvider, C: Clock> AncestorStream<M, C> {
     }
 }
 
+impl<M, C> Clone for AncestorStream<M, C>
+where
+    M: BlockProvider + Clone,
+    C: Clock,
+{
+    fn clone(&self) -> Self {
+        Self {
+            buffered: self.buffered.clone(),
+            marshal: self.marshal.clone(),
+            fetch_duration: self.fetch_duration.clone(),
+            clock: self.clock.clone(),
+            pending: None.into(),
+        }
+    }
+}
+
 impl<M, C> Ancestry<M::Block> for AncestorStream<M, C>
 where
-    M: BlockProvider,
+    M: BlockProvider + Clone,
     C: Clock,
 {
     fn peek(&self) -> Option<&M::Block> {

consensus/src/marshal/coding/marshaled.rs

@@ -220,6 +220,7 @@ where
             Block = B,
             SigningScheme = Z::Scheme,
             Context = Context<Commitment, <Z::Scheme as Verifier>::PublicKey>,
+            Input = (),
         >,
     B: CertifiableBlock<Context = <A as Application<E>>::Context>,
     C: CodingScheme,
@@ -645,6 +646,7 @@ where
             Block = B,
             SigningScheme = Z::Scheme,
             Context = Context<Commitment, <Z::Scheme as Verifier>::PublicKey>,
+            Input = (),
         >,
     B: CertifiableBlock<Context = <A as Application<E>>::Context>,
     C: CodingScheme,
@@ -821,6 +823,7 @@ where
                             consensus_context.clone(),
                         ),
                         ancestor_stream,
+                        (),
                     )
                     .instrument(info_span!(
                         "marshal.coding.application.propose",
@@ -1088,6 +1091,7 @@ where
             Block = B,
             SigningScheme = Z::Scheme,
             Context = Context<Commitment, <Z::Scheme as Verifier>::PublicKey>,
+            Input = (),
         >,
     B: CertifiableBlock<Context = <A as Application<E>>::Context>,
     C: CodingScheme,

consensus/src/marshal/mocks/verifying.rs

@@ -71,11 +71,13 @@ where
     type Block = B;
     type Context = B::Context;
     type SigningScheme = S;
+    type Input = ();
 
     async fn propose(
         &mut self,
         _context: (deterministic::Context, Self::Context),
         _ancestry: impl Ancestry<Self::Block>,
+        _input: Self::Input,
     ) -> Option<Self::Block> {
         self.propose_result.clone()
     }
@@ -126,11 +128,13 @@ where
     type Block = B;
     type Context = B::Context;
     type SigningScheme = S;
+    type Input = ();
 
     async fn propose(
         &mut self,
         _context: (deterministic::Context, Self::Context),
         _ancestry: impl Ancestry<Self::Block>,
+        _input: Self::Input,
     ) -> Option<Self::Block> {
         None
     }

consensus/src/marshal/standard/deferred.rs

@@ -184,7 +184,13 @@ impl<E, S, A, B, ES> Deferred<E, S, A, B, ES>
 where
     E: Rng + Spawner + Metrics + Clock,
     S: Scheme,
-    A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+    A: Application<
+            E,
+            Block = B,
+            SigningScheme = S,
+            Context = Context<B::Digest, S::PublicKey>,
+            Input = (),
+        >,
     B: CertifiableBlock<Context = <A as Application<E>>::Context>,
     ES: Epocher,
 {
@@ -463,7 +469,13 @@ impl<E, S, A, B, ES> Automaton for Deferred<E, S, A, B, ES>
 where
     E: Rng + Spawner + Metrics + Clock,
     S: Scheme,
-    A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+    A: Application<
+            E,
+            Block = B,
+            SigningScheme = S,
+            Context = Context<B::Digest, S::PublicKey>,
+            Input = (),
+        >,
     B: CertifiableBlock<Context = <A as Application<E>>::Context>,
     ES: Epocher,
 {
@@ -627,6 +639,7 @@ where
                             consensus_context.clone(),
                         ),
                         ancestor_stream,
+                        (),
                     )
                     .instrument(info_span!(
                         "marshal.deferred.application.propose",
@@ -822,7 +835,13 @@ impl<E, S, A, B, ES> CertifiableAutomaton for Deferred<E, S, A, B, ES>
 where
     E: Rng + Spawner + Metrics + Clock,
     S: Scheme,
-    A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+    A: Application<
+            E,
+            Block = B,
+            SigningScheme = S,
+            Context = Context<B::Digest, S::PublicKey>,
+            Input = (),
+        >,
     B: CertifiableBlock<Context = <A as Application<E>>::Context>,
     ES: Epocher,
 {

consensus/src/marshal/standard/inline.rs

@@ -174,7 +174,13 @@ impl<E, S, A, B, ES> Inline<E, S, A, B, ES>
 where
     E: Rng + Spawner + Metrics + Clock,
     S: Scheme,
-    A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+    A: Application<
+            E,
+            Block = B,
+            SigningScheme = S,
+            Context = Context<B::Digest, S::PublicKey>,
+            Input = (),
+        >,
     B: Block + Clone,
     ES: Epocher,
 {
@@ -218,7 +224,13 @@ impl<E, S, A, B, ES> Automaton for Inline<E, S, A, B, ES>
 where
     E: Rng + Spawner + Metrics + Clock,
     S: Scheme,
-    A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+    A: Application<
+            E,
+            Block = B,
+            SigningScheme = S,
+            Context = Context<B::Digest, S::PublicKey>,
+            Input = (),
+        >,
     B: Block + Clone,
     ES: Epocher,
 {
@@ -349,6 +361,7 @@ where
                             consensus_context.clone(),
                         ),
                         ancestor_stream,
+                        (),
                     )
                     .instrument(info_span!(
                         "marshal.inline.application.propose",
@@ -558,7 +571,13 @@ impl<E, S, A, B, ES> CertifiableAutomaton for Inline<E, S, A, B, ES>
 where
     E: Rng + Spawner + Metrics + Clock,
     S: Scheme,
-    A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+    A: Application<
+            E,
+            Block = B,
+            SigningScheme = S,
+            Context = Context<B::Digest, S::PublicKey>,
+            Input = (),
+        >,
     B: Block + Clone,
     ES: Epocher,
 {
@@ -707,7 +726,13 @@ mod tests {
     where
         E: Rng + Spawner + Metrics + Clock,
         S: Scheme,
-        A: Application<E, Block = B, SigningScheme = S, Context = Context<B::Digest, S::PublicKey>>,
+        A: Application<
+                E,
+                Block = B,
+                SigningScheme = S,
+                Context = Context<B::Digest, S::PublicKey>,
+                Input = (),
+            >,
         B: Block + Clone,
         ES: crate::types::Epocher,
     {

consensus/src/types.rs

@@ -603,6 +603,14 @@ impl FixedEpocher {
         let last = first.checked_add(self.0 - 1)?;
         Some((Height::new(first), Height::new(last)))
     }
+
+    /// Returns the midpoint block height in the given epoch.
+    ///
+    /// Returns `None` if the epoch is not supported.
+    pub fn midpoint(&self, epoch: Epoch) -> Option<Height> {
+        let (first, _) = self.bounds(epoch)?;
+        first.get().checked_add(self.0 / 2).map(Height::new)
+    }
 }
 
 impl Epocher for FixedEpocher {

examples/reshare/src/application/core.rs

@@ -65,11 +65,13 @@ where
     type Context = Context<H::Digest, C::PublicKey>;
     type SigningScheme = S;
     type Block = Block<H, C, V>;
+    type Input = ();
 
     async fn propose(
         &mut self,
         (_, context): (E, Self::Context),
         mut ancestry: impl Ancestry<Self::Block>,
+        _: Self::Input,
     ) -> Option<Self::Block> {
         // Fetch the parent block from the ancestry stream.
         let parent_block = ancestry.next().await?;