Add replication support for lagging nodes (#70)

* added replicatoin

* separate out replication state

* all tests working

---------

Signed-off-by: Sam Liokumovich <65994425+samliok@users.noreply.github.com>

Author: samliok

epoch.go

@@ -20,43 +20,74 @@ import (
 func TestSimplexMultiNodeSimple(t *testing.T) {
 	bb := newTestControlledBlockBuilder(t)
 
-	var net inMemNetwork
-	net.nodes = []NodeID{{1}, {2}, {3}, {4}}
-
-	n1 := newSimplexNode(t, 1, &net, bb)
-	n2 := newSimplexNode(t, 2, &net, bb)
-	n3 := newSimplexNode(t, 3, &net, bb)
-	n4 := newSimplexNode(t, 4, &net, bb)
+	nodes := []NodeID{{1}, {2}, {3}, {4}}
+	net := newInMemNetwork(t, nodes)
+	newSimplexNode(t, nodes[0], net, bb, false)
+	newSimplexNode(t, nodes[1], net, bb, false)
+	newSimplexNode(t, nodes[2], net, bb, false)
+	newSimplexNode(t, nodes[3], net, bb, false)
 
 	bb.triggerNewBlock()
 
-	instances := []*testInstance{n4, n3, n2, n1}
-
-	for _, n := range instances {
-		n.start()
-	}
+	net.startInstances()
 
 	for seq := 0; seq < 10; seq++ {
-		for _, n := range instances {
-			n.ledger.waitForBlockCommit(uint64(seq))
+		for _, n := range net.instances {
+			n.storage.waitForBlockCommit(uint64(seq))
 		}
 		bb.triggerNewBlock()
 	}
 }
 
-func (t *testInstance) start() {
+func (t *testNode) start() {
 	go t.handleMessages()
 	require.NoError(t.t, t.e.Start())
 }
 
-func newSimplexNode(t *testing.T, id uint8, net *inMemNetwork, bb BlockBuilder) *testInstance {
-	l := testutil.MakeLogger(t, int(id))
-	storage := newInMemStorage()
+func newSimplexNodeWithStorage(t *testing.T, nodeID NodeID, net *inMemNetwork, bb BlockBuilder, storage []FinalizedBlock) *testNode {
+	wal := newTestWAL(t)
+	conf := defaultTestNodeEpochConfig(t, nodeID, net, wal, bb, true)
+	for _, data := range storage {
+		conf.Storage.Index(data.Block, data.FCert)
+	}
+	e, err := NewEpoch(conf)
+	require.NoError(t, err)
+	ti := &testNode{
+		wal:     wal,
+		e:       e,
+		t:       t,
+		storage: conf.Storage.(*InMemStorage),
+		ingress: make(chan struct {
+			msg  *Message
+			from NodeID
+		}, 100)}
 
-	nodeID := NodeID{id}
+	net.addNode(ti)
+	return ti
+}
 
+func newSimplexNode(t *testing.T, nodeID NodeID, net *inMemNetwork, bb BlockBuilder, replicationEnabled bool) *testNode {
 	wal := newTestWAL(t)
+	conf := defaultTestNodeEpochConfig(t, nodeID, net, wal, bb, replicationEnabled)
+	e, err := NewEpoch(conf)
+	require.NoError(t, err)
+	ti := &testNode{
+		wal:     wal,
+		e:       e,
+		t:       t,
+		storage: conf.Storage.(*InMemStorage),
+		ingress: make(chan struct {
+			msg  *Message
+			from NodeID
+		}, 100)}
+
+	net.addNode(ti)
+	return ti
+}
 
+func defaultTestNodeEpochConfig(t *testing.T, nodeID NodeID, net *inMemNetwork, wal WriteAheadLog, bb BlockBuilder, replicationEnabled bool) EpochConfig {
+	l := testutil.MakeLogger(t, int(nodeID[0]))
+	storage := newInMemStorage()
 	conf := EpochConfig{
 		MaxProposalWait: DefaultMaxProposalWaitTime,
 		Comm: &testComm{
@@ -71,29 +102,16 @@ func newSimplexNode(t *testing.T, id uint8, net *inMemNetwork, bb BlockBuilder)
 		Storage:             storage,
 		BlockBuilder:        bb,
 		SignatureAggregator: &testSignatureAggregator{},
+		BlockDeserializer:   &blockDeserializer{},
+		QCDeserializer:      &testQCDeserializer{t: t},
+		ReplicationEnabled:  replicationEnabled,
 	}
-
-	e, err := NewEpoch(conf)
-	require.NoError(t, err)
-
-	ti := &testInstance{
-		wal:    wal,
-		e:      e,
-		t:      t,
-		ledger: storage,
-		ingress: make(chan struct {
-			msg  *Message
-			from NodeID
-		}, 100)}
-
-	net.instances = append(net.instances, ti)
-
-	return ti
+	return conf
 }
 
-type testInstance struct {
+type testNode struct {
 	wal     *testWAL
-	ledger  *InMemStorage
+	storage *InMemStorage
 	e       *Epoch
 	ingress chan struct {
 		msg  *Message
@@ -102,13 +120,13 @@ type testInstance struct {
 	t *testing.T
 }
 
-func (t *testInstance) HandleMessage(msg *Message, from NodeID) error {
+func (t *testNode) HandleMessage(msg *Message, from NodeID) error {
 	err := t.e.HandleMessage(msg, from)
 	require.NoError(t.t, err)
 	return err
 }
 
-func (t *testInstance) handleMessages() {
+func (t *testNode) handleMessages() {
 	for msg := range t.ingress {
 		err := t.HandleMessage(msg.msg, msg.from)
 		require.NoError(t.t, err)
@@ -225,10 +243,46 @@ func (c *testComm) Broadcast(msg *Message) {
 }
 
 type inMemNetwork struct {
+	t         *testing.T
 	nodes     []NodeID
-	instances []*testInstance
+	instances []*testNode
+}
+
+// newInMemNetwork creates an in-memory network. Node IDs must be provided before
+// adding instances, as nodes require prior knowledge of all participants.
+func newInMemNetwork(t *testing.T, nodes []NodeID) *inMemNetwork {
+	net := &inMemNetwork{
+		t:         t,
+		nodes:     nodes,
+		instances: make([]*testNode, 0),
+	}
+	return net
+}
+
+func (n *inMemNetwork) addNode(node *testNode) {
+	allowed := false
+	for _, id := range n.nodes {
+		if bytes.Equal(id, node.e.ID) {
+			allowed = true
+			break
+		}
+	}
+	require.True(node.t, allowed, "node must be declared before adding")
+	n.instances = append(n.instances, node)
+}
+
+// startInstances starts all instances in the network.
+// The first one is typically the leader, so we make sure to start it last.
+func (n *inMemNetwork) startInstances() {
+	require.Equal(n.t, len(n.nodes), len(n.instances))
+
+	for i := len(n.nodes) - 1; i >= 0; i-- {
+		n.instances[i].start()
+	}
 }
 
+// testControlledBlockBuilder is a test block builder that blocks
+// block building until a trigger is received
 type testControlledBlockBuilder struct {
 	t       *testing.T
 	control chan struct{}

epoch_multinode_test.go

@@ -17,6 +17,8 @@ type Message struct {
 	Notarization            *Notarization
 	Finalization            *Finalization
 	FinalizationCertificate *FinalizationCertificate
+	ReplicationResponse     *ReplicationResponse
+	ReplicationRequest      *ReplicationRequest
 }
 
 type ToBeSignedEmptyVote struct {
@@ -206,3 +208,25 @@ type QuorumCertificate interface {
 	// Bytes returns a raw representation of the given QuorumCertificate.
 	Bytes() []byte
 }
+
+type ReplicationRequest struct {
+	FinalizationCertificateRequest *FinalizationCertificateRequest
+}
+
+type ReplicationResponse struct {
+	FinalizationCertificateResponse *FinalizationCertificateResponse
+}
+
+// request a finalization certificate for the given sequence number
+type FinalizationCertificateRequest struct {
+	Sequences []uint64
+}
+
+type FinalizedBlock struct {
+	Block Block
+	FCert FinalizationCertificate
+}
+
+type FinalizationCertificateResponse struct {
+	Data []FinalizedBlock
+}

msg.go

@@ -11,23 +11,28 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func newNotarizationRecord(logger simplex.Logger, signatureAggregator simplex.SignatureAggregator, block simplex.Block, ids []simplex.NodeID) ([]byte, error) {
+func newNotarization(logger simplex.Logger, signatureAggregator simplex.SignatureAggregator, block simplex.Block, ids []simplex.NodeID) (simplex.Notarization, error) {
 	votesForCurrentRound := make(map[string]*simplex.Vote)
 	for _, id := range ids {
 		vote, err := newTestVote(block, id)
 		if err != nil {
-			return nil, err
+			return simplex.Notarization{}, err
 		}
+
 		votesForCurrentRound[string(id)] = vote
 	}
 
 	notarization, err := simplex.NewNotarization(logger, signatureAggregator, votesForCurrentRound, block.BlockHeader())
+	return notarization, err
+}
+
+func newNotarizationRecord(logger simplex.Logger, signatureAggregator simplex.SignatureAggregator, block simplex.Block, ids []simplex.NodeID) ([]byte, error) {
+	notarization, err := newNotarization(logger, signatureAggregator, block, ids)
 	if err != nil {
 		return nil, err
 	}
 
 	record := simplex.NewQuorumRecord(notarization.QC.Bytes(), notarization.Vote.Bytes(), record.NotarizationRecordType)
-
 	return record, nil
 }
 

record_test.go

@@ -0,0 +1,125 @@
+// Copyright (C) 2019-2024, Ava Labs, Inc. All rights reserved.
+// See the file LICENSE for licensing terms.
+
+package simplex
+
+import (
+	"bytes"
+	"fmt"
+	"math"
+
+	"go.uber.org/zap"
+)
+
+type ReplicationState struct {
+	logger         Logger
+	enabled        bool
+	maxRoundWindow uint64
+	comm           Communication
+	id             NodeID
+
+	// latest seq requested
+	lastSequenceRequested uint64
+
+	// highest sequence we have received a finalization certificate for
+	highestFCertReceived *FinalizationCertificate
+
+	// received
+	receivedFinalizationCertificates map[uint64]FinalizedBlock
+}
+
+func NewReplicationState(logger Logger, comm Communication, id NodeID, maxRoundWindow uint64, enabled bool) *ReplicationState {
+	return &ReplicationState{
+		logger:                           logger,
+		enabled:                          enabled,
+		comm:                             comm,
+		id:                               id,
+		maxRoundWindow:                   maxRoundWindow,
+		receivedFinalizationCertificates: make(map[uint64]FinalizedBlock),
+	}
+}
+
+func (r *ReplicationState) collectFutureFinalizationCertificates(fCert *FinalizationCertificate, currentRound uint64, nextSeqToCommit uint64) {
+	if !r.enabled {
+		return
+	}
+	fCertRound := fCert.Finalization.Round
+	// Don't exceed the max round window
+	endSeq := math.Min(float64(fCertRound), float64(r.maxRoundWindow+currentRound))
+	if r.highestFCertReceived == nil || fCertRound > r.highestFCertReceived.Finalization.Seq {
+		r.highestFCertReceived = fCert
+	}
+	// Node is behind, but we've already sent messages to collect future fCerts
+	if r.lastSequenceRequested >= uint64(endSeq) {
+		return
+	}
+
+	startSeq := math.Max(float64(nextSeqToCommit), float64(r.lastSequenceRequested))
+	r.logger.Debug("Node is behind, requesting missing finalization certificates", zap.Uint64("round", fCertRound), zap.Uint64("startSeq", uint64(startSeq)), zap.Uint64("endSeq", uint64(endSeq)))
+	r.sendFutureCertficatesRequests(uint64(startSeq), uint64(endSeq))
+}
+
+// sendFutureCertficatesRequests sends requests for future finalization certificates for the
+// range of sequences [start, end] <- inclusive
+func (r *ReplicationState) sendFutureCertficatesRequests(start uint64, end uint64) {
+	seqs := make([]uint64, (end+1)-start)
+	for i := start; i <= end; i++ {
+		seqs[i-start] = i
+	}
+
+	roundRequest := &ReplicationRequest{
+		FinalizationCertificateRequest: &FinalizationCertificateRequest{
+			Sequences: seqs,
+		},
+	}
+	msg := &Message{ReplicationRequest: roundRequest}
+
+	requestFrom := r.requestFrom()
+
+	r.lastSequenceRequested = end
+	r.comm.SendMessage(msg, requestFrom)
+}
+
+// requestFrom returns a node to send a message request to
+// this is used to ensure that we are not sending a message to ourselves
+func (r *ReplicationState) requestFrom() NodeID {
+	nodes := r.comm.ListNodes()
+	for _, node := range nodes {
+		if !node.Equals(r.id) {
+			return node
+		}
+	}
+	return NodeID{}
+}
+
+// maybeCollectFutureFinalizationCertificates attempts to collect future finalization certificates if
+// there are more fCerts to be collected and the round has caught up.
+func (r *ReplicationState) maybeCollectFutureFinalizationCertificates(round uint64, nextSequenceToCommit uint64) {
+	if r.highestFCertReceived == nil {
+		return
+	}
+
+	// we send out more request once our round has caught up to 1/2 of the maxRoundWindow
+	if r.lastSequenceRequested >= r.highestFCertReceived.Finalization.Round {
+		return
+	}
+	if round+r.maxRoundWindow/2 > r.lastSequenceRequested {
+		r.collectFutureFinalizationCertificates(r.highestFCertReceived, round, nextSequenceToCommit)
+	}
+}
+
+func (r *ReplicationState) StoreFinalizedBlock(data FinalizedBlock) error {
+	// ensure the finalization certificate we get relates to the block
+	blockDigest := data.Block.BlockHeader().Digest
+	if !bytes.Equal(blockDigest[:], data.FCert.Finalization.Digest[:]) {
+		return fmt.Errorf("finalization certificate does not match the block")
+	}
+
+	// don't store the same finalization certificate twice
+	if _, ok := r.receivedFinalizationCertificates[data.FCert.Finalization.Seq]; ok {
+		return nil
+	}
+
+	r.receivedFinalizationCertificates[data.FCert.Finalization.Seq] = data
+	return nil
+}