[Consensus] Add MinBlockTime to delay mempool reaping

consensus/doc/CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- Implement minimum block production pace by delaying block preparation
+
 ## [0.0.0.54] - 2023-06-13
 
 - Fix tests

consensus/e2e_tests/pacemaker_test.go

@@ -172,6 +172,56 @@ func forcePacemakerTimeout(t *testing.T, clockMock *clock.Mock, paceMakerTimeout
 	advanceTime(t, clockMock, paceMakerTimeout+10*time.Millisecond)
 }
 
+func TestPacemakerMinBlockTime(t *testing.T) {
+	// Test preparation
+	clockMock := clock.NewMock()
+	timeReminder(t, clockMock, time.Second)
+
+	// UnitTestNet configs
+	paceMakerTimeoutMsec := uint64(10000)
+	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)
+
+	// Debug message to start consensus by triggering next view
+	for _, pocketNode := range pocketNodes {
+		TriggerNextView(t, pocketNode)
+	}
+
+	consMod := pocketNodes[1].GetBus().GetConsensusModule()
+
+	newRoundMessages, err := waitForProposalMsgs(t, clockMock, eventsChannel, pocketNodes, 1, uint8(consensus.NewRound), 0, 0, numValidators*numValidators, consensusMessageTimeout, true)
+	require.NoError(t, err)
+	whenBroadcast := clockMock.Now()
+	broadcastMessages(t, newRoundMessages, pocketNodes)
+	finishedBroadcast := uint64(clockMock.Now().Sub(whenBroadcast).Milliseconds())
+	beforePrepareTimeout := time.Duration((paceMakerMinBlockTimeMsec - finishedBroadcast) * uint64(time.Millisecond))
+
+	step := typesCons.HotstuffStep(consMod.CurrentStep())
+	require.Equal(t, consensus.NewRound, step)
+
+	advanceTime(t, clockMock, clock.Duration(beforePrepareTimeout))
+	step = typesCons.HotstuffStep(consMod.CurrentStep())
+	// Should still be blocking proposal step
+	require.Equal(t, consensus.NewRound, step)
+
+	//advanceTime(t, clockMock, 8000*time.Millisecond)
+	//step = typesCons.HotstuffStep(consMod.CurrentStep())
+	//require.Equal(t, consensus.Prepare, step)
+}
+
 // 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,11 @@ 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) {
+		doPrepare := m.paceMaker.ProcessDelayedBlockPrepare()
+		if !doPrepare {
+			m.logger.Info().Msg("skip prepare new block")
+			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,7 @@ type Pacemaker interface {
 	PacemakerDebug
 
 	ShouldHandleMessage(message *typesCons.HotstuffMessage) (bool, error)
+	ProcessDelayedBlockPrepare() bool
 
 	RestartTimer()
 	NewHeight()
@@ -48,9 +50,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
+	latestPrepareRequest latestPrepareRequest
 
 	// Only used for development and debugging.
 	debug pacemakerDebug
@@ -60,6 +63,15 @@ type pacemaker struct {
 	logPrefix string
 }
 
+// Structure to handle delaying block preparation (reaping the block mempool)
+type latestPrepareRequest struct {
+	m              sync.Mutex
+	ch             chan bool
+	cancelFunc     context.CancelFunc
+	blockProposed  bool
+	deadlinePassed bool
+}
+
 func CreatePacemaker(bus modules.Bus, options ...modules.ModuleOption) (modules.Module, error) {
 	return new(pacemaker).Create(bus, options...)
 }
@@ -85,13 +97,21 @@ func (*pacemaker) Create(bus modules.Bus, options ...modules.ModuleOption) (modu
 		debugTimeBetweenStepsMsec: m.pacemakerCfg.GetDebugTimeBetweenStepsMsec(),
 		quorumCertificate:         nil,
 	}
+	m.latestPrepareRequest = latestPrepareRequest{
+		m:              sync.Mutex{},
+		ch:             nil,
+		cancelFunc:     nil,
+		blockProposed:  false,
+		deadlinePassed: false,
+	}
 
 	return m, nil
 }
 
 func (m *pacemaker) Start() error {
 	m.logger = logger.Global.CreateLoggerForModule(m.GetModuleName())
 	m.RestartTimer()
+	m.RestartBlockProposalTimer()
 	return nil
 }
 
@@ -171,6 +191,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()
 	}
@@ -191,6 +212,45 @@ func (m *pacemaker) RestartTimer() {
 	}()
 }
 
+func (m *pacemaker) RestartBlockProposalTimer() {
+	m.latestPrepareRequest.m.Lock()
+	defer m.latestPrepareRequest.m.Unlock()
+
+	if m.latestPrepareRequest.ch != nil {
+		m.latestPrepareRequest.ch <- false
+	}
+
+	if m.latestPrepareRequest.cancelFunc != nil {
+		m.latestPrepareRequest.cancelFunc()
+	}
+
+	m.latestPrepareRequest.blockProposed = false
+	m.latestPrepareRequest.deadlinePassed = false
+
+	ctx, cancel := context.WithCancel(context.TODO())
+	m.latestPrepareRequest.cancelFunc = cancel
+	clock := m.GetBus().GetRuntimeMgr().GetClock()
+	minBlockTime := time.Duration(m.pacemakerCfg.MinBlockTimeMsec * uint64(time.Millisecond))
+
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-clock.After(minBlockTime):
+			m.latestPrepareRequest.m.Lock()
+			defer m.latestPrepareRequest.m.Unlock()
+
+			if m.latestPrepareRequest.ch != nil {
+				m.latestPrepareRequest.blockProposed = true
+				m.latestPrepareRequest.ch <- true
+				m.latestPrepareRequest.ch = nil
+			}
+
+			m.latestPrepareRequest.deadlinePassed = true
+		}
+	}()
+}
+
 func (m *pacemaker) InterruptRound(reason string) {
 	defer m.RestartTimer()
 
@@ -225,6 +285,7 @@ func (m *pacemaker) InterruptRound(reason string) {
 
 func (m *pacemaker) NewHeight() {
 	defer m.RestartTimer()
+	defer m.RestartBlockProposalTimer()
 
 	consensusMod := m.GetBus().GetConsensusModule()
 	consensusMod.ResetRound(true)
@@ -243,6 +304,29 @@ func (m *pacemaker) NewHeight() {
 		)
 }
 
+func (m *pacemaker) ProcessDelayedBlockPrepare() bool {
+	m.latestPrepareRequest.m.Lock()
+
+	if m.latestPrepareRequest.blockProposed {
+		return false
+	}
+
+	if m.latestPrepareRequest.deadlinePassed {
+		return true
+	}
+
+	// there is already a block preparer candidate, we cancel it and start a new one
+	// DISCUSS: This is needed if we want to get the latest QC for the block proposal.
+	if m.latestPrepareRequest.ch != nil {
+		m.latestPrepareRequest.ch <- false
+	}
+
+	m.latestPrepareRequest.ch = make(chan bool)
+	m.latestPrepareRequest.m.Unlock()
+
+	return <-m.latestPrepareRequest.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;
 }

runtime/defaults/defaults.go

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

runtime/docs/CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- Add a new MinBlockTimeMsec config field to consensus
+
 ## [0.0.0.44] - 2023-06-26
 
 - Add a new ServiceConfig field to servicer config

runtime/manager_test.go

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