diff --git a/adapters.go b/adapters.go index 5851c987..daedca43 100644 --- a/adapters.go +++ b/adapters.go @@ -20,6 +20,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) } @@ -32,48 +41,53 @@ 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 { +// InstanceStorage is a wrapper around Storage that skips indexing Telocks +// and delegates post-index handling to a caller-provided onIndex hook. +type InstanceStorage struct { // CachedStorage is used to ensure that we prune the cache on Index. *CachedStorage - msm *metadata.StateMachine - onEpochChange func(seq uint64, validators common.Nodes) error - epoch uint64 + msm *metadata.StateMachine + + onIndex func(block *ParsedBlock) error +} + +func NewInstanceStorage(storage *CachedStorage, msm *metadata.StateMachine, onIndex func(block *ParsedBlock) error) *InstanceStorage { + return &InstanceStorage{ + CachedStorage: 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.CachedStorage.Index(ctx, block, certificate); err != nil { + if err := s.CachedStorage.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 bc9cad04..c16f424e 100644 --- a/instance.go +++ b/instance.go @@ -52,25 +52,21 @@ 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 } +// TODO: replace noop-index with a function that checks whether we need to send approvals or auxiliary information +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 @@ -87,8 +83,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), } } @@ -120,16 +116,27 @@ 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.EpochConfig) + if err != nil { + return fmt.Errorf("error creating simplex epoch: %w", err) + } + + epoch.Epoch = epochConfig.Epoch + i.e = epoch + i.epochOrNV = epoch + epochConfig.bbw.e = 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 } @@ -144,35 +151,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{ - CachedStorage: i.cs, - epoch: epochNum, - 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 @@ -181,30 +170,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 + } } } @@ -246,7 +246,7 @@ func (i *Instance) Stop() { close(i.stopCh) } - i.stopValidator() + i.stopValidator(false) i.stopNonValidator() } @@ -258,9 +258,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 } @@ -360,68 +369,41 @@ 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))) + // Hold the lock so the transition cannot interleave with Stop or HandleMessage. + i.lock.Lock() + + if i.isStopped() { + i.lock.Unlock() + i.Config.Logger.Info("instance is already stopped, skipping epoch change") return } - if err != nil { - i.Config.Logger.Error("Error transitioning epoch", zap.Uint8("role", uint8(epochChange.nodeRole)), zap.Error(err)) - i.Stop() - } -} -// 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 *epochConfig) error { - epoch, err := simplex.NewEpoch(epochConfig.EpochConfig) - if err != nil { - return fmt.Errorf("error creating simplex epoch: %w", err) - } - epoch.Epoch = epochConfig.Epoch - i.e = epoch - i.epochOrNV = epoch - epochConfig.bbw.e = 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() (*epochConfig, error) { - lastBlock, _, err := i.lastBlock() - if err != nil { - return nil, err - } - - genesisValidatorSet := i.Config.PlatformChain.GenesisValidatorSet() - nodes, epochNum, err := getLastAcceptedEpochAndValidatorSet(&i.Config) +func (i *Instance) createEpochConfig(epoch uint64, validators common.Nodes) (*epochConfig, error) { + lastBlock, _, err := LastBlock(i.Config.Storage) if err != nil { return nil, err } @@ -450,7 +432,7 @@ func (i *Instance) createEpochConfig() (*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, @@ -475,26 +457,20 @@ func (i *Instance) createEpochConfig() (*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{ - CachedStorage: i.cs, - msm: msm, - epoch: epochNum, - 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) } ec := 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, @@ -507,10 +483,11 @@ func (i *Instance) createEpochConfig() (*epochConfig, error) { QCDeserializer: i.Config.CryptoOps, Signer: i.Config.CryptoOps, Verifier: i.Config.CryptoOps, - Storage: epochAwareStorage, + Storage: instanceStorage, Comm: comm, BlockBuilder: blockBuilder, BlockDeserializer: &blockDeserializer{vm: i.Config.VM, cs: i.cs}, + OnSealingBlockIndex: onEpochChange, } return &epochConfig{ EpochConfig: ec, @@ -533,56 +510,23 @@ 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)) +func GetHighestValidatorSet(platform PlatformChain) (common.Nodes, error) { + height := platform.GetCurrentHeight() + mappings, err := platform.GetValidatorSet(height) + if err != nil { + return nil, err } - return i.startAtEpoch(epochChange.validators, epochChange.epochNum) + return mappings.Nodes(), nil } type epochConfig struct { diff --git a/instance_test.go b/instance_test.go index 8f651bcf..cbcfa144 100644 --- a/instance_test.go +++ b/instance_test.go @@ -190,9 +190,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) @@ -245,17 +243,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: @@ -294,16 +286,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..ca565860 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,20 @@ func (n *NonValidator) newFinalizedBlockTask(block common.Block, finalization *c return md.Digest } + // If we are indexing a sealing block, we may need to transition to become a validator + if block.SealingBlockInfo() != nil { + highestEpoch, highestValidatorSet := n.epochs.highestEpoch() + + // We should only transition to become a validator, if the sealing block is creating the highest + // epoch we have validated. Since we are fetching from the epochs map, we know this epoch has been validated + // either by a threshold of responses, or backwards hash chain validation. + if highestValidatorSet.Contains(n.ID) && highestEpoch == md.Seq { + if n.TransitionToValidator != nil { + 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/nonvalidator/non_validator_test.go b/nonvalidator/non_validator_test.go index 7d57ac4b..68da9414 100644 --- a/nonvalidator/non_validator_test.go +++ b/nonvalidator/non_validator_test.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "slices" + "sync" "testing" "time" @@ -319,6 +320,154 @@ func TestHandleMessages_DuplicateBlock(t *testing.T) { require.NoError(t, nv.HandleMessage(fin.msg, fin.from)) } +// transitionCall records one TransitionToValidator invocation. +type transitionCall struct { + epoch uint64 + validators common.Nodes +} + +// TestNonValidator_CallsTransition asserts TransitionToValidator fires exactly when +// an indexed sealing block opens the highest known epoch and our ID is in its new +// validator set. +func TestNonValidator_CallsTransition(t *testing.T) { + newValidatorID := common.NodeID{5} + joinedSet := append(slices.Clone(testNodes), common.Node{Id: newValidatorID, Weight: 1}) + otherSet := append(slices.Clone(testNodes), common.Node{Id: common.NodeID{6}, Weight: 1}) + + tests := []struct { + name string + setup func(t *testing.T) (*testChain, []*messageInfo) + expectedCalls []transitionCall + }{ + { + name: "joins the new validator set", + setup: func(t *testing.T) (*testChain, []*messageInfo) { + tc := newSeededChain(t, testNodes, 2) + b3 := tc.appendSealing(joinedSet) + b4 := tc.appendBlock() + return tc, []*messageInfo{ + blockMsg(t, b3, testNodes), + finalizationMsg(t, b3, testNodes), + blockMsg(t, b4, joinedSet), + finalizationMsg(t, b4, joinedSet), + } + }, + expectedCalls: []transitionCall{{epoch: 3, validators: joinedSet}}, + }, + { + name: "not in the new validator set", + setup: func(t *testing.T) (*testChain, []*messageInfo) { + tc := newSeededChain(t, testNodes, 2) + b3 := tc.appendSealing(testNodes) + b4 := tc.appendBlock() + return tc, []*messageInfo{ + blockMsg(t, b3, testNodes), + finalizationMsg(t, b3, testNodes), + blockMsg(t, b4, testNodes), + finalizationMsg(t, b4, testNodes), + } + }, + }, + { + // b3 seals an epoch we are not part of, b4 seals the one we join. + // b3 never triggers: either epoch 4 is already known, or epoch 3's + // set does not contain us. + name: "only the highest known epoch triggers", + setup: func(t *testing.T) (*testChain, []*messageInfo) { + tc := newSeededChain(t, testNodes, 2) + b3 := tc.appendSealing(otherSet) + b4 := tc.appendSealing(joinedSet) + b5 := tc.appendBlock() + return tc, []*messageInfo{ + blockMsg(t, b3, testNodes), + finalizationMsg(t, b3, testNodes), + blockMsg(t, b4, otherSet), + finalizationMsg(t, b4, otherSet), + blockMsg(t, b5, joinedSet), + finalizationMsg(t, b5, joinedSet), + } + }, + expectedCalls: []transitionCall{{epoch: 4, validators: joinedSet}}, + }, + { + // A threshold of quorum rounds for b5, the highest sealing block, + // validates epoch 5 before anything indexes. b3 then indexes while a + // higher epoch is already known, so even though we are in b3's set + // only b5 triggers the transition. + name: "highest epoch validated up front", + setup: func(t *testing.T) (*testChain, []*messageInfo) { + tc := newSeededChain(t, testNodes, 2) + b3 := tc.appendSealing(joinedSet) + b4 := tc.appendSealing(otherSet) + b5 := tc.appendSealing(joinedSet) + b6 := tc.appendBlock() + + f5 := tc.newFinalization(b5) + qrMsg := &common.Message{ + ReplicationResponse: &common.ReplicationResponse{ + Data: []common.QuorumRound{{Block: b5, Finalization: &f5}}, + }, + } + + threshold := common.F(len(joinedSet)) + 1 + msgs := make([]*messageInfo, 0, threshold+8) + for i := 0; i < threshold; i++ { + msgs = append(msgs, &messageInfo{msg: qrMsg, from: joinedSet.NodeIDs()[i]}) + } + return tc, append(msgs, + blockMsg(t, b3, testNodes), + finalizationMsg(t, b3, testNodes), + blockMsg(t, b4, joinedSet), + finalizationMsg(t, b4, joinedSet), + blockMsg(t, b5, otherSet), + finalizationMsg(t, b5, otherSet), + blockMsg(t, b6, joinedSet), + finalizationMsg(t, b6, joinedSet), + ) + }, + expectedCalls: []transitionCall{{epoch: 5, validators: joinedSet}}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc, msgs := tt.setup(t) + lastSeq := tc.seq + + var lock sync.Mutex + var calls []transitionCall + + nv, err := NewNonValidator( + Config{ + Storage: tc, + Comm: testutil.NewNoopComm(tc.nodes().NodeIDs()), + Logger: testutil.MakeLogger(t, 1), + SignatureAggregatorCreator: tc.signatureAggregatorCreator, + MaxSequenceWindow: simplex.DefaultMaxRoundWindow, + ID: newValidatorID, + TransitionToValidator: func(epoch uint64, validators common.Nodes) { + lock.Lock() + defer lock.Unlock() + calls = append(calls, transitionCall{epoch: epoch, validators: validators}) + }, + }, + ) + require.NoError(t, err) + defer nv.Stop() + + for _, m := range msgs { + require.NoError(t, nv.HandleMessage(m.msg, m.from)) + } + + tc.WaitForBlockCommit(lastSeq) + + lock.Lock() + defer lock.Unlock() + require.Equal(t, tt.expectedCalls, calls) + }) + } +} + // TestNonValidator_RequestHighestEpochOnStart verifies that a non-validator // starting behind the network issues a replication request for the highest // epoch on startup. diff --git a/simplex/epoch.go b/simplex/epoch.go index 51a04d49..e984870c 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 { @@ -793,6 +794,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 { @@ -1500,6 +1502,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 9a6fb0f2..edc95801 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