diff --git a/code/crates/core-consensus/src/effect.rs b/code/crates/core-consensus/src/effect.rs index 6bd2cb257..44aa32829 100644 --- a/code/crates/core-consensus/src/effect.rs +++ b/code/crates/core-consensus/src/effect.rs @@ -49,36 +49,53 @@ pub enum Effect 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, 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, 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, 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,23 +142,28 @@ where ), /// Persist a consensus message in the Write-Ahead Log for crash recovery - /// Resume with: [`Resume::Continue`]` + /// + /// Resume with: [`resume::Continue`]` PersistMessage(SignedConsensusMsg, 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>, PublicKey, @@ -141,6 +171,8 @@ where ), /// Verify a commit certificate + /// + /// Resume with: [`resume::CertificateValidity`] VerifyCertificate( CommitCertificate, Ctx::ValidatorSet, diff --git a/code/crates/core-consensus/src/handle/driver.rs b/code/crates/core-consensus/src/handle/driver.rs index 092c6b560..94e5ba7f4 100644 --- a/code/crates/core-consensus/src/handle/driver.rs +++ b/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(()) diff --git a/code/crates/engine/src/consensus.rs b/code/crates/engine/src/consensus.rs index 456a53a52..457d9bd35 100644 --- a/code/crates/engine/src/consensus.rs +++ b/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?;