chore(code): Rename Effect::Broadcast to Effect::Publish

code/crates/core-consensus/src/effect.rs

@@ -49,36 +49,53 @@ pub enum Effect<Ctx>
 where
     Ctx: Context,
 {
-    /// Reset all timeouts
-    /// Resume with: [`Resume::Continue`]
+    /// Reset all timeouts to their initial values
+    ///
+    /// Resume with: [`resume::Continue`]
     ResetTimeouts(resume::Continue),
 
-    /// Cancel all timeouts
-    /// Resume with: [`Resume::Continue`]
+    /// Cancel all outstanding timeouts
+    ///
+    /// Resume with: [`resume::Continue`]
     CancelAllTimeouts(resume::Continue),
 
     /// Cancel a given timeout
-    /// Resume with: [`Resume::Continue`]
+    ///
+    /// Resume with: [`resume::Continue`]
     CancelTimeout(Timeout, resume::Continue),
 
     /// Schedule a timeout
-    /// Resume with: [`Resume::Continue`]
+    ///
+    /// Resume with: [`resume::Continue`]
     ScheduleTimeout(Timeout, resume::Continue),
 
     /// Consensus is starting a new round with the given proposer
-    /// Resume with: [`Resume::Continue`]
+    ///
+    /// Resume with: [`resume::Continue`]
     StartRound(Ctx::Height, Round, Ctx::Address, resume::Continue),
 
-    /// Broadcast a message
-    /// Resume with: [`Resume::Continue`]
-    Broadcast(SignedConsensusMsg<Ctx>, resume::Continue),
-
-    /// Get a value to propose at the given height and round, within the given timeout
-    /// Resume with: [`Resume::Continue`]
+    /// Publish a message to peers
+    ///
+    /// Resume with: [`resume::Continue`]
+    Publish(SignedConsensusMsg<Ctx>, resume::Continue),
+
+    /// Requests the application to build a value for consensus to run on.
+    ///
+    /// Because this operation may be asynchronous, this effect does not expect a resumption
+    /// with a value, rather the application is expected to propose a value within the timeout duration.
+    ///
+    /// The application MUST eventually feed a [`ProposeValue`][crate::input::Input::ProposeValue]
+    /// input to consensus within the specified timeout duration.
+    ///
+    /// Resume with: [`resume::Continue`]
     GetValue(Ctx::Height, Round, Timeout, resume::Continue),
 
-    /// Restream the value identified by the given information.
-    /// Resume with: [`Resume::Continue`]
+    /// Requests the application to re-stream a proposal that it has already seen.
+    ///
+    /// The application MUST re-publish again to its pwers all
+    /// the proposal parts pertaining to that value.
+    ///
+    /// Resume with: [`resume::Continue`]
     RestreamValue(
         /// Height of the value
         Ctx::Height,
@@ -95,19 +112,27 @@ where
     ),
 
     /// Get the validator set at the given height
-    /// Resume with: [`Resume::ValidatorSet`]
+    ///
+    /// Resume with: [`resume::ValidatorSet`]
     GetValidatorSet(Ctx::Height, resume::ValidatorSet),
 
-    /// Consensus has decided on a value
-    /// Resume with: [`Resume::Continue`]
+    /// Notifies the application that consensus has decided on a value.
+    ///
+    /// This message includes a commit certificate containing the ID of
+    /// the value that was decided on, the height and round at which it was decided,
+    /// and the aggregated signatures of the validators that committed to it.
+    ///
+    /// Resume with: [`resume::Continue`]
     Decide(CommitCertificate<Ctx>, resume::Continue),
 
     /// Consensus has been stuck in Prevote or Precommit step, ask for vote sets from peers
-    /// Resume with: [`Resume::Continue`]
+    ///
+    /// Resume with: [`resume::Continue`]
     GetVoteSet(Ctx::Height, Round, resume::Continue),
 
     /// A peer has required our vote set, send the response
-    /// Resume with: [`Resume::Continue`]`
+    ///
+    /// Resume with: [`resume::Continue`]`
     SendVoteSetResponse(
         RequestId,
         Ctx::Height,
@@ -117,30 +142,37 @@ where
     ),
 
     /// Persist a consensus message in the Write-Ahead Log for crash recovery
-    /// Resume with: [`Resume::Continue`]`
+    ///
+    /// Resume with: [`resume::Continue`]`
     PersistMessage(SignedConsensusMsg<Ctx>, resume::Continue),
 
     /// Persist a timeout in the Write-Ahead Log for crash recovery
-    /// Resume with: [`Resume::Continue`]`
+    ///
+    /// Resume with: [`resume::Continue`]`
     PersistTimeout(Timeout, resume::Continue),
 
     /// Sign a vote with this node's private key
-    /// Resume with: [`Resume::SignedVote`]
+    ///
+    /// Resume with: [`resume::SignedVote`]
     SignVote(Ctx::Vote, resume::SignedVote),
 
     /// Sign a proposal with this node's private key
-    /// Resume with: [`Resume::SignedProposal`]
+    ///
+    /// Resume with: [`resume::SignedProposal`]
     SignProposal(Ctx::Proposal, resume::SignedProposal),
 
     /// Verify a signature
-    /// Resume with: [`Resume::SignatureValidity`]
+    ///
+    /// Resume with: [`resume::SignatureValidity`]
     VerifySignature(
         SignedMessage<Ctx, ConsensusMsg<Ctx>>,
         PublicKey<Ctx>,
         resume::SignatureValidity,
     ),
 
     /// Verify a commit certificate
+    ///
+    /// Resume with: [`resume::CertificateValidity`]
     VerifyCertificate(
         CommitCertificate<Ctx>,
         Ctx::ValidatorSet,

code/crates/core-consensus/src/handle/driver.rs

@@ -217,7 +217,7 @@ where
             if state.params.value_payload.include_proposal() {
                 perform!(
                     co,
-                    Effect::Broadcast(
+                    Effect::Publish(
                         SignedConsensusMsg::Proposal(signed_proposal),
                         Default::default()
                     )
@@ -242,7 +242,7 @@ where
 
             perform!(
                 co,
-                Effect::Broadcast(SignedConsensusMsg::Vote(signed_vote), Default::default())
+                Effect::Publish(SignedConsensusMsg::Vote(signed_vote), Default::default())
             );
 
             Ok(())

code/crates/engine/src/consensus.rs

@@ -870,7 +870,7 @@ where
                 Ok(r.resume_with(valid))
             }
 
-            Effect::Broadcast(msg, r) => {
+            Effect::Publish(msg, r) => {
                 // Sync the WAL to disk before we broadcast the message
                 // NOTE: The message has already been append to the WAL by the `PersistMessage` effect.
                 self.wal_flush(phase).await?;