Re-broadcast dropped finalize votes (#279)

* Re-broadcast dropped finalize votes

This commit makes the Simplex instance be able to discover when it has a round that isn't finalized and remains not finalized.

If that happens, Simplex will trigger re-broadcasting of finalize votes.

Signed-off-by: Yacov Manevich <[email protected]>

* address nits

Signed-off-by: Yacov Manevich <[email protected]>

---------

Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

epoch.go

@@ -28,10 +28,11 @@ const (
 	DefaultMaxRoundWindow   = 10
 	DefaultMaxPendingBlocks = 20
 
-	DefaultMaxProposalWaitTime         = 5 * time.Second
-	DefaultReplicationRequestTimeout   = 5 * time.Second
-	DefaultEmptyVoteRebroadcastTimeout = 5 * time.Second
-	EmptyVoteTimeoutID                 = "rebroadcast_empty_vote"
+	DefaultMaxProposalWaitTime            = 5 * time.Second
+	DefaultReplicationRequestTimeout      = 5 * time.Second
+	DefaultEmptyVoteRebroadcastTimeout    = 5 * time.Second
+	DefaultFinalizeVoteRebroadcastTimeout = 6 * time.Second
+	EmptyVoteTimeoutID                    = "rebroadcast_empty_vote"
 )
 
 type EmptyVoteSet struct {
@@ -59,22 +60,23 @@ func NewRound(block VerifiedBlock) *Round {
 }
 
 type EpochConfig struct {
-	MaxProposalWait     time.Duration
-	MaxRebroadcastWait  time.Duration
-	QCDeserializer      QCDeserializer
-	Logger              Logger
-	ID                  NodeID
-	Signer              Signer
-	Verifier            SignatureVerifier
-	BlockDeserializer   BlockDeserializer
-	SignatureAggregator SignatureAggregator
-	Comm                Communication
-	Storage             Storage
-	WAL                 WriteAheadLog
-	BlockBuilder        BlockBuilder
-	Epoch               uint64
-	StartTime           time.Time
-	ReplicationEnabled  bool
+	MaxProposalWait            time.Duration
+	MaxRebroadcastWait         time.Duration
+	FinalizeRebroadcastTimeout time.Duration
+	QCDeserializer             QCDeserializer
+	Logger                     Logger
+	ID                         NodeID
+	Signer                     Signer
+	Verifier                   SignatureVerifier
+	BlockDeserializer          BlockDeserializer
+	SignatureAggregator        SignatureAggregator
+	Comm                       Communication
+	Storage                    Storage
+	WAL                        WriteAheadLog
+	BlockBuilder               BlockBuilder
+	Epoch                      uint64
+	StartTime                  time.Time
+	ReplicationEnabled         bool
 }
 
 type Epoch struct {
@@ -94,6 +96,7 @@ type Epoch struct {
 	quorumSize                     int
 	rounds                         map[uint64]*Round
 	emptyVotes                     map[uint64]*EmptyVoteSet
+	oldestNotFinalizedNotarization NotarizationTime
 	futureMessages                 messagesFromNode
 	round                          uint64 // The current round we notarize
 	maxRoundWindow                 uint64
@@ -119,6 +122,7 @@ func (e *Epoch) AdvanceTime(t time.Time) {
 	e.monitor.AdvanceTime(t)
 	e.replicationState.AdvanceTime(t)
 	e.timeoutHandler.Tick(t)
+	e.oldestNotFinalizedNotarization.CheckForNotFinalizedNotarizedBlocks(t)
 }
 
 // HandleMessage notifies the engine about a reception of a message.
@@ -174,6 +178,8 @@ func (e *Epoch) HandleMessage(msg *Message, from NodeID) error {
 }
 
 func (e *Epoch) init() error {
+	e.maybeAssignDefaultConfig()
+	e.initOldestNotFinalizedNotarization()
 	e.oneTimeVerifier = newOneTimeVerifier(e.Logger)
 	e.sched = NewScheduler(e.Logger)
 	e.monitor = NewMonitor(e.StartTime, e.Logger)
@@ -211,6 +217,40 @@ func (e *Epoch) init() error {
 	return e.setMetadataFromStorage()
 }
 
+func (e *Epoch) initOldestNotFinalizedNotarization() {
+	rebroadcastFinalizationVotes := func() {
+		e.lock.Lock()
+		defer e.lock.Unlock()
+
+		if err := e.rebroadcastPastFinalizeVotes(); err != nil {
+			e.Logger.Error("Could not rebroadcast past finalization votes", zap.Error(err))
+		}
+	}
+	e.oldestNotFinalizedNotarization = NewNotarizationTime(
+		e.FinalizeRebroadcastTimeout,
+		e.haveNotFinalizedNotarizedRound,
+		rebroadcastFinalizationVotes, e.getRound)
+}
+
+func (e *Epoch) getRound() uint64 {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+
+	return e.round
+}
+
+func (e *Epoch) maybeAssignDefaultConfig() {
+	if e.FinalizeRebroadcastTimeout == 0 {
+		e.FinalizeRebroadcastTimeout = DefaultFinalizeVoteRebroadcastTimeout
+	}
+	if e.MaxProposalWait == 0 {
+		e.MaxProposalWait = DefaultMaxProposalWaitTime
+	}
+	if e.MaxRebroadcastWait == 0 {
+		e.MaxRebroadcastWait = DefaultEmptyVoteRebroadcastTimeout
+	}
+}
+
 func (e *Epoch) Start() error {
 	if e.canReceiveMessages.Load() {
 		return ErrAlreadyStarted
@@ -2688,6 +2728,24 @@ func (e *Epoch) locateQuorumRecord(seq uint64) *VerifiedQuorumRound {
 	}
 }
 
+func (e *Epoch) haveNotFinalizedNotarizedRound() (uint64, bool) {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+
+	var minRoundNum uint64
+	var found bool
+	for _, round := range e.rounds {
+		if !found {
+			minRoundNum = round.num
+			found = true
+		} else if round.num < minRoundNum {
+			minRoundNum = round.num
+		}
+	}
+
+	return minRoundNum, found
+}
+
 func (e *Epoch) handleReplicationResponse(resp *ReplicationResponse, from NodeID) error {
 	if !e.ReplicationEnabled {
 		return nil

epoch_multinode_test.go

@@ -4,13 +4,120 @@
 package simplex_test
 
 import (
-	"testing"
-
 	"github.com/ava-labs/simplex"
 	"github.com/ava-labs/simplex/testutil"
 	"github.com/stretchr/testify/require"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
 )
 
+func TestSimplexRebroadcastFinalizationVotes(t *testing.T) {
+	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
+	net := testutil.NewInMemNetwork(t, nodes)
+
+	var allowFinalizeVotes atomic.Bool
+
+	var numFinalizeVotesSent atomic.Uint32
+
+	config := func(from simplex.NodeID) *testutil.TestNodeConfig {
+		return &testutil.TestNodeConfig{
+			Comm: testutil.NewTestComm(from, net, func(msg *simplex.Message, from simplex.NodeID, to simplex.NodeID) bool {
+				if msg.Finalization != nil && !allowFinalizeVotes.Load() {
+					return false
+				}
+				if allowFinalizeVotes.Load() && msg.FinalizeVote != nil {
+					numFinalizeVotesSent.Add(1)
+				}
+				return allowFinalizeVotes.Load() || msg.FinalizeVote == nil
+			}),
+		}
+	}
+
+	testutil.NewSimplexNode(t, nodes[0], net, config(nodes[0]))
+	testutil.NewSimplexNode(t, nodes[1], net, config(nodes[1]))
+	testutil.NewSimplexNode(t, nodes[2], net, config(nodes[2]))
+	testutil.NewSimplexNode(t, nodes[3], net, config(nodes[3]))
+
+	net.StartInstances()
+
+	lastSeq := uint64(9)
+
+	for seq := uint64(0); seq <= lastSeq; seq++ {
+		for _, n := range net.Instances {
+			testutil.WaitToEnterRound(t, n.E, seq)
+		}
+		net.TriggerLeaderBlockBuilder(seq)
+		for _, n := range net.Instances {
+			n.WAL.AssertNotarization(seq)
+		}
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(len(net.Instances))
+
+	for _, n := range net.Instances {
+		go func(n *testutil.TestNode) {
+			defer wg.Done()
+			n.Storage.EnsureNoBlockCommit(t, 0)
+		}(n)
+	}
+
+	wg.Wait()
+
+	allowFinalizeVotes.Store(true)
+
+	require.Eventually(t, func() bool {
+		var allHaveFinalized bool = true
+		for _, n := range net.Instances {
+			if n.Storage.NumBlocks() < lastSeq+1 {
+				allHaveFinalized = false
+				break
+			}
+		}
+
+		if allHaveFinalized {
+			return true
+		}
+
+		for _, n := range net.Instances {
+			n.AdvanceTime(simplex.DefaultFinalizeVoteRebroadcastTimeout / 3)
+		}
+
+		return false
+
+	}, time.Second*10, time.Millisecond*100)
+
+	// Close the recorded messages channel. A message sent to this channel will cause a panic.
+	finalizeVoteSentCount := numFinalizeVotesSent.Load()
+
+	// Advance the time to make sure we do not continue to send finalize votes.
+	for _, n := range net.Instances {
+		n.AdvanceTime(simplex.DefaultFinalizeVoteRebroadcastTimeout * 2)
+		n.AdvanceTime(simplex.DefaultFinalizeVoteRebroadcastTimeout * 2)
+		n.AdvanceTime(simplex.DefaultFinalizeVoteRebroadcastTimeout * 2)
+	}
+
+	require.Equal(t, finalizeVoteSentCount, numFinalizeVotesSent.Load(), "no more finalize votes should have been sent")
+
+	// Next, we run the nodes and notarize 100 blocks, and ensure that less than 400 finalize votes were sent.
+	previousVoteSentCount := finalizeVoteSentCount // We copy the value just to give it a better name.
+	for seq := lastSeq + 1; seq < lastSeq+101; seq++ {
+		for _, n := range net.Instances {
+			testutil.WaitToEnterRound(t, n.E, seq)
+		}
+		net.TriggerLeaderBlockBuilder(seq)
+		for _, n := range net.Instances {
+			n.Storage.WaitForBlockCommit(seq)
+			n.AdvanceTime(simplex.DefaultFinalizeVoteRebroadcastTimeout)
+		}
+	}
+
+	require.LessOrEqual(t, previousVoteSentCount+400, numFinalizeVotesSent.Load())
+
+}
+
 func TestSimplexMultiNodeSimple(t *testing.T) {
 	nodes := []simplex.NodeID{{1}, {2}, {3}, {4}}
 	net := testutil.NewInMemNetwork(t, nodes)

testutil/node.go

@@ -5,17 +5,20 @@ package testutil
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/ava-labs/simplex"
 	"github.com/stretchr/testify/require"
 )
 
 type TestNode struct {
-	WAL     *TestWAL
-	Storage *InMemStorage
-	E       *simplex.Epoch
-	ingress chan struct {
+	currentTime atomic.Int64
+	WAL         *TestWAL
+	Storage     *InMemStorage
+	E           *simplex.Epoch
+	ingress     chan struct {
 		msg  *simplex.Message
 		from simplex.NodeID
 	}
@@ -48,6 +51,8 @@ func NewSimplexNode(t *testing.T, nodeID simplex.NodeID, net *InMemNetwork, conf
 			from simplex.NodeID
 		}, 100)}
 
+	ti.currentTime.Store(epochConfig.StartTime.UnixMilli())
+
 	net.addNode(ti)
 	return ti
 }
@@ -87,6 +92,12 @@ func (t *TestNode) TriggerBlockShouldBeBuilt() {
 	}
 }
 
+func (t *TestNode) AdvanceTime(duration time.Duration) {
+	now := time.UnixMilli(t.currentTime.Load()).Add(duration)
+	t.currentTime.Store(now.UnixMilli())
+	t.E.AdvanceTime(now)
+}
+
 func (t *TestNode) Silence() {
 	t.l.Silence()
 }

testutil/util.go

@@ -18,20 +18,21 @@ func DefaultTestNodeEpochConfig(t *testing.T, nodeID simplex.NodeID, comm simple
 	storage := NewInMemStorage()
 	wal := NewTestWAL(t)
 	conf := simplex.EpochConfig{
-		MaxProposalWait:     simplex.DefaultMaxProposalWaitTime,
-		MaxRebroadcastWait:  simplex.DefaultEmptyVoteRebroadcastTimeout,
-		Comm:                comm,
-		Logger:              l,
-		ID:                  nodeID,
-		Signer:              &testSigner{},
-		WAL:                 wal,
-		Verifier:            &testVerifier{},
-		Storage:             storage,
-		BlockBuilder:        bb,
-		SignatureAggregator: &TestSignatureAggregator{},
-		BlockDeserializer:   &BlockDeserializer{},
-		QCDeserializer:      &testQCDeserializer{t: t},
-		StartTime:           time.Now(),
+		MaxProposalWait:            simplex.DefaultMaxProposalWaitTime,
+		MaxRebroadcastWait:         simplex.DefaultEmptyVoteRebroadcastTimeout,
+		FinalizeRebroadcastTimeout: simplex.DefaultFinalizeVoteRebroadcastTimeout,
+		Comm:                       comm,
+		Logger:                     l,
+		ID:                         nodeID,
+		Signer:                     &testSigner{},
+		WAL:                        wal,
+		Verifier:                   &testVerifier{},
+		Storage:                    storage,
+		BlockBuilder:               bb,
+		SignatureAggregator:        &TestSignatureAggregator{},
+		BlockDeserializer:          &BlockDeserializer{},
+		QCDeserializer:             &testQCDeserializer{t: t},
+		StartTime:                  time.Now(),
 	}
 	return conf, wal, storage
 }