[Consensus] Add MinBlockTime to delay mempool reaping

consensus/e2e_tests/pacemaker_test.go

@@ -172,6 +172,113 @@ func forcePacemakerTimeout(t *testing.T, clockMock *clock.Mock, paceMakerTimeout
 	advanceTime(t, clockMock, paceMakerTimeout+10*time.Millisecond)
 }
 
+func TestPacemaker_MinBlockTime_DelayBlockPrep(t *testing.T) {
+	// Test preparation
+	clockMock := clock.NewMock()
+	timeReminder(t, clockMock, time.Second)
+
+	// UnitTestNet configs
+	paceMakerTimeoutMsec := uint64(300000)
+	consensusMessageTimeout := time.Duration(paceMakerTimeoutMsec / 5) // Must be smaller than pacemaker timeout because we expect a deterministic number of consensus messages.
+	paceMakerMinBlockTimeMsec := uint64(paceMakerTimeoutMsec / 6)
+	runtimeMgrs := GenerateNodeRuntimeMgrs(t, numValidators, clockMock)
+	for _, runtimeConfig := range runtimeMgrs {
+		consCfg := runtimeConfig.GetConfig().Consensus.PacemakerConfig
+		consCfg.TimeoutMsec = paceMakerTimeoutMsec
+		consCfg.MinBlockTimeMsec = paceMakerMinBlockTimeMsec
+	}
+	buses := GenerateBuses(t, runtimeMgrs)
+
+	// Create & start test pocket nodes
+	eventsChannel := make(modules.EventsChannel, 100)
+	pocketNodes := CreateTestConsensusPocketNodes(t, buses, eventsChannel)
+	err := StartAllTestPocketNodes(t, pocketNodes)
+	require.NoError(t, err)
+
+	// First round ever has leaderId=2 ((height+round+step-1)%numValidators)+1
+	// See: consensus/leader_election/module.go#electNextLeaderDeterministicRoundRobin
+	leaderId := typesCons.NodeId(2)
+	numReplicas := len(pocketNodes) - 1
+
+	// Debug message to start consensus by triggering next view
+	// Get leader out of replica set
+	for _, pocketNode := range pocketNodes {
+		TriggerNextView(t, pocketNode)
+
+		// Right after triggering the next view
+		// Consensus started and all nodes are at NewRound step
+		step := typesCons.HotstuffStep(pocketNode.GetBus().GetConsensusModule().CurrentStep())
+		require.Equal(t, consensus.NewRound, step)
+	}
+
+	newRoundMessages, err := WaitForNetworkConsensusEvents(
+		t, clockMock, eventsChannel, typesCons.HotstuffStep(consensus.NewRound), typesCons.HotstuffMessageType(consensus.NewRound),
+		numReplicas, // We want new round messages from replicas only
+		consensusMessageTimeout, false,
+	)
+	require.NoError(t, err)
+
+	// Broadcast new round messages to leader to build a block
+	broadcastMessages(t, newRoundMessages, IdToNodeMapping{leaderId: pocketNodes[leaderId]})
+
+	var step typesCons.HotstuffStep
+	minTimeIncrement := 1 * time.Millisecond // Min time it takes to switch from NewRound to Prepare step
+
+	// Give go routines time to trigger
+	advanceTime(t, clockMock, 0)
+
+	// We get consensus module from leader to get its POV
+	leaderConsensusModule := pocketNodes[leaderId].GetBus().GetConsensusModule()
+
+	// Make sure all nodes are aligned to the same leader
+	for _, pocketNode := range pocketNodes {
+		nodeLeader := pocketNode.GetBus().GetConsensusModule().GetLeaderForView(1, 0, uint8(consensus.NewRound))
+		require.Equal(t, typesCons.NodeId(nodeLeader), leaderId)
+	}
+
+	// Timer is blocking the proposal step
+	step = typesCons.HotstuffStep(leaderConsensusModule.CurrentStep())
+	require.Equal(t, consensus.NewRound, step)
+
+	// Advance time right before minBlockTime triggers
+	paceMakerPreActivationTime := time.Duration(paceMakerMinBlockTimeMsec*uint64(time.Millisecond)) - minTimeIncrement
+	advanceTime(t, clockMock, paceMakerPreActivationTime)
+
+	// Should still be blocking proposal step
+	step = typesCons.HotstuffStep(leaderConsensusModule.CurrentStep())
+	require.Equal(t, consensus.NewRound, step)
+
+	// Advance time just enough to trigger minBlockTime
+	advanceTime(t, clockMock, minTimeIncrement)
+	step = typesCons.HotstuffStep(leaderConsensusModule.CurrentStep())
+
+	// Time advanced by minBlockTime
+	require.Equal(t, uint64(clockMock.Now().UnixMilli()), paceMakerMinBlockTimeMsec)
+	// Leader is at proposal step
+	require.Equal(t, consensus.Prepare, step)
+}
+
+// TODO: Block preparation triggers ASAP if conditions are met AFTER minBlockTime has triggered.
+func TestPacemaker_MinBlockTime_BlockPrepAsapAfterTrigger(t *testing.T) {
+	t.Skip()
+}
+
+// TODO: Block preparation is always discarded if a new one with better QC is received within minBlockTime.
+func TestPacemaker_MinBlockTime_AllowOnlyLatestBlockPrep(t *testing.T) {
+	t.Skip()
+}
+
+// TODO: Mempool reaped is the one present at minBlockTime or later.
+// Since leader could just reap an earlier mempool and just blocks broadcasting the proposal
+func TestPacemaker_MinBlockTime_DelayReapMempool(t *testing.T) {
+	t.Skip()
+}
+
+// TODO: Successive blocks timings are at least minBlockTime apart.
+func TestPacemaker_MinBlockTime_ConsecutiveBlocksAtLeastMinBlockTimeApart(t *testing.T) {
+	t.Skip()
+}
+
 // TODO: Implement these tests and use them as a starting point for new ones. Consider using ChatGPT to help you out :)
 
 func TestPacemakerDifferentHeightsCatchup(t *testing.T) {

consensus/hotstuff_leader.go

@@ -35,8 +35,6 @@ func (handler *HotstuffLeaderMessageHandler) HandleNewRoundMessage(m *consensusM
 		return
 	}
 
-	// DISCUSS: Do we need to pause for `MinBlockFreqMSec` here to let more transactions or should we stick with optimistic responsiveness?
-
 	if err := m.didReceiveEnoughMessageForStep(NewRound); err != nil {
 		m.logger.Info().Fields(hotstuffMsgToLoggingFields(msg)).Msgf("⏳ Waiting ⏳for more messages; %s", err.Error())
 		return
@@ -64,6 +62,14 @@ func (handler *HotstuffLeaderMessageHandler) HandleNewRoundMessage(m *consensusM
 	// TODO: Add test to make sure same block is not applied twice if round is interrupted after being 'Applied'.
 	// TODO: Add more unit tests for these checks...
 	if m.shouldPrepareNewBlock(highPrepareQC) {
+		// Leader should prepare a new block. Introducing a delay based on configurations.
+		m.paceMaker.StartMinBlockTimeDelay()
+
+		// This function delays block preparation and returns false if a concurrent preparation request with higher QC is available
+		if shouldPrepareBlock := m.paceMaker.DelayBlockPreparation(); !shouldPrepareBlock {
+			m.logger.Info().Msg("skip prepare new block, a candidate with higher QC is available")
+			return
+		}
 		block, err := m.prepareBlock(highPrepareQC)
 		if err != nil {
 			m.logger.Error().Err(err).Msg(typesCons.ErrPrepareBlock.Error())

consensus/pacemaker/module.go

@@ -3,6 +3,7 @@ package pacemaker
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	consensusTelemetry "github.com/pokt-network/pocket/consensus/telemetry"
@@ -38,6 +39,8 @@ type Pacemaker interface {
 	PacemakerDebug
 
 	ShouldHandleMessage(message *typesCons.HotstuffMessage) (bool, error)
+	StartMinBlockTimeDelay()
+	DelayBlockPreparation() bool
 
 	RestartTimer()
 	NewHeight()
@@ -48,9 +51,10 @@ type pacemaker struct {
 	base_modules.IntegrableModule
 	base_modules.InterruptableModule
 
-	pacemakerCfg    *configs.PacemakerConfig
-	roundTimeout    time.Duration
-	roundCancelFunc context.CancelFunc
+	pacemakerCfg       *configs.PacemakerConfig
+	roundTimeout       time.Duration
+	roundCancelFunc    context.CancelFunc
+	prepareStepDelayer prepareStepDelayer
 
 	// Only used for development and debugging.
 	debug pacemakerDebug
@@ -60,6 +64,17 @@ type pacemaker struct {
 	logPrefix string
 }
 
+// prepareStepDelayer delays block preparation (mempool reaping)
+// by adding a delay before the next prepare request to prevent block creation
+type prepareStepDelayer struct {
+	m                  sync.Mutex         // mutex locking access to this structure and prevent inconsistent state
+	ch                 chan bool          // whenever there is a block proposal request arriving before the timeout, this channel is used to signal to it to whether build the block or not (if better candidate request with higher QC)
+	cancelFunc         context.CancelFunc // cancels an ongoing timeout. It should not happen, but future code changes may no longer preserve this guarantee, so this feature maintain its own cancellation logic
+	shouldProposeBlock bool               // a flag to capture whether a block was proposed, so later calls will skip building the block
+	delayExhausted     bool               // the delay for this step/round has already passed, so should create the block ASAP
+	minBlockTime       time.Duration      // the minimum time to wait before trying to propose a block
+}
+
 func CreatePacemaker(bus modules.Bus, options ...modules.ModuleOption) (modules.Module, error) {
 	return new(pacemaker).Create(bus, options...)
 }
@@ -85,6 +100,14 @@ func (*pacemaker) Create(bus modules.Bus, options ...modules.ModuleOption) (modu
 		debugTimeBetweenStepsMsec: m.pacemakerCfg.GetDebugTimeBetweenStepsMsec(),
 		quorumCertificate:         nil,
 	}
+	m.prepareStepDelayer = prepareStepDelayer{
+		m:                  sync.Mutex{},
+		ch:                 nil,
+		cancelFunc:         nil,
+		shouldProposeBlock: false,
+		delayExhausted:     false,
+		minBlockTime:       time.Duration(m.pacemakerCfg.MinBlockTimeMsec * uint64(time.Millisecond)),
+	}
 
 	return m, nil
 }
@@ -243,6 +266,86 @@ func (m *pacemaker) NewHeight() {
 		)
 }
 
+// StartMinBlockTimeDelay should be called when a delay should be introduced into proposing a new block
+func (m *pacemaker) StartMinBlockTimeDelay() {
+	// Discard any previous timer if one exists
+	if m.prepareStepDelayer.cancelFunc != nil {
+		m.logger.Warn().Msg("RegisterMinBlockTimeDelay has an existing timer which should not happen. Releasing for now...")
+		m.prepareStepDelayer.cancelFunc()
+	}
+
+	m.prepareStepDelayer.shouldProposeBlock = false
+	m.prepareStepDelayer.delayExhausted = false
+
+	ctx, cancel := context.WithCancel(context.TODO())
+	m.prepareStepDelayer.cancelFunc = cancel
+
+	// Start a timer to wait for the MinBlockTimeMsec delay
+	// If a channel is provided, signal when the timer expires to it
+	go func() {
+		select {
+		// only called if the delay timer is explicitly cancelled
+		case <-ctx.Done():
+			return
+		// called after the minimum block delay is exhausted meaning it is time to propose a new block
+		case <-m.GetBus().GetRuntimeMgr().GetClock().After(m.prepareStepDelayer.minBlockTime):
+			m.prepareStepDelayer.m.Lock()
+			defer m.prepareStepDelayer.m.Unlock()
+
+			// After the timeout, if there was any `HotstuffLeaderMessageHandler.HandleNewRoundMessage()` call delayed to propose a block, unblock it by emitting true
+			if m.prepareStepDelayer.ch != nil {
+				m.prepareStepDelayer.ch <- true
+				close(m.prepareStepDelayer.ch)
+				m.prepareStepDelayer.shouldProposeBlock = true
+			}
+
+			// From now on, build the block ASAP
+			m.prepareStepDelayer.delayExhausted = true
+
+			// No need to cancel the context anymore
+			m.prepareStepDelayer.cancelFunc = nil
+		}
+	}()
+}
+
+// DelayBlockPreparation is called when conditions are met by the leader to build a block but still needs to wait for the MinBlockTimeMsec delay before reaping the mempool.
+// With MinBlockTimeMsec delay, multiple concurrent calls may happen
+// DelayBlockPreparation is a synchronous blocking function that waits for channel to emit whether to propose a block or not given multiple HotstuffLeaderMessageHandler.HandleNewRoundMessage calling this function concurrently.
+// It makes sure that:
+//   - Block proposal is made by only one of the possible `HotstuffLeaderMessageHandler.HandleNewRoundMessage()` concurrent (because delayed) calls
+//   - If the timer expires, the first call to this method will trigger the block proposal
+//   - If a late message is received AFTER the a block is marked as proposed by another call, the late message is discarded
+//   - Reads and assignments to pacemaker.prepareStepDelayer state are protected by a mutex
+func (m *pacemaker) DelayBlockPreparation() bool {
+	m.prepareStepDelayer.m.Lock()
+
+	if m.prepareStepDelayer.shouldProposeBlock {
+		m.prepareStepDelayer.m.Unlock()
+		return false
+	}
+
+	// If there already is a channel signaling block proposal, make sure it does not propose a block
+	if m.prepareStepDelayer.ch != nil {
+		m.prepareStepDelayer.ch <- false
+		close(m.prepareStepDelayer.ch)
+	}
+
+	// Deadline has passed, no need to have a channel, propose a block now
+	if m.prepareStepDelayer.delayExhausted {
+		m.prepareStepDelayer.shouldProposeBlock = true
+		m.prepareStepDelayer.m.Unlock()
+		return true
+	}
+
+	// We still need to wait, create a channel and discard the old candidate if any
+	ch := make(chan bool)
+	m.prepareStepDelayer.ch = ch
+	// We cannot defer the unlock here because the channel read is blocking
+	m.prepareStepDelayer.m.Unlock()
+
+	return <-ch
+}
+
 func (m *pacemaker) startNextView(qc *typesCons.QuorumCertificate, forceNextView bool) {
 	defer m.RestartTimer()
 

runtime/configs/config.go

@@ -122,6 +122,7 @@ func NewDefaultConfig(options ...func(*Config)) *Config {
 				TimeoutMsec:               defaults.DefaultPacemakerTimeoutMsec,
 				Manual:                    defaults.DefaultPacemakerManual,
 				DebugTimeBetweenStepsMsec: defaults.DefaultPacemakerDebugTimeBetweenStepsMsec,
+				MinBlockTimeMsec:          defaults.DefaultPacemakerMinBlockTimeMsec,
 			},
 		},
 		Utility: &UtilityConfig{

runtime/configs/proto/consensus_config.proto

@@ -15,4 +15,7 @@ message PacemakerConfig {
   uint64 timeout_msec = 1;
   bool manual = 2;
   uint64 debug_time_between_steps_msec = 3;
+  // consenus could produce blocks as soon as a quorum is reached; responsivness per the Hotstuff whitepaper.
+  // This option allows to set min time between blocks and gives more time to the mempool to fill up; similar to timeout_propose in Tendermint.
+  uint64 min_block_time_msec = 4;
 }

runtime/defaults/defaults.go

@@ -43,6 +43,7 @@ var (
 	DefaultPacemakerTimeoutMsec               = uint64(10000)
 	DefaultPacemakerManual                    = true
 	DefaultPacemakerDebugTimeBetweenStepsMsec = uint64(1000)
+	DefaultPacemakerMinBlockTimeMsec          = uint64(5000) // 5 seconds
 	// utility
 	DefaultUtilityMaxMempoolTransactionBytes = uint64(1024 ^ 3) // 1GB V0 defaults
 	DefaultUtilityMaxMempoolTransactions     = uint32(9000)

runtime/manager_test.go

@@ -1775,6 +1775,7 @@ func TestNewManagerFromReaders(t *testing.T) {
 							TimeoutMsec:               10000,
 							Manual:                    true,
 							DebugTimeBetweenStepsMsec: 1000,
+							MinBlockTimeMsec:          5000,
 						},
 						ServerModeEnabled: true,
 					},