add non-validator storage

adapters.go

@@ -13,32 +13,60 @@ import (
 	"github.com/ava-labs/simplex/common"
 	metadata "github.com/ava-labs/simplex/msm"
 	"github.com/ava-labs/simplex/simplex"
+	"go.uber.org/zap"
 )
 
-type Communication struct {
-	nodes atomic.Value // common.Nodes
+type communication struct {
+	nodes common.Nodes
 	Sender
 	Broadcaster
 }
 
-func newCommunication(sender Sender, broadcaster Broadcaster, validators common.Nodes) *Communication {
-	c := &Communication{
+func newCommunication(sender Sender, broadcaster Broadcaster, validators common.Nodes) *communication {
+	return &communication{
+		nodes:       validators,
 		Sender:      sender,
 		Broadcaster: broadcaster,
 	}
-	c.SetValidators(validators)
-	return c
 }
 
-func (c *Communication) SetValidators(nodes common.Nodes) {
-	c.nodes.Store(nodes)
+func (c *communication) Validators() common.Nodes {
+	return c.nodes
 }
 
-func (c *Communication) Validators() common.Nodes {
-	nodes, ok := c.nodes.Load().(common.Nodes)
-	if !ok {
-		return nil
+// nonValidatorCommunication has no fixed validator set, so it fetches the latest one on every call.
+type nonValidatorCommunication struct {
+	platformChain  PlatformChain
+	logger         common.Logger
+	lastValidators atomic.Pointer[common.Nodes]
+	Sender
+	Broadcaster
+}
+
+func newNonValidatorCommunication(sender Sender, broadcaster Broadcaster, platformChain PlatformChain, logger common.Logger) (*nonValidatorCommunication, error) {
+	validatorSet, err := getLatestPlatformChainValidatorSet(platformChain)
+	if err != nil {
+		return nil, err
+	}
+	c := &nonValidatorCommunication{
+		platformChain: platformChain,
+		logger:        logger,
+		Sender:        sender,
+		Broadcaster:   broadcaster,
+	}
+	nodes := validatorSet.Nodes()
+	c.lastValidators.Store(&nodes)
+	return c, nil
+}
+
+func (c *nonValidatorCommunication) Validators() common.Nodes {
+	validatorSet, err := getLatestPlatformChainValidatorSet(c.platformChain)
+	if err != nil {
+		c.logger.Warn("Failed fetching latest validator set, using last known set", zap.Error(err))
+		return *c.lastValidators.Load()
 	}
+	nodes := validatorSet.Nodes()
+	c.lastValidators.Store(&nodes)
 	return nodes
 }
 

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.maybeBootstrap(); err != nil {
+		return err
 	}
 
 	go i.tick()
@@ -128,6 +123,32 @@ func (i *Instance) Start(ctx context.Context) error {
 	return nil
 }
 
+func (i *Instance) maybeBootstrap() error {
+	i.Config.Logger.Debug("Checking if bootstrapping 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 bootstrapping 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 bootstrapping 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 bootstrapping and convert our non-validator to a validator in this case.
+	i.Config.Logger.Debug("Node starting bootstrapping 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. bootstrapped is true when we already hold the
+// newest sealing block, such as when a validator leaves the validator set.
+func (i *Instance) startNonValidator(bootstrapped bool) error {
+	config, err := i.createNonValidatorConfig(bootstrapped)
 	if err != nil {
 		return err
 	}
@@ -162,20 +185,17 @@ func (i *Instance) startNonValidator() error {
 	return nil
 }
 
-func (i *Instance) createNonValidatorConfig() (nonvalidator.Config, error) {
+func (i *Instance) createNonValidatorConfig(bootstrapped 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)
+	comm, err := newNonValidatorCommunication(i.Config.Sender, i.Config.Broadcaster, i.Config.PlatformChain, i.Config.Logger)
 	if err != nil {
 		return nonvalidator.Config{}, err
 	}
 
-	comm := newCommunication(i.Config.Sender, i.Config.Broadcaster, mappings.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
 	// used to wire blocks and is never asked to verify them.
@@ -203,6 +223,7 @@ func (i *Instance) createNonValidatorConfig() (nonvalidator.Config, error) {
 		SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator,
 		MaxSequenceWindow:          simplex.DefaultMaxRoundWindow,
 		TransitionToValidator:      i.notifyEpochChange,
+		Bootstrapped:               bootstrapped,
 	}
 	return config, nil
 }
@@ -353,13 +374,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 +617,7 @@ func (i *Instance) startAtEpoch(validators common.Nodes) error {
 		return i.startValidator(validators)
 	}
 
-	return i.startNonValidator()
+	return i.startNonValidator(true)
 }
 
 type epochConfig struct {