Full Replication Logic Pt 2

api.go

@@ -57,7 +57,6 @@ type Storage interface {
 }
 
 type Communication interface {
-
 	// Nodes returns all nodes that participate in the epoch.
 	Nodes() []NodeID
 

block_scheduler_test.go

@@ -18,24 +18,6 @@ const (
 	defaultWaitDuration = 500 * time.Millisecond
 )
 
-func waitNoReceive(t *testing.T, ch <-chan struct{}) {
-	select {
-	case <-ch:
-		t.Fatal("channel unexpectedly signaled")
-	case <-time.After(defaultWaitDuration):
-		// good
-	}
-}
-
-func waitReceive(t *testing.T, ch <-chan struct{}) {
-	select {
-	case <-ch:
-		// good
-	case <-time.After(defaultWaitDuration):
-		t.Fatal("timed out waiting for signal")
-	}
-}
-
 func TestBlockVerificationScheduler(t *testing.T) {
 	t.Run("Schedules immediately when no dependencies", func(t *testing.T) {
 		scheduler := simplex.NewScheduler(testutil.MakeLogger(t), defaultMaxDeps)

epoch_multinode_test.go

@@ -4,6 +4,7 @@
 package simplex_test
 
 import (
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -155,10 +156,6 @@ func TestSimplexMultiNodeBlacklist(t *testing.T) {
 	testutil.NewSimplexNode(t, nodes[2], net, testEpochConfig)
 	testutil.NewSimplexNode(t, nodes[3], net, testEpochConfig)
 
-	for _, n := range net.Instances[:3] {
-		n.Silence()
-	}
-
 	net.StartInstances()
 
 	// Advance to the fourth node's turn by building three blocks
@@ -353,6 +350,9 @@ func TestSplitVotes(t *testing.T) {
 		}
 	}
 
+	// allow outstanding messages to be dropped
+	time.Sleep(100 * time.Millisecond)
+
 	net.SetAllNodesMessageFilter(testutil.AllowAllMessages)
 
 	time2 := splitNode2.E.StartTime
@@ -374,7 +374,7 @@ func TestSplitVotes(t *testing.T) {
 	splitNode3.WAL.AssertNotarization(0)
 
 	for _, n := range net.Instances {
-		require.Equal(t, uint64(0), n.Storage.NumBlocks())
+		require.Equal(t, uint64(0), n.Storage.NumBlocks(), fmt.Sprintf("node %s should not have", n.E.ID))
 		require.Equal(t, uint64(1), n.E.Metadata().Round)
 		require.Equal(t, uint64(1), n.E.Metadata().Seq)
 	}

epoch_test.go

@@ -1263,14 +1263,12 @@ func TestEpochVotesForEquivocatedVotes(t *testing.T) {
 	equivocatedBlock.Data = []byte{1, 2, 3}
 	equivocatedBlock.ComputeDigest()
 	testutil.InjectTestVote(t, e, equivocatedBlock, nodes[1])
-	eqbh := equivocatedBlock.BlockHeader()
 
 	// We should not have sent a notarization yet, since we have not received enough votes for the block we received from the leader
 	require.Never(t, func() bool {
 		select {
 		case msg := <-recordedMessages:
 			if msg.Notarization != nil {
-				fmt.Println(msg.Notarization.Vote.BlockHeader.Equals(&eqbh))
 				return true
 			}
 		default:
@@ -1301,6 +1299,52 @@ func TestEpochVotesForEquivocatedVotes(t *testing.T) {
 	}
 }
 
+// Ensures we don't double increment the round on persisting a notarization
+func TestDoubleIncrementOnPersistNotarization(t *testing.T) {
+	// add an empty notarization, then a notarization for a previous round
+	bb := &testutil.TestBlockBuilder{Out: make(chan *testutil.TestBlock, 1)}
+	nodes := []NodeID{{1}, {2}, {3}, {4}}
+	conf, wal, _ := testutil.DefaultTestNodeEpochConfig(t, nodes[3], testutil.NewNoopComm(nodes), bb)
+	conf.ReplicationEnabled = true
+
+	e, err := NewEpoch(conf)
+	require.NoError(t, err)
+
+	require.NoError(t, e.Start())
+
+	advanceRoundFromEmpty(t, e)
+	require.Equal(t, uint64(1), e.Metadata().Round)
+
+	// create a notarization for round 0
+	md := ProtocolMetadata{
+		Epoch: 0,
+		Round: 0,
+		Seq:   0,
+	}
+	_, ok := bb.BuildBlock(context.Background(), md, emptyBlacklist)
+	require.True(t, ok)
+
+	block := <-bb.Out
+	notarization, err := testutil.NewNotarization(conf.Logger, conf.SignatureAggregator, block, nodes)
+	require.NoError(t, err)
+
+	err = e.HandleMessage(&Message{
+		ReplicationResponse: &ReplicationResponse{
+			Data: []QuorumRound{
+				{
+					Block:        block,
+					Notarization: &notarization,
+				},
+			},
+		},
+	}, nodes[0])
+	require.NoError(t, err)
+
+	wal.AssertWALSize(2)
+	// ensure the round is still 1
+	require.Equal(t, uint64(1), e.Metadata().Round)
+}
+
 // ListnerComm is a comm that listens for incoming messages
 // and sends them to the [in] channel
 type listenerComm struct {

msg.go

@@ -223,18 +223,22 @@ type QuorumCertificate interface {
 }
 
 type ReplicationRequest struct {
-	Seqs        []uint64 // sequences we are requesting
-	LatestRound uint64   // latest round that we are aware of
+	Seqs               []uint64 // sequences we are requesting
+	Rounds             []uint64 // rounds we are requesting
+	LatestRound        uint64   // latest round that we are aware of
+	LatestFinalizedSeq uint64   // latest finalized sequence that we are aware of
 }
 
 type ReplicationResponse struct {
 	Data        []QuorumRound
 	LatestRound *QuorumRound
+	LatestSeq   *QuorumRound
 }
 
 type VerifiedReplicationResponse struct {
-	Data        []VerifiedQuorumRound
-	LatestRound *VerifiedQuorumRound
+	Data               []VerifiedQuorumRound
+	LatestRound        *VerifiedQuorumRound
+	LatestFinalizedSeq *VerifiedQuorumRound
 }
 
 // QuorumRound represents a round that has achieved quorum on either
@@ -250,13 +254,15 @@ type QuorumRound struct {
 // (block, notarization) or (block, finalization) or
 // (empty notarization)
 func (q *QuorumRound) IsWellFormed() error {
-	if q.EmptyNotarization != nil && q.Block == nil {
-		return nil
-	} else if q.Block != nil && (q.Notarization != nil || q.Finalization != nil) {
-		return nil
+	if q.Block == nil && q.EmptyNotarization == nil {
+		return fmt.Errorf("malformed QuorumRound, empty block and notarization fields")
+	}
+
+	if q.Block != nil && (q.Notarization == nil && q.Finalization == nil) {
+		return fmt.Errorf("malformed QuorumRound, block but no notarization or finalization")
 	}
 
-	return fmt.Errorf("malformed QuorumRound")
+	return nil
 }
 
 func (q *QuorumRound) GetRound() uint64 {
@@ -284,10 +290,15 @@ func (q *QuorumRound) VerifyQCConsistentWithBlock() error {
 		return err
 	}
 
-	if q.EmptyNotarization != nil {
+	if q.Block == nil {
 		return nil
 	}
 
+	// if an empty notarization is included, ensure the round is equal to the block round
+	if q.EmptyNotarization != nil && q.EmptyNotarization.Vote.Round != q.Block.BlockHeader().Round {
+		return fmt.Errorf("empty round does not match block round")
+	}
+
 	// ensure the finalization or notarization we get relates to the block
 	blockDigest := q.Block.BlockHeader().Digest
 
@@ -314,7 +325,7 @@ func (q *QuorumRound) String() string {
 		if err != nil {
 			return fmt.Sprintf("QuorumRound{Error: %s}", err)
 		} else {
-			return fmt.Sprintf("QuorumRound{Round: %d, Seq: %d}", q.GetRound(), q.GetSequence())
+			return fmt.Sprintf("QuorumRound{Round: %d, Seq: %d, Finalized: %t}", q.GetRound(), q.GetSequence(), q.Finalization != nil)
 		}
 	}
 

msg_test.go

@@ -82,6 +82,24 @@ func TestQuorumRoundMalformed(t *testing.T) {
 			},
 			expectedErr: true,
 		},
+		{
+			name: "block and notarization and empty notarization",
+			qr: simplex.QuorumRound{
+				Block:             &testutil.TestBlock{},
+				Notarization:      &simplex.Notarization{},
+				EmptyNotarization: &simplex.EmptyNotarization{},
+			},
+			expectedErr: false,
+		},
+		{
+			name: "block and finalization and empty notarization",
+			qr: simplex.QuorumRound{
+				Block:             &testutil.TestBlock{},
+				Finalization:      &simplex.Finalization{},
+				EmptyNotarization: &simplex.EmptyNotarization{},
+			},
+			expectedErr: false,
+		},
 	}
 
 	for _, test := range tests {

pos_test.go

@@ -5,12 +5,12 @@ package simplex_test
 
 import (
 	"bytes"
-	"fmt"
+	"testing"
+	"time"
+
 	"github.com/ava-labs/simplex"
 	"github.com/ava-labs/simplex/testutil"
 	"github.com/stretchr/testify/require"
-	"testing"
-	"time"
 )
 
 func TestPoS(t *testing.T) {
@@ -133,8 +133,6 @@ func TestPoS(t *testing.T) {
 		testutil.WaitToEnterRound(t, n.E, 15)
 	}
 
-	fmt.Println(simplex.LeaderForRound(nodes, 15))
-
 	// Now, disconnect the node with the highest stake (node 3) and observe the network is stuck
 	net.Disconnect(nodes[2])
 	net.TriggerLeaderBlockBuilder(15)