add connect/disconnect to test network

Author: samliok

epoch_multinode_test.go

@@ -89,11 +89,8 @@ func defaultTestNodeEpochConfig(t *testing.T, nodeID NodeID, net *inMemNetwork,
 	l := testutil.MakeLogger(t, int(nodeID[0]))
 	storage := newInMemStorage()
 	conf := EpochConfig{
-		MaxProposalWait: DefaultMaxProposalWaitTime,
-		Comm: &testComm{
-			from: nodeID,
-			net:  net,
-		},
+		MaxProposalWait:     DefaultMaxProposalWaitTime,
+		Comm:                newTestComm(nodeID, net),
 		Logger:              l,
 		ID:                  nodeID,
 		Signer:              &testSigner{},
@@ -213,11 +210,23 @@ type testComm struct {
 	net  *inMemNetwork
 }
 
+func newTestComm(from NodeID, net *inMemNetwork) *testComm {
+	return &testComm{
+		from: from,
+		net:  net,
+	}
+}
+
 func (c *testComm) ListNodes() []NodeID {
 	return c.net.nodes
 }
 
 func (c *testComm) SendMessage(msg *Message, destination NodeID) {
+	// cannot send if either [from] or [destination] is not connected
+	if !c.net.IsConnected(destination) || !c.net.IsConnected(c.from) {
+		return
+	}
+
 	for _, instance := range c.net.instances {
 		if bytes.Equal(instance.e.ID, destination) {
 			instance.ingress <- struct {
@@ -230,11 +239,19 @@ func (c *testComm) SendMessage(msg *Message, destination NodeID) {
 }
 
 func (c *testComm) Broadcast(msg *Message) {
+	if !c.net.IsConnected(c.from) {
+		return
+	}
+
 	for _, instance := range c.net.instances {
 		// Skip sending the message to yourself
 		if bytes.Equal(c.from, instance.e.ID) {
 			continue
 		}
+		if !c.net.IsConnected(instance.e.ID) {
+			continue
+		}
+
 		instance.ingress <- struct {
 			msg  *Message
 			from NodeID
@@ -246,15 +263,22 @@ type inMemNetwork struct {
 	t         *testing.T
 	nodes     []NodeID
 	instances []*testNode
+	connected map[string]bool
 }
 
 // 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 {
+	connected := make(map[string]bool)
+	for _, node := range nodes {
+		connected[string(node)] = true
+	}
+
 	net := &inMemNetwork{
 		t:         t,
 		nodes:     nodes,
 		instances: make([]*testNode, 0),
+		connected: connected,
 	}
 	return net
 }
@@ -271,6 +295,23 @@ func (n *inMemNetwork) addNode(node *testNode) {
 	n.instances = append(n.instances, node)
 }
 
+func (n *inMemNetwork) IsConnected(node NodeID) bool {
+	for _, instance := range n.instances {
+		if bytes.Equal(instance.e.ID, node) {
+			return n.connected[string(node)]
+		}
+	}
+	return false
+}
+
+func (n *inMemNetwork) Connect(node NodeID) {
+	n.connected[string(node)] = true
+}
+
+func (n *inMemNetwork) Disconnect(node NodeID) {
+	n.connected[string(node)] = false
+}
+
 // 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() {

replication_test.go

@@ -4,6 +4,7 @@
 package simplex_test
 
 import (
+	"bytes"
 	"context"
 	"simplex"
 	"simplex/testutil"
@@ -172,6 +173,72 @@ func TestReplicationStartsBeforeCurrentRound(t *testing.T) {
 	}
 }
 
+// TestReplicationAfterNodeDisconnects tests the replication process of a node that
+// disconnects from the network and reconnects after the rest of the network has made progress.
+func TestReplicationAfterNodeDisconnects(t *testing.T) {
+	bb := newTestControlledBlockBuilder(t)
+	nodes := []simplex.NodeID{{1}, {2}, {3}, []byte("lagging")}
+	net := newInMemNetwork(t, nodes)
+	startDisconnect := uint64(5)
+	endDisconnect := uint64(18)
+
+	normalNode1 := newSimplexNode(t, nodes[0], net, bb, true)
+	normalNode2 := newSimplexNode(t, nodes[1], net, bb, true)
+	normalNode3 := newSimplexNode(t, nodes[2], net, bb, true)
+	laggingNode := newSimplexNode(t, nodes[3], net, bb, true)
+
+	require.Equal(t, uint64(0), normalNode1.storage.Height())
+	require.Equal(t, uint64(0), normalNode2.storage.Height())
+	require.Equal(t, uint64(0), normalNode3.storage.Height())
+	require.Equal(t, uint64(0), laggingNode.storage.Height())
+
+	bb.triggerNewBlock()
+
+	net.startInstances()
+
+	for i := uint64(0); i < startDisconnect; i++ {
+		for _, n := range net.instances {
+			n.storage.waitForBlockCommit(i)
+		}
+
+		bb.triggerNewBlock()
+	}
+
+	require.Equal(t, startDisconnect, normalNode1.storage.Height())
+	require.Equal(t, startDisconnect, normalNode2.storage.Height())
+	require.Equal(t, startDisconnect, normalNode3.storage.Height())
+	require.Equal(t, startDisconnect, laggingNode.storage.Height())
+
+	// lagging node disconnects
+	net.Disconnect(nodes[3])
+
+	// normal nodes continue to make progress
+	for i := startDisconnect; i < endDisconnect; i++ {
+		if bytes.Equal(simplex.LeaderForRound(nodes, i), nodes[3]) {
+			// TODO: build empty block
+		}
+		for _, n := range net.instances[:3] {
+			n.storage.waitForBlockCommit(i)
+		}
+		bb.triggerNewBlock()
+	}
+
+	require.Equal(t, endDisconnect, normalNode1.storage.Height())
+	require.Equal(t, endDisconnect, normalNode2.storage.Height())
+	require.Equal(t, endDisconnect, normalNode3.storage.Height())
+	require.Equal(t, startDisconnect, laggingNode.storage.Height())
+
+	// lagging node reconnects
+	net.Connect(nodes[3])
+
+	// lagging node catches up
+	bb.triggerNewBlock()
+	require.Equal(t, endDisconnect+1, laggingNode.storage.Height())
+	require.Equal(t, endDisconnect+1, normalNode1.storage.Height())
+	require.Equal(t, endDisconnect+1, normalNode2.storage.Height())
+	require.Equal(t, endDisconnect+1, normalNode3.storage.Height())
+}
+
 func createBlocks(t *testing.T, nodes []simplex.NodeID, bb simplex.BlockBuilder, seqCount uint64) []simplex.FinalizedBlock {
 	logger := testutil.MakeLogger(t, int(0))
 	ctx := context.Background()