[Consensus] Add MinBlockTime to delay mempool reaping

consensus/e2e_tests/pacemaker_test.go

@@ -172,6 +172,118 @@ func forcePacemakerTimeout(t *testing.T, clockMock *clock.Mock, paceMakerTimeout
 	advanceTime(t, clockMock, paceMakerTimeout+10*time.Millisecond)
 }
 
+func TestPacemaker_MinBlockTime(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(5000)                          // Make sure it is larger than the consensus message timeout
+	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)
+
+	replicas := IdToNodeMapping{}
+	// First round ever has leaderId=2 ((height+round+step-1)%numValidators)+1
+	// See: consensus/leader_election/module.go#electNextLeaderDeterministicRoundRobin
+	leaderId := typesCons.NodeId(2)
+	leader := IdToNodeMapping{}
+	numReplicas := len(pocketNodes) - 1
+
+	// Debug message to start consensus by triggering next view
+	// Get leader out of replica set
+	for id, pocketNode := range pocketNodes {
+		TriggerNextView(t, pocketNode)
+		if id == leaderId {
+			leader[id] = pocketNode
+		} else {
+			replicas[id] = 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, leader)
+
+	var step typesCons.HotstuffStep
+	var pivotTime = 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 := leader[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
+	advanceTime(t, clockMock, time.Duration(paceMakerMinBlockTimeMsec*uint64(time.Millisecond))-pivotTime)
+
+	// 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, pivotTime)
+	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.
+func TestPacemaker_MinBlockTime_DelayReapMempool(t *testing.T) {
+	t.Skip()
+}
+
+// TODO: Successive blocks timings are at least minBlockTime apart.
+func TestPacemaker_MinBlockTime_BehaviorAcrossMultipleBlocks(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/e2e_tests/utils_test.go

@@ -840,6 +840,8 @@ func advanceTime(t *testing.T, clck *clock.Mock, duration time.Duration) {
 	clck.Add(duration)
 	t.Logf("[⌚ CLOCK ⏩] advanced by %v", duration)
 	logTime(t, clck)
+	// Give goroutines a chance to run
+	clck.Add(0)
 }
 
 // sleep pauses the goroutine for the given duration on the mock clock and logs what just happened.

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,15 @@ 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) {
+		// Place this where we want to start the timer
+		// Placed here, timer starts when we receive enough NewRound messages
+		m.paceMaker.RegisterMinBlockTimeDelay()
+
+		// 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)
+	RegisterMinBlockTimeDelay()
+	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,16 @@ type pacemaker struct {
 	logPrefix string
 }
 
+// Structure to handle delaying block preparation (reaping the block mempool)
+// by adding a delay before the next prepare request
+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 // This is used to cancel 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
+	blockProposed  bool
+	deadlinePassed bool
+}
+
 func CreatePacemaker(bus modules.Bus, options ...modules.ModuleOption) (modules.Module, error) {
 	return new(pacemaker).Create(bus, options...)
 }
@@ -85,6 +99,13 @@ 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,
+		blockProposed:  false,
+		deadlinePassed: false,
+	}
 
 	return m, nil
 }
@@ -171,6 +192,7 @@ func (m *pacemaker) ShouldHandleMessage(msg *typesCons.HotstuffMessage) (bool, e
 
 func (m *pacemaker) RestartTimer() {
 	// NOTE: Not deferring a cancel call because this function is asynchronous.
+	// DISCUSS: Should we have a lock to manipulate m.roundCancelFunc?
 	if m.roundCancelFunc != nil {
 		m.roundCancelFunc()
 	}
@@ -243,6 +265,82 @@ func (m *pacemaker) NewHeight() {
 		)
 }
 
+// This is called each time the system wants a delayed signal to build a block
+func (m *pacemaker) RegisterMinBlockTimeDelay() {
+	// Discard any previous timer
+	// DISCUSS: This should not happen, Identify cases where an active timer has to be discarded, remove cancellation logic if none
+	if m.prepareStepDelayer.cancelFunc != nil {
+		m.prepareStepDelayer.cancelFunc()
+	}
+
+	m.prepareStepDelayer.blockProposed = false
+	m.prepareStepDelayer.deadlinePassed = false
+
+	minBlockTime := time.Duration(m.pacemakerCfg.MinBlockTimeMsec * uint64(time.Millisecond))
+	ctx, cancel := context.WithCancel(context.TODO())
+	m.prepareStepDelayer.cancelFunc = cancel
+
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-m.GetBus().GetRuntimeMgr().GetClock().After(minBlockTime):
+			m.prepareStepDelayer.m.Lock()
+			defer m.prepareStepDelayer.m.Unlock()
+
+			// After the timeout, if there was any candidate request waiting for a signal, tell it to build the block
+			if m.prepareStepDelayer.ch != nil {
+				m.prepareStepDelayer.ch <- true
+				close(m.prepareStepDelayer.ch)
+				m.prepareStepDelayer.blockProposed = true
+			}
+
+			// From now on, build the block ASAP
+			m.prepareStepDelayer.deadlinePassed = true
+
+			// No need to cancel the context anymore
+			m.prepareStepDelayer.cancelFunc = nil
+		}
+	}()
+}
+
+// This 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
+// 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 affectations to pacemaker.DelayBlockPreparation state are protected by a mutex
+func (m *pacemaker) DelayBlockPreparation() bool {
+	m.prepareStepDelayer.m.Lock()
+
+	if m.prepareStepDelayer.blockProposed {
+		m.prepareStepDelayer.m.Unlock()
+		return false
+	}
+
+	// If there already is a channel signaling block proposal, make sure it gives up
+	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.deadlinePassed {
+		m.prepareStepDelayer.blockProposed = 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,5 @@ message PacemakerConfig {
   uint64 timeout_msec = 1;
   bool manual = 2;
   uint64 debug_time_between_steps_msec = 3;
+  uint64 min_block_time_msec = 4; // consenus protocol could produce blocks as soon as quorum is reached. This option allows to set min time between blocks and give more time to the mempool to fill up
 }

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,
 					},