diff --git a/adapters.go b/adapters.go index 4d6b2153..f6b6d35b 100644 --- a/adapters.go +++ b/adapters.go @@ -19,6 +19,15 @@ type Communication struct { Broadcaster } +func newCommunication(sender Sender, broadcaster Broadcaster, validators common.Nodes) *Communication { + c := &Communication{ + Sender: sender, + Broadcaster: broadcaster, + } + c.SetValidators(validators) + return c +} + func (c *Communication) SetValidators(nodes common.Nodes) { c.nodes.Store(nodes) } @@ -31,45 +40,52 @@ func (c *Communication) Validators() common.Nodes { return nodes } -// EpochAwareStorage is a wrapper around Storage that is aware of epoch changes. -// Upon an epoch change, it will ignore blocks from previous epochs -// and will call the onEpochChange callback when a new epoch is detected. -type EpochAwareStorage struct { - msm *metadata.StateMachine - onEpochChange func(seq uint64, validators common.Nodes) error +// InstanceStorage is a wrapper around Storage that skips indexing Telocks +// and delegates post-index handling to a caller-provided onIndex hook. +type InstanceStorage struct { Storage - epoch uint64 + + msm *metadata.StateMachine + + onIndex func(block *ParsedBlock) error +} + +func NewInstanceStorage(storage Storage, msm *metadata.StateMachine, onIndex func(block *ParsedBlock) error) *InstanceStorage { + return &InstanceStorage{ + Storage: storage, + msm: msm, + onIndex: onIndex, + } } -func (e *EpochAwareStorage) Retrieve(seq uint64) (common.VerifiedBlock, common.Finalization, error) { - block, finalization, err := e.GetBlock(seq) +func (s *InstanceStorage) Retrieve(seq uint64) (common.VerifiedBlock, common.Finalization, error) { + block, finalization, err := s.GetBlock(seq) if err != nil { return nil, common.Finalization{}, err } parsedBlock := &ParsedBlock{ - msm: e.msm, + msm: s.msm, StateMachineBlock: block, } return parsedBlock, *finalization, nil } -func (e *EpochAwareStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error { - if block.BlockHeader().Epoch < e.epoch { - // This is a Telock from a previous epoch, so we ignore it and do not index it. +func (s *InstanceStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error { + pb, ok := block.(*ParsedBlock) + if !ok { + return fmt.Errorf("expected ParsedBlock, got %T", block) + } + + // A Telock only extends time until the epoch transition finalizes, so we never index it. + if pb.Type() == metadata.BlockTypeTelock { return nil } - if err := e.Storage.Index(ctx, block, certificate); err != nil { + + if err := s.Storage.Index(ctx, block, certificate); err != nil { return err } - // This is a sealing block, and it is not the zero block - if block.SealingBlockInfo() != nil && block.SealingBlockInfo().PrevSealingBlockHash != [32]byte{} { - if err := e.onEpochChange(block.BlockHeader().Seq, block.SealingBlockInfo().ValidatorSet); err != nil { - return err - } - // We are now in a new epoch, so we update the epoch number to prevent indexing Telocks from the previous epoch. - e.epoch = block.BlockHeader().Seq - } - return nil + + return s.onIndex(pb) } // cachedBlock is a wrapper around ParsedBlock that caches the block in the CachedStorage upon verification. diff --git a/instance.go b/instance.go index 848e14b2..c0ebd0ac 100644 --- a/instance.go +++ b/instance.go @@ -5,6 +5,7 @@ package simplex import ( "context" + "errors" "fmt" "math" "sync" @@ -19,6 +20,8 @@ import ( "go.uber.org/zap" ) +var errAlreadyStarted = errors.New("instance already started") + const ( // tickInterval is the interval at which the instance will call AdvanceTime on the current epoch or non-validator. tickInterval = time.Millisecond * 100 @@ -26,7 +29,7 @@ const ( type Config struct { // LastNonSimplexInnerBlock is the last non-simplex inner block that was persisted to storage. - // This is used to determine the current epoch and validator set. + // The genesis validator state from pchain is used to determine the current epoch and validator set. Can be the genesis block LastNonSimplexInnerBlock avalanchego.VMBlock // ParameterConfig is the configuration for the simplex instance. ParameterConfig ParameterConfig @@ -34,6 +37,8 @@ type Config struct { PlatformChain PlatformChain // Broadcaster is the interface to broadcast messages to other nodes in the network. Broadcaster Broadcaster + // Sender is an interface to send messages to a specific node in the network + Sender Sender // CryptoOps is the interface to the cryptographic operations needed by the simplex instance. CryptoOps CryptoOps // WalCreator is the interface to create new write-ahead logs for the simplex instance. @@ -41,7 +46,6 @@ type Config struct { // Storage is the interface to the block storage layer for the simplex instance. Storage Storage Logger common.Logger - Sender Sender WALs []wal.DeletableWAL VM VM ICMETransition metadata.ICMEpochTransition @@ -49,25 +53,19 @@ type Config struct { ID common.NodeID } -type nodeRole byte - -const ( - nonValidator nodeRole = iota - validator -) - type epochChange struct { - epochNum uint64 + epoch uint64 validators common.Nodes - nodeRole nodeRole } +func noopOnIndex(*ParsedBlock) error { return nil } type timeAdvancer interface { AdvanceTime(t time.Time) } type Instance struct { - Config Config + Config Config + lock sync.Mutex started bool cs *CachedStorage @@ -84,8 +82,8 @@ func NewInstance(config Config) *Instance { return &Instance{ Config: config, stopCh: make(chan struct{}), - cs: NewCachedStorage(config.Storage), epochChanges: make(chan epochChange, 1), + cs: NewCachedStorage(config.Storage), } } @@ -95,21 +93,14 @@ func (i *Instance) Start(ctx context.Context) error { defer i.lock.Unlock() if i.started { - return fmt.Errorf("instance already started") + return errAlreadyStarted } i.started = true context.AfterFunc(ctx, i.Stop) - lastBlock, numBlocks, err := i.lastBlock() - if err != nil { - return fmt.Errorf("error retrieving last block: %w", err) - } - - lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height() - genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet() - nodes, epochNum, err := constructEpochAndValidatorSet(i.Config.Logger, lastNonSimplexHeight, genesisValidatorSet, numBlocks, &ParsedBlock{StateMachineBlock: lastBlock}, i.Config.Storage) + nodes, epochNum, err := getLastAcceptedEpochAndValidatorSet(&i.Config) if err != nil { return fmt.Errorf("error determining latest epoch and validator set: %w", err) } @@ -124,16 +115,26 @@ func (i *Instance) Start(ctx context.Context) error { return nil } -func (i *Instance) startValidator() error { - epochConfig, err := i.createEpochConfig() +func (i *Instance) startValidator(epochNum uint64, validators common.Nodes) error { + epochConfig, err := i.createEpochConfig(epochNum, validators) if err != nil { return err } - return i.startEpoch(epochConfig) + + epoch, err := simplex.NewEpoch(epochConfig) + if err != nil { + return fmt.Errorf("error creating simplex epoch: %w", err) + } + + epoch.Epoch = epochConfig.Epoch + i.e = epoch + i.epochOrNV = epoch + + return epoch.Start() } -func (i *Instance) startNonValidator(epochNum uint64, validators common.Nodes) error { - config, err := i.createNonValidatorConfig(epochNum, validators) +func (i *Instance) startNonValidator() error { + config, err := i.createNonValidatorConfig() if err != nil { return err } @@ -148,35 +149,17 @@ func (i *Instance) startNonValidator(epochNum uint64, validators common.Nodes) e return nil } -func (i *Instance) createNonValidatorConfig(epochNum uint64, validators common.Nodes) (nonvalidator.Config, error) { +func (i *Instance) createNonValidatorConfig() (nonvalidator.Config, error) { source, err := simplex.NewRandomSource() if err != nil { return nonvalidator.Config{}, err } - comm := &Communication{Sender: i.Config.Sender, Broadcaster: i.Config.Broadcaster} - comm.SetValidators(validators) - - epochAwareStorage := &EpochAwareStorage{ - epoch: epochNum, - Storage: i.Config.Storage, - onEpochChange: func(epoch uint64, validators common.Nodes) error { - height := i.Config.PlatformChain.GetCurrentHeight() - vdrs, err := i.Config.PlatformChain.GetValidatorSet(height) - if err != nil { - i.Config.Logger.Error("error getting validator set", zap.Error(err)) - return fmt.Errorf("error getting validator set from platform chain: %w", err) - } - comm.SetValidators(validators) - if i.iAmValidator(vdrs.Nodes()) { - i.notifyEpochChange(epoch, validators, nonValidator) - } else { - i.Config.Logger.Debug("I am still a non-validator at the tip of the P-chain, skipping role change", - zap.Uint64("height", height)) - } - return nil - }, + nodes, err := GetHighestValidatorSet(i.Config.PlatformChain) + if err != nil { + return nonvalidator.Config{}, err } + comm := newCommunication(i.Config.Sender, i.Config.Broadcaster, 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 @@ -185,30 +168,41 @@ func (i *Instance) createNonValidatorConfig(epochNum uint64, validators common.N Config: &metadata.Config{}, } i.cs.msm = i.msm + instanceStorage := NewInstanceStorage(i.cs, i.msm, noopOnIndex) config := nonvalidator.Config{ ID: i.Config.ID, RandomSource: source, - Storage: epochAwareStorage, + Storage: instanceStorage, Comm: comm, Logger: i.Config.Logger, StartTime: time.Now(), SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator, MaxSequenceWindow: simplex.DefaultMaxRoundWindow, + TransitionToValidator: i.notifyEpochChange, } return config, nil } -func (i *Instance) notifyEpochChange(epoch uint64, validators common.Nodes, role nodeRole) { - select { - case i.epochChanges <- epochChange{ - epochNum: epoch, +func (i *Instance) notifyEpochChange(epoch uint64, validators common.Nodes) { + ec := epochChange{ + epoch: epoch, validators: validators, - nodeRole: role, - }: - case <-i.stopCh: - // If the instance is stopped, we don't need to notify about epoch changes. - return + } + + for { + select { + case i.epochChanges <- ec: + return + // The slot holds a stale epoch change: take it, keep the newer of the two and retry. + case pending := <-i.epochChanges: + if pending.epoch > ec.epoch { + ec = pending + } + case <-i.stopCh: + // If the instance is stopped, we don't need to notify about epoch changes. + return + } } } @@ -250,7 +244,7 @@ func (i *Instance) Stop() { close(i.stopCh) } - i.stopValidator() + i.stopValidator(false) i.stopNonValidator() } @@ -262,9 +256,18 @@ func (i *Instance) stopNonValidator() { } } -func (i *Instance) stopValidator() { +func (i *Instance) stopValidator(garbageCollectWAL bool) { if i.e != nil { i.e.Stop() + // Wipe out the WALs from the config so we won't try to load them again + if garbageCollectWAL { + i.Config.WALs = nil + // On epoch change, garbage collect the WAL to remove all entries from previous epochs. + if err := i.wal.GarbageCollect(math.MaxUint64); err != nil { + i.Config.Logger.Error("Error garbage collecting epoch config on epoch change", zap.Error(err)) + } + } + i.e = nil i.epochOrNV = nil } @@ -364,73 +367,40 @@ func (i *Instance) listenForEpochChanges() { } func (i *Instance) processEpochChange(epochChange epochChange) { - var err error - switch epochChange.nodeRole { - case nonValidator: - err = i.transitionEpochNonValidator(epochChange) - case validator: - err = i.transitionEpochValidator(epochChange) - default: // This should never happen, but we log it just in case. - i.Config.Logger.Fatal("Unknown node role on epoch change", - zap.String("role", fmt.Sprintf("%v", epochChange.nodeRole))) - return - } - if err != nil { - i.Config.Logger.Error("Error transitioning epoch", zap.Uint8("role", uint8(epochChange.nodeRole)), zap.Error(err)) - i.Stop() - } -} + // Hold the lock so the transition cannot interleave with Stop or HandleMessage. + i.lock.Lock() -// startEpoch starts a new epoch with the given configuration. -// Must be called under the lock, and assumes that the previous epoch has been stopped (if any). -func (i *Instance) startEpoch(epochConfig simplex.EpochConfig) error { - epoch, err := simplex.NewEpoch(epochConfig) - if err != nil { - return fmt.Errorf("error creating simplex epoch: %w", err) + if i.isStopped() { + i.lock.Unlock() + i.Config.Logger.Info("instance is already stopped, skipping epoch change") + return } - epoch.Epoch = epochConfig.Epoch - i.e = epoch - i.epochOrNV = epoch - return epoch.Start() -} + var err error -func (i *Instance) lastBlock() (metadata.StateMachineBlock, uint64, error) { - numBlocks := i.Config.Storage.NumBlocks() - if numBlocks == 0 { - return metadata.StateMachineBlock{}, 0, fmt.Errorf("no genesis block found in storage") + switch { + case i.nv != nil: + // Stop the non-validator before doing anything else, so that we don't process any more messages while we are changing epochs. + i.stopNonValidator() + err = i.startAtEpoch(epochChange.validators, epochChange.epoch) + case i.e != nil: + i.stopValidator(true) + + err = i.startAtEpoch(epochChange.validators, epochChange.epoch) + default: // This should never happen, but we log it just in case. + i.lock.Unlock() + i.Config.Logger.Fatal("We are not running either a validator or non-validator") + return } + i.lock.Unlock() - lastBlock, _, err := i.Config.Storage.GetBlock(numBlocks - 1) if err != nil { - return metadata.StateMachineBlock{}, 0, fmt.Errorf("error retrieving last block from storage: %w", err) - } - - return lastBlock, numBlocks, nil -} - -func (i *Instance) iAmValidator(nodes common.Nodes) bool { - for _, node := range nodes { - if i.Config.ID.Equals(node.Id) { - return true - } + i.Config.Logger.Error("Error transitioning epoch", zap.Error(err)) + i.Stop() } - return false } -func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { - lastBlock, numBlocks, err := i.lastBlock() - if err != nil { - return simplex.EpochConfig{}, err - } - - lastNonSimplexHeight := i.Config.LastNonSimplexInnerBlock.Height() - genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet() - nodes, epochNum, err := constructEpochAndValidatorSet(i.Config.Logger, lastNonSimplexHeight, genesisValidatorSet, numBlocks, &ParsedBlock{StateMachineBlock: lastBlock}, i.Config.Storage) - if err != nil { - return simplex.EpochConfig{}, err - } - +func (i *Instance) createEpochConfig(epoch uint64, validators common.Nodes) (simplex.EpochConfig, error) { wal, err := wal.NewGarbageCollectedWAL(i.Config.WALs, i.Config.WalCreator, &common.WALRetentionReader{}, i.Config.ParameterConfig.WALMaxEntryCount) if err != nil { return simplex.EpochConfig{}, fmt.Errorf("error creating garbage collected wal: %w", err) @@ -440,7 +410,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { // We might have crashed right after a sealing block was persisted to storage, // but before the WAL was garbage collected. // In that case, we need to garbage collect the WAL to remove all entries from previous epochs. - if err := i.maybeGarbageCollectWAL(lastBlock); err != nil { + if err := i.maybeGarbageCollectWAL(); err != nil { return simplex.EpochConfig{}, err } @@ -455,7 +425,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { MaxBlockBuildingWaitTime: i.Config.ParameterConfig.MaxNetworkDelay, Logger: i.Config.Logger, Signer: i.Config.CryptoOps, - GenesisValidatorSet: genesisValidatorSet, + GenesisValidatorSet: i.Config.PlatformChain.GenesisValidatorSet(), LastNonSimplexBlockPChainHeight: i.Config.PlatformChain.LastNonSimplexBlockPChainHeight(), SignatureAggregatorCreator: i.Config.CryptoOps.CreateSignatureAggregator, BlockBuilder: i.Config.VM, @@ -480,26 +450,19 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { blockBuilder := &BlockBuilderWaiter{vm: i.Config.VM, msm: msm} - comm := &Communication{Sender: i.Config.Sender, Broadcaster: i.Config.Broadcaster} - comm.SetValidators(nodes) - - epochAwareStorage := &EpochAwareStorage{ - msm: msm, - epoch: epochNum, - Storage: i.cs, - onEpochChange: func(epoch uint64, validators common.Nodes) error { - blockBuilder.stop() - comm.SetValidators(validators) - i.notifyEpochChange(epoch, validators, validator) - return nil - }, - } + comm := newCommunication(i.Config.Sender, i.Config.Broadcaster, validators) + + instanceStorage := NewInstanceStorage(i.cs, msm, noopOnIndex) + onEpochChange := func(epoch uint64, validators common.Nodes) { + blockBuilder.stop() + i.notifyEpochChange(epoch, validators) + } epochConfig := simplex.EpochConfig{ - Epoch: epochNum, + Epoch: epoch, ReplicationEnabled: true, StartTime: time.Now(), - // TODO: For simpicity, we use the same value for all timeouts. If needed we can expand the config. + // TODO: For simplicity, we use the same value for all timeouts. If needed we can expand the config. MaxProposalWait: i.Config.ParameterConfig.MaxNetworkDelay * 2, // 1 proposal + 1 vote MaxRebroadcastWait: i.Config.ParameterConfig.MaxNetworkDelay * 2, FinalizeRebroadcastTimeout: i.Config.ParameterConfig.MaxNetworkDelay * 2, @@ -512,15 +475,21 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { QCDeserializer: i.Config.CryptoOps, Signer: i.Config.CryptoOps, Verifier: i.Config.CryptoOps, - Storage: epochAwareStorage, + Storage: instanceStorage, Comm: comm, BlockBuilder: blockBuilder, BlockDeserializer: &blockDeserializer{deserializer: i.Config.BlockDeserializer, msm: msm}, + OnSealingBlockIndex: onEpochChange, } return epochConfig, nil } -func (i *Instance) maybeGarbageCollectWAL(lastBlock metadata.StateMachineBlock) error { +func (i *Instance) maybeGarbageCollectWAL() error { + lastBlock, _, err := LastBlock(i.Config.Storage) + if err != nil { + return fmt.Errorf("error retrieving last block: %w", err) + } + if lastBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor != nil { i.Config.Logger.Info("Last block is a sealing block, garbage collecting all WALs preceding it to start a new epoch") // We figure out the round number of the latest block and garbage collect all WALs preceding it. @@ -535,119 +504,21 @@ func (i *Instance) maybeGarbageCollectWAL(lastBlock metadata.StateMachineBlock) return nil } -func (i *Instance) transitionEpochNonValidator(epochChange epochChange) error { - i.lock.Lock() - defer i.lock.Unlock() - - if i.isStopped() { - i.Config.Logger.Info("instance is already stopped, skipping epoch change") - return nil - } - - if !i.iAmValidator(epochChange.validators) { - i.Config.Logger.Debug("Skipping restarting a non-validator because I am not a validator yet") - return nil - } - - // Stop the non-validator before doing anything else, so that we don't process any more messages while we are changing epochs. - i.stopNonValidator() - - return i.startAtEpoch(epochChange.validators, epochChange.epochNum) -} - +// startAtEpoch starts either a validator or non-validator at `epoch“. func (i *Instance) startAtEpoch(validators common.Nodes, epoch uint64) error { - if i.iAmValidator(validators) { - if err := i.startValidator(); err != nil { - i.Config.Logger.Error("Error starting validator on epoch change", zap.Error(err)) - return err - } - return nil + if validators.Contains(i.Config.ID) { + return i.startValidator(epoch, validators) } - if err := i.startNonValidator(epoch, validators); err != nil { - i.Config.Logger.Error("Error starting non-validator on epoch change", zap.Error(err)) - return err - } - return nil + return i.startNonValidator() } -func (i *Instance) transitionEpochValidator(epochChange epochChange) error { - i.lock.Lock() - defer i.lock.Unlock() - - // Stop the epoch before doing anything else, so that we don't process any more messages while we are changing epochs. - i.stopValidator() - // Wipe out the WALs from the config so we won't try to load them again - i.Config.WALs = nil - // On epoch change, garbage collect the WAL to remove all entries from previous epochs. - if err := i.wal.GarbageCollect(math.MaxUint64); err != nil { - i.Config.Logger.Error("Error garbage collecting epoch config on epoch change", zap.Error(err)) - } - - return i.startAtEpoch(epochChange.validators, epochChange.epochNum) -} - -func constructEpochAndValidatorSet(logger common.Logger, lastNonSimplexInnerBlockHeight uint64, genesisValidatorSet metadata.NodeBLSMappings, numBlocks uint64, lastBlock *ParsedBlock, storage Storage) (common.Nodes, uint64, error) { - epochNum := lastBlock.BlockHeader().Epoch - - var validatorSet metadata.NodeBLSMappings - var nodes common.Nodes - - switch { - // If all we have in the ledger is non-Simplex blocks, load the validator set from genesis - case lastNonSimplexInnerBlockHeight+1 == numBlocks: - nodes = validatorSetToNodes(genesisValidatorSet) - epochNum = lastNonSimplexInnerBlockHeight + 1 - logger.Debug("Determined epoch and validator set from genesis (ledger holds only non-Simplex blocks)", - zap.Uint64("epoch", epochNum)) - // If the last block persisted is a sealing block, then we are in the next epoch. - case lastBlock.SealingBlockInfo() != nil: - epochNum = lastBlock.BlockHeader().Seq - nodes = lastBlock.SealingBlockInfo().ValidatorSet - logger.Debug("Determined epoch and validator set from sealing block at tip", - zap.Uint64("epoch", epochNum)) - // Else, we have at least one Simplex block in the ledger, and it's not a sealing block. - default: - // Therefore, the sequence of the sealing block is the epoch number. - sealingBlockSeq := lastBlock.BlockHeader().Epoch - sealingBlock, _, err := storage.GetBlock(sealingBlockSeq) - if err != nil { - return nil, 0, fmt.Errorf("error retrieving sealing block from storage: %w", err) - } - if sealingBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil { - return nil, 0, fmt.Errorf("expected sealing block at seq %d, but got a non-sealing block", sealingBlockSeq) - } - validatorSet = constructValidatorSetFromSealingBlock(&ParsedBlock{StateMachineBlock: sealingBlock}) - nodes = validatorSetToNodes(validatorSet) - logger.Debug("Determined epoch and validator set from sealing block in storage", - zap.Uint64("epoch", epochNum), zap.Uint64("sealingBlockSeq", sealingBlockSeq)) +func GetHighestValidatorSet(platform PlatformChain) (common.Nodes, error) { + height := platform.GetCurrentHeight() + mappings, err := platform.GetValidatorSet(height) + if err != nil { + return nil, err } - return nodes, epochNum, nil -} - -func validatorSetToNodes(validatorSet metadata.NodeBLSMappings) common.Nodes { - var nodes common.Nodes - for i := range validatorSet { - vdr := &validatorSet[i] - nodes = append(nodes, common.Node{ - Id: vdr.NodeID[:], - Weight: vdr.Weight, - PK: vdr.BLSKey, - }) - } - return nodes -} -func constructValidatorSetFromSealingBlock(lastBlock *ParsedBlock) metadata.NodeBLSMappings { - var validatorSet metadata.NodeBLSMappings - vdrs := lastBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor.AggregatedMembership.Members - for i := range vdrs { - vdr := &vdrs[i] - validatorSet = append(validatorSet, metadata.NodeBLSMapping{ - NodeID: vdr.NodeID, - BLSKey: vdr.BLSKey, - Weight: vdr.Weight, - }) - } - return validatorSet + return mappings.Nodes(), nil } diff --git a/instance_test.go b/instance_test.go index e271fcd9..dfa281d4 100644 --- a/instance_test.go +++ b/instance_test.go @@ -124,9 +124,7 @@ func TestInstanceNonValidatorBootstraps(t *testing.T) { // One node is a validator and progresses the chain by building blocks, // and its weight changes while the chain progresses in 3 different P-chain epoch heights. // Then, we add another node which is a non-validator. - // The node should bootstrap the chain but without shutting down the non-validator instance, - // and the test should detect the log entry "I am still a non-validator at the tip of the P-chain, skipping role change" - // being printed several times until the non-validator node bootstraps. + // The node should bootstrap the chain but without shutting down the non-validator instance. // Later on, the non-validator becomes a validator. const ( basePChainHeight = uint64(1) @@ -179,17 +177,11 @@ func TestInstanceNonValidatorBootstraps(t *testing.T) { validatorInstance := newInstance(t, validatorNodeID, storage, net, pChain, cops, genesisBlock) nonValidatorInstance := newInstance(t, nonValidatorNodeID, storage2, net, pChain, cops, genesisBlock) - // Count how many times the non-validator reports that it is still not a validator at the - // tip of the P-chain while it replicates across the sealed epochs. - var stillNonValidatorLogs atomic.Uint64 // transitioned is closed when the node starts a Simplex epoch, i.e. becomes a validator. // The node only ever starts an epoch here as part of its non-validator -> validator // transition. transitioned := make(chan struct{}) nonValidatorInstance.Config.Logger.(*testutil.TestLogger).Intercept(func(entry zapcore.Entry) error { - if strings.Contains(entry.Message, "I am still a non-validator at the tip of the P-chain, skipping role change") { - stillNonValidatorLogs.Add(1) - } if strings.Contains(entry.Message, "Starting Simplex Epoch") { select { case <-transitioned: @@ -228,16 +220,16 @@ func TestInstanceNonValidatorBootstraps(t *testing.T) { require.NoError(t, nonValidatorInstance.Start(t.Context())) t.Cleanup(nonValidatorInstance.Stop) - // The non-validator replicates every sealed epoch. It stays a non-validator throughout, - // so on each sealing block it logs that it is still a non-validator at the tip. + // The non-validator replicates every sealed epoch and stays a non-validator throughout. bootstrapTarget := storage.NumBlocks() waitForNumBlocks(t, storage2, bootstrapTarget) - // The "still a non-validator" message was printed once per sealed epoch it replicated - // through, so once for each of the two epochs the weight changes above sealed. - require.Eventually(t, func() bool { - return stillNonValidatorLogs.Load() >= 2 - }, 20*time.Second, 100*time.Millisecond) + // It replicated through the sealed epochs without becoming a validator. + select { + case <-transitioned: + t.Fatal("non-validator transitioned to validator before joining the set") + default: + } // Now grow the validator set to include the peer at the P-chain tip. pChain.advanceTo(joinEpochP) diff --git a/nonvalidator/non_validator.go b/nonvalidator/non_validator.go index bf704067..2ae53353 100644 --- a/nonvalidator/non_validator.go +++ b/nonvalidator/non_validator.go @@ -54,6 +54,10 @@ type Config struct { // RandomSource is used by the replication state to pick which nodes to // request sequences from. If nil, a cryptographically secure source is used. RandomSource *rand.Rand + + // TransitionToValidator is called when our non-validator indexes the highest known epoch + // and it is in the validator set + TransitionToValidator func(epoch uint64, validators common.Nodes) } type NonValidator struct { @@ -266,6 +270,14 @@ func (n *NonValidator) newFinalizedBlockTask(block common.Block, finalization *c return md.Digest } + if block.SealingBlockInfo() != nil { + // are we the highest validator + highestEpoch, highestValidatorSet := n.epochs.highestEpoch() + if highestValidatorSet.Contains(n.ID) && highestEpoch == md.Seq { + n.TransitionToValidator(block.BlockHeader().Seq, block.SealingBlockInfo().ValidatorSet) + } + } + n.Logger.Info("Verified and Indexed Block", zap.Uint64("Block Seq", md.Seq), zap.Stringer("Block Digest", md.Digest)) n.removeOldSequencesAndEpochs(md.Seq, md.Epoch) diff --git a/simplex/epoch.go b/simplex/epoch.go index 1377da11..845254c1 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -86,6 +86,7 @@ type EpochConfig struct { StartTime time.Time ReplicationEnabled bool RandomSource *rand.Rand + OnSealingBlockIndex func(epoch uint64, validators common.Nodes) } type Epoch struct { @@ -788,6 +789,7 @@ func (e *Epoch) Stop() { e.buildBlockScheduler.Close() e.timeoutHandler.Close() e.replicationState.Close() + e.Logger.Info("Node shutdown complete") } func (e *Epoch) isEpochSealed() bool { @@ -1490,6 +1492,9 @@ func (e *Epoch) indexFinalization(block common.VerifiedBlock, finalization commo e.broadcast(finalizationMsg) e.epochSealed.Store(true) + if e.OnSealingBlockIndex != nil { + e.OnSealingBlockIndex(block.BlockHeader().Seq, block.SealingBlockInfo().ValidatorSet) + } } // We have committed because we have collected a finalization. diff --git a/simplex/epoch_test.go b/simplex/epoch_test.go index a8e4f3e0..08b139ac 100644 --- a/simplex/epoch_test.go +++ b/simplex/epoch_test.go @@ -451,6 +451,53 @@ func TestEpochIndexFinalization(t *testing.T) { storage.WaitForBlockCommit(2) } +// Committing a sealing block's finalization invokes OnSealingBlockIndex +// with the sealing block's seq and validator set. +func TestEpochCallsOnSealingBlockIndex(t *testing.T) { + bb := testutil.NewTestBlockBuilder() + nodes := []NodeID{{1}, {2}, {3}, {4}} + conf, _, storage := testutil.DefaultTestNodeEpochConfig(t, nodes[3], testutil.NewNoopComm(nodes), bb) + + type sealingCall struct { + epoch uint64 + validators Nodes + } + calls := make(chan sealingCall, 1) + conf.OnSealingBlockIndex = func(epoch uint64, validators Nodes) { + calls <- sealingCall{epoch: epoch, validators: validators} + } + + e, err := NewEpoch(conf) + require.NoError(t, err) + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + block := testutil.NewTestBlock(ProtocolMetadata{Round: 0, Seq: 0}, emptyBlacklist) + block.SealingInfo = &SealingBlockInfo{ + ValidatorSet: NodeIDs(nodes).EqualWeightedNodes(), + PrevSealingBlockHash: Digest{1}, + } + + vote, err := testutil.NewTestVote(block, nodes[0]) + require.NoError(t, err) + require.NoError(t, e.HandleMessage(&Message{ + BlockMessage: &BlockMessage{Block: block, Vote: *vote}, + }, nodes[0])) + + sigAggr := e.SignatureAggregatorCreator(conf.Comm.Validators()) + finalization, _ := testutil.NewFinalizationRecord(t, sigAggr, block, nodes[:Quorum(len(nodes))]) + testutil.InjectTestFinalization(t, e, &finalization, nodes[1]) + storage.WaitForBlockCommit(0) + + select { + case call := <-calls: + require.Equal(t, uint64(0), call.epoch) + require.Equal(t, block.SealingInfo.ValidatorSet, call.validators) + case <-time.After(5 * time.Second): + t.Fatal("OnSealingBlockIndex was not called after committing the sealing block") + } +} + func TestEquivocatedBlock(t *testing.T) { // Tests a case where a Byzantine leader equivocates: // it sends block A to the node while the honest majority certifies a different diff --git a/util.go b/util.go new file mode 100644 index 00000000..49a8b9cd --- /dev/null +++ b/util.go @@ -0,0 +1,111 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex + +import ( + "errors" + "fmt" + + "github.com/ava-labs/simplex/common" + metadata "github.com/ava-labs/simplex/msm" + "go.uber.org/zap" +) + +var ( + errNoGenesisBlock = errors.New("no genesis block found in storage") + errNonSealingBlock = errors.New("expected sealing block, got a non-sealing block") +) + +// LastBlock returns the last block in storage along with the total number of blocks. +func LastBlock(storage Storage) (metadata.StateMachineBlock, uint64, error) { + numBlocks := storage.NumBlocks() + if numBlocks == 0 { + return metadata.StateMachineBlock{}, 0, errNoGenesisBlock + } + + lastBlock, _, err := storage.GetBlock(numBlocks - 1) + if err != nil { + return metadata.StateMachineBlock{}, 0, fmt.Errorf("error retrieving last block from storage: %w", err) + } + + return lastBlock, numBlocks, nil +} + +// getLastAcceptedEpoch determines the epoch the instance should start at based on +// the last block in storage. If the ledger only contains non-Simplex blocks, the +// epoch is the first Simplex height. If the last block is a sealing block, the +// epoch it seals has ended, so the next epoch is returned. Otherwise, the epoch +// of the last block is returned. +func getLastAcceptedEpochAndValidatorSet(config *Config) (common.Nodes, uint64, error) { + lastBlock, numBlocks, err := LastBlock(config.Storage) + if err != nil { + return nil, 0, fmt.Errorf("error retrieving last block: %w", err) + } + + lastNonSimplexHeight := config.LastNonSimplexInnerBlock.Height() + parsedLastBlock := ParsedBlock{StateMachineBlock: lastBlock} + epochNum := parsedLastBlock.BlockHeader().Epoch + genesisValidatorSet := config.PlatformChain.GenesisValidatorSet() + + var validatorSet metadata.NodeBLSMappings + var nodes common.Nodes + + switch { + // If all we have in the ledger is non-Simplex blocks, load the validator set from genesis + case lastNonSimplexHeight+1 == numBlocks: + nodes = validatorSetToNodes(genesisValidatorSet) + epochNum = lastNonSimplexHeight + 1 + config.Logger.Debug("Determined epoch and validator set from genesis (ledger holds only non-Simplex blocks)", + zap.Uint64("epoch", epochNum)) + // If the last block persisted is a sealing block, then we are in the next epoch. + case lastBlock.SealingBlockInfo() != nil: + epochNum = parsedLastBlock.BlockHeader().Seq + nodes = lastBlock.SealingBlockInfo().ValidatorSet + config.Logger.Debug("Determined epoch and validator set from sealing block at tip", + zap.Uint64("epoch", epochNum)) + // Else, we have at least one Simplex block in the ledger, and it's not a sealing block. + default: + // Therefore, the sequence of the sealing block is the epoch number. + sealingBlockSeq := parsedLastBlock.BlockHeader().Epoch + sealingBlock, _, err := config.Storage.GetBlock(sealingBlockSeq) + if err != nil { + return nil, 0, fmt.Errorf("error retrieving sealing block from storage: %w", err) + } + if sealingBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor == nil { + return nil, 0, fmt.Errorf("%w at seq %d", errNonSealingBlock, sealingBlockSeq) + } + validatorSet = constructValidatorSetFromSealingBlock(&ParsedBlock{StateMachineBlock: sealingBlock}) + nodes = validatorSetToNodes(validatorSet) + config.Logger.Debug("Determined epoch and validator set from sealing block in storage", + zap.Uint64("epoch", epochNum), zap.Uint64("sealingBlockSeq", sealingBlockSeq)) + } + return nodes, epochNum, nil +} + +func validatorSetToNodes(validatorSet metadata.NodeBLSMappings) common.Nodes { + var nodes common.Nodes + for i := range validatorSet { + vdr := &validatorSet[i] + nodes = append(nodes, common.Node{ + Id: vdr.NodeID[:], + Weight: vdr.Weight, + PK: vdr.BLSKey, + }) + } + return nodes +} + +func constructValidatorSetFromSealingBlock(lastBlock *ParsedBlock) metadata.NodeBLSMappings { + var validatorSet metadata.NodeBLSMappings + vdrs := lastBlock.Metadata.SimplexEpochInfo.BlockValidationDescriptor.AggregatedMembership.Members + for i := range vdrs { + vdr := &vdrs[i] + validatorSet = append(validatorSet, metadata.NodeBLSMapping{ + NodeID: vdr.NodeID, + BLSKey: vdr.BLSKey, + Weight: vdr.Weight, + }) + } + return validatorSet +} diff --git a/util_test.go b/util_test.go new file mode 100644 index 00000000..498f268f --- /dev/null +++ b/util_test.go @@ -0,0 +1,198 @@ +package simplex + +import ( + "context" + "errors" + "testing" + + "github.com/ava-labs/simplex/avalanchego" + "github.com/ava-labs/simplex/common" + metadata "github.com/ava-labs/simplex/msm" + "github.com/ava-labs/simplex/testutil" + "github.com/stretchr/testify/require" +) + +// stubStorage is a minimal Storage for exercising util functions. +// When err is set, GetBlock fails at seq errSeq. +type stubStorage struct { + blocks []metadata.StateMachineBlock + errSeq uint64 + err error +} + +func (s *stubStorage) NumBlocks() uint64 { + return uint64(len(s.blocks)) +} + +func (s *stubStorage) GetBlock(seq uint64) (metadata.StateMachineBlock, *common.Finalization, error) { + if s.err != nil && seq == s.errSeq { + return metadata.StateMachineBlock{}, nil, s.err + } + return s.blocks[seq], &common.Finalization{}, nil +} + +func (s *stubStorage) Index(context.Context, common.VerifiedBlock, common.Finalization) error { + return nil +} + +// nonSimplexBlock returns a pre-fork block holding only an inner block at the given height. +func nonSimplexBlock(height uint64) metadata.StateMachineBlock { + return metadata.StateMachineBlock{InnerBlock: &testInnerBlock{Height_: height}} +} + +// simplexBlock returns a non-sealing simplex block at the given epoch and seq. +func simplexBlock(epoch, seq uint64) metadata.StateMachineBlock { + return metadata.StateMachineBlock{ + Metadata: metadata.StateMachineMetadata{ + SimplexProtocolMetadata: common.ProtocolMetadata{Epoch: epoch, Seq: seq}, + }, + } +} + +// sealingBlock returns a sealing block at the given epoch and seq whose +// descriptor holds the given validator set. +func sealingBlock(epoch, seq uint64, members []metadata.NodeBLSMapping) metadata.StateMachineBlock { + block := simplexBlock(epoch, seq) + block.Metadata.SimplexEpochInfo.BlockValidationDescriptor = &metadata.BlockValidationDescriptor{ + AggregatedMembership: metadata.AggregatedMembership{Members: members}, + } + return block +} + +func testValidatorSet() metadata.NodeBLSMappings { + return metadata.NodeBLSMappings{ + {NodeID: avalanchego.NodeID{1}, BLSKey: []byte{1, 2}, Weight: 1}, + {NodeID: avalanchego.NodeID{2}, BLSKey: []byte{3, 4}, Weight: 2}, + } +} + +// epochTestConfig derives the last non-Simplex height from the leading +// blocks in storage that carry no Simplex metadata. +func epochTestConfig(t *testing.T, storage *stubStorage, genesisSet metadata.NodeBLSMappings) *Config { + var lastNonSimplexHeight uint64 + for seq, block := range storage.blocks { + if block.Metadata.SimplexProtocolMetadata.Epoch != 0 { + break + } + lastNonSimplexHeight = uint64(seq) + } + return &Config{ + Storage: storage, + PlatformChain: newTestPlatformChain(0, map[uint64]metadata.NodeBLSMappings{0: genesisSet}), + LastNonSimplexInnerBlock: &testInnerBlock{Height_: lastNonSimplexHeight}, + Logger: testutil.MakeLogger(t, 1), + } +} + +// LastBlock errors on empty storage. +func TestLastBlockEmptyStorage(t *testing.T) { + _, _, err := LastBlock(&stubStorage{}) + require.ErrorIs(t, err, errNoGenesisBlock) +} + +// LastBlock wraps GetBlock errors. +func TestLastBlockGetBlockError(t *testing.T) { + sentinel := errors.New("disk corrupted") + storage := &stubStorage{ + blocks: make([]metadata.StateMachineBlock, 3), + errSeq: 2, + err: sentinel, + } + _, _, err := LastBlock(storage) + require.ErrorIs(t, err, sentinel) +} + +// LastBlock returns the block at seq numBlocks-1 and the block count. +func TestLastBlockSuccess(t *testing.T) { + storage := &stubStorage{ + blocks: []metadata.StateMachineBlock{ + nonSimplexBlock(0), + simplexBlock(1, 1), + }, + } + got, numBlocks, err := LastBlock(storage) + require.NoError(t, err) + require.Equal(t, uint64(2), numBlocks) + require.Equal(t, storage.blocks[1], got) +} + +// Covers each branch of getLastAcceptedEpochAndValidatorSet: genesis, +// sealing block at tip, sealing block in storage, and the error paths. +func TestGetLastAcceptedEpochAndValidatorSet(t *testing.T) { + vdrSet := testValidatorSet() + + tests := []struct { + name string + blocks []metadata.StateMachineBlock + expectedEpoch uint64 + expectedNodes common.Nodes + expectedErr error + }{ + { + name: "only non-Simplex blocks starts at first Simplex height with genesis set", + blocks: []metadata.StateMachineBlock{nonSimplexBlock(0)}, + expectedEpoch: 1, + expectedNodes: vdrSet.Nodes(), + }, + { + name: "multiple non-Simplex blocks start at first Simplex height with genesis set", + blocks: []metadata.StateMachineBlock{ + nonSimplexBlock(0), + nonSimplexBlock(1), + nonSimplexBlock(2), + }, + expectedEpoch: 3, + expectedNodes: vdrSet.Nodes(), + }, + { + name: "sealing block at tip starts next epoch with its descriptor set", + blocks: []metadata.StateMachineBlock{ + simplexBlock(1, 1), + sealingBlock(1, 2, vdrSet), + }, + expectedEpoch: 2, + expectedNodes: vdrSet.Nodes(), + }, + { + name: "non-sealing tip keeps its epoch, set loaded from sealing block at seq==epoch", + blocks: []metadata.StateMachineBlock{ + nonSimplexBlock(0), + simplexBlock(1, 1), + sealingBlock(1, 2, vdrSet), + simplexBlock(2, 3), + }, + expectedEpoch: 2, + expectedNodes: vdrSet.Nodes(), + }, + { + name: "empty storage errors", + expectedErr: errNoGenesisBlock, + }, + { + name: "non-sealing block at the sealing seq errors", + blocks: []metadata.StateMachineBlock{ + nonSimplexBlock(0), + simplexBlock(1, 1), + simplexBlock(1, 2), + simplexBlock(2, 3), + }, + expectedErr: errNonSealingBlock, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + storage := &stubStorage{blocks: tt.blocks} + config := epochTestConfig(t, storage, vdrSet) + + nodes, epoch, err := getLastAcceptedEpochAndValidatorSet(config) + if tt.expectedErr != nil { + require.ErrorIs(t, err, tt.expectedErr) + return + } + require.NoError(t, err) + require.Equal(t, tt.expectedEpoch, epoch) + require.Equal(t, tt.expectedNodes, nodes) + }) + } +}