Merge branch 'main' into connection

samliok · web-flow · commit e710a928cfff · 2025-02-18T12:14:39.000-08:00
diff --git a/epoch.go b/epoch.go
@@ -1246,6 +1246,9 @@ func (e *Epoch) createBlockVerificationTask(block Block, from NodeID, vote Vote)
 
 		record := BlockRecord(md, block.Bytes())
 		e.WAL.Append(record)
+		e.Logger.Debug("Persisted block to WAL",
+			zap.Uint64("round", md.Round),
+			zap.Stringer("digest", md.Digest))
 
 		if !e.storeProposal(block) {
 			e.Logger.Warn("Unable to store proposed block for the round", zap.Stringer("NodeID", from), zap.Uint64("round", md.Round))
@@ -1556,6 +1559,9 @@ func (e *Epoch) triggerProposalWaitTimeExpired(round uint64) {
 		e.Logger.Error("Failed appending empty vote", zap.Error(err))
 		return
 	}
+	e.Logger.Debug("Persisted empty vote to WAL",
+		zap.Uint64("round", round),
+		zap.Int("size", len(emptyVoteRecord)))
 
 	emptyVotes := e.getOrCreateEmptyVoteSetForRound(round)
 	emptyVotes.timedOut = true
diff --git a/epoch_failover_test.go b/epoch_failover_test.go
@@ -16,26 +16,17 @@ import (
 )
 
 func TestEpochLeaderFailover(t *testing.T) {
-	timeoutDetected := make(chan struct{})
-
 	l := testutil.MakeLogger(t, 1)
-	l.Intercept(func(entry zapcore.Entry) error {
-		if entry.Message == `Timed out on block agreement` {
-			close(timeoutDetected)
-		}
-		return nil
-	})
 
 	bb := &testBlockBuilder{out: make(chan *testBlock, 1), blockShouldBeBuilt: make(chan struct{}, 1)}
 	storage := newInMemStorage()
 
 	nodes := []NodeID{{1}, {2}, {3}, {4}}
 	quorum := Quorum(len(nodes))
 
-	start := time.Now()
-
 	wal := newTestWAL(t)
 
+	start := time.Now()
 	conf := EpochConfig{
 		MaxProposalWait:     DefaultMaxProposalWaitTime,
 		StartTime:           start,
@@ -61,20 +52,20 @@ func TestEpochLeaderFailover(t *testing.T) {
 	// Then, don't do anything and wait for our node
 	// to start complaining about a block not being notarized
 
-	for _, round := range []uint64{0, 1, 2} {
+	for round := uint64(0); round < 3; round++ {
 		notarizeAndFinalizeRound(t, nodes, round, round, e, bb, quorum, storage, false)
 	}
 
 	bb.blockShouldBeBuilt <- struct{}{}
 
-	waitForEvent(t, start, e, timeoutDetected)
+	waitForBlockProposerTimeout(t, e, start)
 
 	lastBlock, _, ok := storage.Retrieve(storage.Height() - 1)
 	require.True(t, ok)
 
 	prev := lastBlock.BlockHeader().Digest
 
-	md := ProtocolMetadata{
+	emptyBlockMd := ProtocolMetadata{
 		Round: 3,
 		Seq:   2,
 		Prev:  prev,
@@ -83,8 +74,8 @@ func TestEpochLeaderFailover(t *testing.T) {
 	nextBlockSeqToCommit := uint64(3)
 	nextRoundToCommit := uint64(4)
 
-	emptyVoteFrom1 := createEmptyVote(md, nodes[1])
-	emptyVoteFrom2 := createEmptyVote(md, nodes[2])
+	emptyVoteFrom1 := createEmptyVote(emptyBlockMd, nodes[1])
+	emptyVoteFrom2 := createEmptyVote(emptyBlockMd, nodes[2])
 
 	e.HandleMessage(&Message{
 		EmptyVoteMessage: emptyVoteFrom1,
@@ -93,25 +84,25 @@ func TestEpochLeaderFailover(t *testing.T) {
 		EmptyVoteMessage: emptyVoteFrom2,
 	}, nodes[2])
 
-	// Ensure our node proposes block with sequence 3 for round 4
-	notarizeAndFinalizeRound(t, nodes, nextRoundToCommit, nextBlockSeqToCommit, e, bb, quorum, storage, false)
-
-	// WAL must contain an empty vote and an empty block.
+	wal.lock.Lock()
 	walContent, err := wal.ReadAll()
 	require.NoError(t, err)
+	wal.lock.Unlock()
 
-	// WAL should be: [..., <empty vote>, <empty block>, <notarization for 4>, <block3>]
-	rawEmptyVote, rawEmptyNotarization := walContent[len(walContent)-4], walContent[len(walContent)-3]
-
+	rawEmptyVote, rawEmptyNotarization := walContent[len(walContent)-2], walContent[len(walContent)-1]
 	emptyVote, err := ParseEmptyVoteRecord(rawEmptyVote)
 	require.NoError(t, err)
-	require.Equal(t, createEmptyVote(md, nodes[0]).Vote, emptyVote)
+	require.Equal(t, createEmptyVote(emptyBlockMd, nodes[0]).Vote, emptyVote)
 
 	emptyNotarization, err := EmptyNotarizationFromRecord(rawEmptyNotarization, &testQCDeserializer{t: t})
 	require.NoError(t, err)
 	require.Equal(t, emptyVoteFrom1.Vote, emptyNotarization.Vote)
 	require.Equal(t, uint64(3), emptyNotarization.Vote.Round)
 	require.Equal(t, uint64(2), emptyNotarization.Vote.Seq)
+	require.Equal(t, uint64(3), storage.Height())
+
+	// Ensure our node proposes block with sequence 3 for round 4
+	notarizeAndFinalizeRound(t, nodes, nextRoundToCommit, nextBlockSeqToCommit, e, bb, quorum, storage, false)
 	require.Equal(t, uint64(4), storage.Height())
 }
 
@@ -141,10 +132,9 @@ func TestEpochLeaderFailoverAfterProposal(t *testing.T) {
 	nodes := []NodeID{{1}, {2}, {3}, {4}}
 	quorum := Quorum(len(nodes))
 
-	start := time.Now()
-
 	wal := newTestWAL(t)
 
+	start := time.Now()
 	conf := EpochConfig{
 		MaxProposalWait:     DefaultMaxProposalWaitTime,
 		StartTime:           start,
@@ -194,7 +184,7 @@ func TestEpochLeaderFailoverAfterProposal(t *testing.T) {
 
 	bb.blockShouldBeBuilt <- struct{}{}
 
-	waitForEvent(t, start, e, timeoutDetected)
+	waitForBlockProposerTimeout(t, e, start)
 
 	for i := 1; i < quorum; i++ {
 		// Skip the vote of the block proposer
@@ -204,7 +194,7 @@ func TestEpochLeaderFailoverAfterProposal(t *testing.T) {
 		injectTestVote(t, e, block, nodes[i])
 	}
 
-	waitForEvent(t, start, e, alreadyTimedOut)
+	waitForBlockProposerTimeout(t, e, start)
 
 	lastBlock, _, ok := storage.Retrieve(storage.Height() - 1)
 	require.True(t, ok)
@@ -254,27 +244,17 @@ func TestEpochLeaderFailoverAfterProposal(t *testing.T) {
 }
 
 func TestEpochLeaderFailoverTwice(t *testing.T) {
-	timeoutDetected := make(chan struct{})
-
 	l := testutil.MakeLogger(t, 1)
-	l.Intercept(func(entry zapcore.Entry) error {
-		if entry.Message == `Timed out on block agreement` {
-			close(timeoutDetected)
-			timeoutDetected = make(chan struct{})
-		}
-		return nil
-	})
 
 	bb := &testBlockBuilder{out: make(chan *testBlock, 1), blockShouldBeBuilt: make(chan struct{}, 1)}
 	storage := newInMemStorage()
 
 	nodes := []NodeID{{1}, {2}, {3}, {4}}
 	quorum := Quorum(len(nodes))
 
-	start := time.Now()
-
 	wal := newTestWAL(t)
 
+	start := time.Now()
 	conf := EpochConfig{
 		MaxProposalWait:     DefaultMaxProposalWaitTime,
 		StartTime:           start,
@@ -302,7 +282,7 @@ func TestEpochLeaderFailoverTwice(t *testing.T) {
 
 	bb.blockShouldBeBuilt <- struct{}{}
 
-	waitForEvent(t, start, e, timeoutDetected)
+	waitForBlockProposerTimeout(t, e, start)
 
 	lastBlock, _, ok := storage.Retrieve(storage.Height() - 1)
 	require.True(t, ok)
@@ -331,7 +311,7 @@ func TestEpochLeaderFailoverTwice(t *testing.T) {
 
 	bb.blockShouldBeBuilt <- struct{}{}
 
-	waitForEvent(t, start, e, timeoutDetected)
+	waitForBlockProposerTimeout(t, e, start)
 
 	md = ProtocolMetadata{
 		Round: 3,
@@ -387,17 +367,18 @@ func createEmptyVote(md ProtocolMetadata, signer NodeID) *EmptyVote {
 	return emptyVoteFrom2
 }
 
-func waitForEvent(t *testing.T, start time.Time, e *Epoch, events chan struct{}) {
-	now := start
+func waitForBlockProposerTimeout(t *testing.T, e *Epoch, startTime time.Time) {
+	startRound := e.Metadata().Round
 	timeout := time.NewTimer(time.Minute)
 	defer timeout.Stop()
 
 	for {
-		now = now.Add(e.EpochConfig.MaxProposalWait / 5)
-		e.AdvanceTime(now)
-		select {
-		case <-events:
+		if e.WAL.(*testWAL).containsEmptyVote(startRound) {
 			return
+		}
+		startTime = startTime.Add(e.EpochConfig.MaxProposalWait / 5)
+		e.AdvanceTime(startTime)
+		select {
 		case <-time.After(time.Millisecond * 10):
 			continue
 		case <-timeout.C:
diff --git a/epoch_multinode_test.go b/epoch_multinode_test.go
@@ -205,6 +205,27 @@ func (tw *testWAL) assertNotarization(round uint64) {
 
 }
 
+func (tw *testWAL) containsEmptyVote(round uint64) bool {
+	tw.lock.Lock()
+	defer tw.lock.Unlock()
+
+	rawRecords, err := tw.WriteAheadLog.ReadAll()
+	require.NoError(tw.t, err)
+
+	for _, rawRecord := range rawRecords {
+		if binary.BigEndian.Uint16(rawRecord[:2]) == record.EmptyVoteRecordType {
+			vote, err := ParseEmptyVoteRecord(rawRecord)
+			require.NoError(tw.t, err)
+
+			if vote.Round == round {
+				return true
+			}
+		}
+	}
+
+	return false
+}
+
 type testComm struct {
 	from NodeID
 	net  *inMemNetwork
