Ensure Non-Validators and Validators Bootstrap Before Indexing and Verifying Blocks

common/api.go

@@ -203,6 +203,31 @@ func (nws Nodes) Contains(nodeID NodeID) bool {
 	return false
 }
 
+// Equal returns whether both hold the same nodes, ignoring order.
+func (nws Nodes) Equal(other Nodes) bool {
+	if len(nws) != len(other) {
+		return false
+	}
+
+	nwsClone := slices.Clone(nws)
+	otherClone := slices.Clone(other)
+	SortNodes(nwsClone)
+	SortNodes(otherClone)
+
+	for i := range nwsClone {
+		if !bytes.Equal(nwsClone[i].Id, otherClone[i].Id) {
+			return false
+		}
+		if !bytes.Equal(nwsClone[i].PK, otherClone[i].PK) {
+			return false
+		}
+		if nwsClone[i].Weight != otherClone[i].Weight {
+			return false
+		}
+	}
+	return true
+}
+
 // Node is a struct that pairs a node ID with its weight and public key.
 type Node struct {
 	Id     NodeID

common/global_test.go

@@ -24,3 +24,49 @@ func TestNodeIDs(t *testing.T) {
 		}
 	}
 }
+
+// TestNodesEqual checks Equal ignores order but compares
+// length, Id, PK and Weight of every node.
+func TestNodesEqual(t *testing.T) {
+	a := Node{Id: NodeID{1}, Weight: 10, PK: PublicKeyBytes{0xa}}
+	b := Node{Id: NodeID{2}, Weight: 20, PK: PublicKeyBytes{0xb}}
+	c := Node{Id: NodeID{3}, Weight: 30, PK: PublicKeyBytes{0xc}}
+
+	testCases := []struct {
+		name  string
+		nws   Nodes
+		other Nodes
+		equal bool
+	}{
+		{name: "both nil", equal: true},
+		{name: "nil and empty", other: Nodes{}, equal: true},
+		{name: "same order", nws: Nodes{a, b, c}, other: Nodes{a, b, c}, equal: true},
+		{name: "different order", nws: Nodes{a, b, c}, other: Nodes{c, a, b}, equal: true},
+		{name: "different length", nws: Nodes{a, b}, other: Nodes{a, b, c}, equal: false},
+		{name: "empty and non empty", nws: Nodes{}, other: Nodes{a}, equal: false},
+		{name: "different id", nws: Nodes{a, b}, other: Nodes{a, {Id: NodeID{9}, Weight: b.Weight, PK: b.PK}}, equal: false},
+		{name: "different weight", nws: Nodes{a, b}, other: Nodes{a, {Id: b.Id, Weight: 99, PK: b.PK}}, equal: false},
+		{name: "different pk", nws: Nodes{a, b}, other: Nodes{a, {Id: b.Id, Weight: b.Weight, PK: PublicKeyBytes{0xff}}}, equal: false},
+		{name: "duplicate vs distinct", nws: Nodes{a, a}, other: Nodes{a, b}, equal: false},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			require.Equal(t, testCase.equal, testCase.nws.Equal(testCase.other))
+			require.Equal(t, testCase.equal, testCase.other.Equal(testCase.nws))
+		})
+	}
+}
+
+// TestNodesEqualDoesNotMutate checks Equal sorts clones and
+// leaves the receiver and argument in their original order.
+func TestNodesEqualDoesNotMutate(t *testing.T) {
+	a := Node{Id: NodeID{1}, Weight: 10, PK: PublicKeyBytes{0xa}}
+	b := Node{Id: NodeID{2}, Weight: 20, PK: PublicKeyBytes{0xb}}
+
+	nws := Nodes{b, a}
+	other := Nodes{a, b}
+	require.True(t, nws.Equal(other))
+	require.Equal(t, Nodes{b, a}, nws)
+	require.Equal(t, Nodes{a, b}, other)
+}

common/timeout_handler.go

@@ -133,6 +133,13 @@ func (t *TimeoutHandler[T]) RemoveTask(ID T) {
 	delete(t.tasks, ID)
 }
 
+func (t *TimeoutHandler[T]) Empty() bool {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	return len(t.tasks) == 0
+}
+
 func (t *TimeoutHandler[T]) RemoveOldTasks(shouldRemove func(id T, _ struct{}) bool) {
 	t.lock.Lock()
 	defer t.lock.Unlock()

instance.go

@@ -113,13 +113,8 @@ func (i *Instance) Start(ctx context.Context) error {
 
 	context.AfterFunc(ctx, i.Stop)
 
-	nodes, epochNum, err := getLastAcceptedEpochAndValidatorSet(&i.Config)
-	if err != nil {
-		return fmt.Errorf("error determining latest epoch and validator set: %w", err)
-	}
-
-	if err := i.startAtEpoch(nodes); err != nil {
-		return fmt.Errorf("error starting instance at epoch %d: %w", epochNum, err)
+	if err := i.maybeReplicateEpochs(); err != nil {
+		return err
 	}
 
 	go i.tick()
@@ -128,6 +123,32 @@ func (i *Instance) Start(ctx context.Context) error {
 	return nil
 }
 
+func (i *Instance) maybeReplicateEpochs() error {
+	i.Config.Logger.Debug("Checking if epoch replication is required")
+	latestValidatorSet, err := getLatestPlatformChainValidatorSet(i.Config.PlatformChain)
+	if err != nil {
+		return err
+	}
+
+	latestIndexedEpochValidators, err := getLastAcceptedValidatorSet(&i.Config)
+	if err != nil {
+		return err
+	}
+
+	// We have indexed the latest validator set, therefore we can skip epoch replication and start as a validator.
+	// Note: this may not be the latest epoch, but a future PR will eventually notice we are behind and transition properly.
+	if latestIndexedEpochValidators.Equal(latestValidatorSet.Nodes()) && latestValidatorSet.Nodes().Contains(i.Config.ID) {
+		i.Config.Logger.Debug("Node skipping epoch replication because its latest epoch is up to date with the Platform Chain")
+		return i.startValidator(latestIndexedEpochValidators)
+	}
+
+	// Start as non-validator if our last indexed validator set does not equal, the latest p-chain validator set
+	// Note: the epoch may be transitioning, so the latest p-chain validator set actually points to a future epoch.
+	// The non-validator should finish replicating epochs and convert our non-validator to a validator in this case.
+	i.Config.Logger.Debug("Node starting epoch replication as a non-validator")
+	return i.startNonValidator(false)
+}
+
 func (i *Instance) startValidator(validators common.Nodes) error {
 	epochConfig, err := i.createEpochConfig(validators)
 	if err != nil {
@@ -146,8 +167,10 @@ func (i *Instance) startValidator(validators common.Nodes) error {
 	return epoch.Start()
 }
 
-func (i *Instance) startNonValidator() error {
-	config, err := i.createNonValidatorConfig()
+// startNonValidator runs a non-validator. epochsReplicated is true when we already hold the
+// newest sealing block, such as when a validator leaves the validator set.
+func (i *Instance) startNonValidator(epochsReplicated bool) error {
+	config, err := i.createNonValidatorConfig(epochsReplicated)
 	if err != nil {
 		return err
 	}
@@ -162,19 +185,18 @@ func (i *Instance) startNonValidator() error {
 	return nil
 }
 
-func (i *Instance) createNonValidatorConfig() (nonvalidator.Config, error) {
+func (i *Instance) createNonValidatorConfig(epochsReplicated bool) (nonvalidator.Config, error) {
 	source, err := simplex.NewRandomSource()
 	if err != nil {
 		return nonvalidator.Config{}, err
 	}
 
-	height := i.Config.PlatformChain.GetCurrentHeight()
-	mappings, err := i.Config.PlatformChain.GetValidatorSet(height)
+	latestValidatorSet, err := getLatestPlatformChainValidatorSet(i.Config.PlatformChain)
 	if err != nil {
 		return nonvalidator.Config{}, err
 	}
 
-	comm := newCommunication(i.Config.Sender, i.Config.Broadcaster, mappings.Nodes())
+	comm := newCommunication(i.Config.Sender, i.Config.Broadcaster, latestValidatorSet.Nodes())
 
 	// Plant an artificial MSM. A non-validator never verifies the state machine transition,
 	// it only verifies the inner block (see common.OnlyVMVerifyOpt), so this MSM is only
@@ -203,6 +225,7 @@ func (i *Instance) createNonValidatorConfig() (nonvalidator.Config, error) {
 		SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator,
 		MaxSequenceWindow:          simplex.DefaultMaxRoundWindow,
 		TransitionToValidator:      i.notifyEpochChange,
+		EpochsReplicated:           epochsReplicated,
 	}
 	return config, nil
 }
@@ -353,13 +376,15 @@ func (i *Instance) HandleMessage(msg *common.Message, from common.NodeID) error
 			i.msm.HandleApproval(msg.EpochTransitionApproval, uint64(time.Now().UnixMilli()))
 			return nil
 		}
+
 		return i.e.HandleMessage(msg, from)
 	}
 
 	if i.nv != nil {
 		return i.nv.HandleMessage(msg, from)
 	}
-	return nil
+
+	return errors.New("we are not running as a validator or not validator")
 }
 
 func (i *Instance) wireReplicationResponse(msg *common.Message) error {
@@ -594,7 +619,7 @@ func (i *Instance) startAtEpoch(validators common.Nodes) error {
 		return i.startValidator(validators)
 	}
 
-	return i.startNonValidator()
+	return i.startNonValidator(true)
 }
 
 type epochConfig struct {

instance_helpers_test.go

@@ -384,6 +384,10 @@ func (i *instanceComm) enqueue(m inflightMessage) {
 }
 
 func (c *instanceComm) Send(msg *common.Message, destination common.NodeID) {
+	if c.n.isOffline(c.id) || c.n.isOffline(destination) {
+		return
+	}
+
 	for _, n := range c.n.nodesSnapshot() {
 		if !bytes.Equal(n.id, destination) {
 			continue
@@ -400,9 +404,13 @@ func (c *instanceComm) Send(msg *common.Message, destination common.NodeID) {
 }
 
 func (c *instanceComm) Broadcast(msg *common.Message) {
+	if c.n.isOffline(c.id) {
+		return
+	}
+
 	// every node in the network but ourselves, each with its own re-parsed copy
 	for _, n := range c.n.nodesSnapshot() {
-		if bytes.Equal(n.id, c.id) {
+		if bytes.Equal(n.id, c.id) || c.n.isOffline(n.id) {
 			continue
 		}
 
@@ -601,20 +609,24 @@ func (n *node) restart() *node {
 	return newNode
 }
 
+// role reports whether the instance currently runs a validator epoch rather than a
+// non-validator, and whether it has finished replicating epochs.
+func (n *node) role() (isValidator bool, epochsReplicated bool) {
+	n.inst.lock.Lock()
+	defer n.inst.lock.Unlock()
+
+	if n.inst.e != nil {
+		return true, true
+	}
+	return false, n.inst.nv != nil && n.inst.nv.HasReplicatedEpochs()
+}
+
 // sync syncs a node by waiting for the commit of the latest sequence.
 func (n *node) sync() *node {
 	n.storage.WaitForBlockCommit(n.net.seq - 1)
 	return n
 }
 
-// role reports whether the node is running a validator rather than a non-validator.
-func (n *node) role() (isValidator bool) {
-	n.inst.lock.Lock()
-	defer n.inst.lock.Unlock()
-
-	return n.inst.e != nil
-}
-
 const firstEverEpoch uint64 = 1
 
 type network struct {
@@ -628,9 +640,11 @@ type network struct {
 	// pending holds the block the network has been asked to build, claimable by any leader.
 	pending *pendingBlockSignal
 
-	// lock guards nodes, which comm goroutines read while addNode appends.
+	// lock guards nodes and offline, which comm goroutines read while tests mutate them.
 	lock  sync.Mutex
 	nodes []node
+	// offline nodes stay in the network but neither send nor receive messages.
+	offline map[string]struct{}
 }
 
 func (n *network) nodesSnapshot() []node {
@@ -639,6 +653,25 @@ func (n *network) nodesSnapshot() []node {
 	return append([]node(nil), n.nodes...)
 }
 
+func (n *network) setOffline(id common.NodeID) {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+	n.offline[string(id)] = struct{}{}
+}
+
+func (n *network) setOnline(id common.NodeID) {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+	delete(n.offline, string(id))
+}
+
+func (n *network) isOffline(id common.NodeID) bool {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+	_, offline := n.offline[string(id)]
+	return offline
+}
+
 func newNetwork(t *testing.T, pChain *testPlatformChain) *network {
 	genesisNodes := pChain.GenesisValidatorSet().Nodes()
 	common.SortNodes(genesisNodes)
@@ -647,6 +680,7 @@ func newNetwork(t *testing.T, pChain *testPlatformChain) *network {
 		t:                 t,
 		pChain:            pChain,
 		pending:           newPendingBlockSignal(),
+		offline:           make(map[string]struct{}),
 		epochValidatorSet: genesisNodes,
 
 		// Genesis at seq 0. Then first simplex block is built automatically
@@ -754,8 +788,10 @@ func (n *network) waitUntilValidatorsReady() {
 			continue
 		}
 
-		require.Eventually(n.t, node.role, time.Minute, time.Millisecond,
-			"node %x never started running a validator", node.id)
+		require.Eventually(n.t, func() bool {
+			isValidator, _ := node.role()
+			return isValidator
+		}, time.Minute, time.Millisecond, "node %x never started running a validator", node.id)
 	}
 }
 
@@ -828,6 +864,9 @@ func (n *network) waitUntilSealingBlock(expectedValidatorSet common.Nodes) commo
 	for {
 		var block common.VerifiedBlock
 		for _, node := range n.nodes {
+			if n.isOffline(node.id) {
+				continue
+			}
 			committedBlock := node.storage.WaitForBlockCommit(n.seq)
 			if block == nil {
 				block = committedBlock