diff --git a/adapters.go b/adapters.go index 8ba02318..5851c987 100644 --- a/adapters.go +++ b/adapters.go @@ -11,6 +11,7 @@ import ( "github.com/ava-labs/simplex/common" metadata "github.com/ava-labs/simplex/msm" + "github.com/ava-labs/simplex/simplex" ) type Communication struct { @@ -200,6 +201,7 @@ type BlockBuilderWaiter struct { lock sync.Mutex cancel context.CancelFunc msm *metadata.StateMachine + e *simplex.Epoch vm VM } @@ -221,7 +223,9 @@ func (bw *BlockBuilderWaiter) WaitForPendingBlock(ctx context.Context) { bw.cancel = cancel bw.lock.Unlock() defer cancel() - bw.vm.WaitForPendingBlock(ctx) + + md := bw.e.Metadata() + bw.msm.WaitForPendingBlock(ctx, md) } func (bw *BlockBuilderWaiter) BuildBlock(ctx context.Context, metadata common.ProtocolMetadata, blacklist common.Blacklist) (common.VerifiedBlock, bool) { diff --git a/instance.go b/instance.go index 6bcb5220..558396ae 100644 --- a/instance.go +++ b/instance.go @@ -382,15 +382,15 @@ func (i *Instance) processEpochChange(epochChange epochChange) { // 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) +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() } @@ -417,22 +417,22 @@ func (i *Instance) iAmValidator(nodes common.Nodes) bool { return false } -func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { +func (i *Instance) createEpochConfig() (*epochConfig, error) { lastBlock, numBlocks, err := i.lastBlock() if err != nil { - return simplex.EpochConfig{}, err + return nil, 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 + return nil, err } 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) + return nil, fmt.Errorf("error creating garbage collected wal: %w", err) } i.wal = wal @@ -440,7 +440,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { // 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 { - return simplex.EpochConfig{}, err + return nil, err } msm, err := metadata.NewStateMachine(&metadata.Config{ @@ -466,7 +466,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { GetBlock: i.cs.RetrieveBlock, }) if err != nil { - return simplex.EpochConfig{}, fmt.Errorf("error creating metadata state machine: %w", err) + return nil, fmt.Errorf("error creating metadata state machine: %w", err) } i.msm = msm @@ -474,7 +474,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { source, err := simplex.NewRandomSource() if err != nil { - return simplex.EpochConfig{}, err + return nil, err } blockBuilder := &BlockBuilderWaiter{vm: i.Config.VM, msm: msm} @@ -494,7 +494,7 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { }, } - epochConfig := simplex.EpochConfig{ + ec := simplex.EpochConfig{ Epoch: epochNum, ReplicationEnabled: true, StartTime: time.Now(), @@ -516,7 +516,10 @@ func (i *Instance) createEpochConfig() (simplex.EpochConfig, error) { BlockBuilder: blockBuilder, BlockDeserializer: &blockDeserializer{vm: i.Config.VM, cs: i.cs}, } - return epochConfig, nil + return &epochConfig{ + EpochConfig: ec, + bbw: blockBuilder, + }, nil } func (i *Instance) maybeGarbageCollectWAL(lastBlock metadata.StateMachineBlock) error { @@ -650,3 +653,8 @@ func constructValidatorSetFromSealingBlock(lastBlock *ParsedBlock) metadata.Node } return validatorSet } + +type epochConfig struct { + simplex.EpochConfig + bbw *BlockBuilderWaiter +} diff --git a/instance_test.go b/instance_test.go index 9cc5e443..5d22517f 100644 --- a/instance_test.go +++ b/instance_test.go @@ -21,6 +21,7 @@ import ( "github.com/ava-labs/simplex/avalanchego" "github.com/ava-labs/simplex/common" metadata "github.com/ava-labs/simplex/msm" + "github.com/ava-labs/simplex/simplex" "github.com/ava-labs/simplex/testutil" "github.com/ava-labs/simplex/wal" @@ -120,6 +121,71 @@ func TestInstanceMixedNodeType(t *testing.T) { require.Equal(t, secondInstance.Config.ID, latestValidatorID(t, storage)) } +// emptyVoteRecorder wraps a Broadcaster and signals the first time an empty vote is broadcast. +type emptyVoteRecorder struct { + Broadcaster + got chan struct{} +} + +func (r *emptyVoteRecorder) Broadcast(msg *common.Message) { + if msg.EmptyVoteMessage != nil { + select { + case r.got <- struct{}{}: + default: + } + } + r.Broadcaster.Broadcast(msg) +} + +func TestEpochInvokesMSMWaitForPendingBlock(t *testing.T) { + const basePChainHeight = uint64(1) + + // Two validators, but only one is instantiated. Our node has the smaller ID so it sorts to + // index 0 and is a non-leader for round 1 (LeaderForRound picks index 1%2). The other + // validator is the round leader but is never created, so no block is ever proposed. + var ourID, leaderID [20]byte + ourID[0], leaderID[0] = 0x01, 0x02 + ourNode := common.NodeID(ourID[:]) + + require.NotEqual(t, ourNode, simplex.LeaderForRound([]common.NodeID{ourNode, leaderID[:]}, 1)) // ensure the leader is not our node + + validatorSetsAtHeight := map[uint64]metadata.NodeBLSMappings{ + basePChainHeight: { + {NodeID: ourID, BLSKey: []byte{0xaa}, Weight: 1}, + {NodeID: leaderID, BLSKey: []byte{0xbb}, Weight: 1}, + }, + } + + pChain := newTestPlatformChain(basePChainHeight, validatorSetsAtHeight) + cops := &testCryptoOps{} + genesisBlock := &testInnerBlock{Height_: 0, TS: time.Now(), Payload: []byte("genesis")} + + net := newInMemNetwork(t) + t.Cleanup(net.stop) + + storage := newStorageWithGenesis(t, genesisBlock) + + // A paused VM never has a pending block, so its WaitForPendingBlock blocks until its context + // is cancelled: the MSM must decide to build on its own for the round to make progress. + vm := newTestVM() + vm.pause() + + inst := newInstanceWithVM(t, ourNode, storage, net, pChain, cops, genesisBlock, vm) + + // Capture the empty vote the node broadcasts once it gives up waiting for the leader. + recorder := &emptyVoteRecorder{Broadcaster: inst.Config.Broadcaster, got: make(chan struct{}, 1)} + inst.Config.Broadcaster = recorder + + require.NoError(t, inst.Start(t.Context())) + t.Cleanup(inst.Stop) + + select { + case <-recorder.got: + case <-time.After(10 * time.Second): + require.FailNow(t, "node never broadcast an empty vote, so the Epoch did not drive the MSM's WaitForPendingBlock") + } +} + 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. diff --git a/msm/msm.go b/msm/msm.go index d149515f..d7b94b95 100644 --- a/msm/msm.go +++ b/msm/msm.go @@ -271,6 +271,97 @@ func (sm *StateMachine) maybeInitializeApprovalStore(validatorSet NodeBLSMapping return sm.approvalStore } +// WaitForPendingBlock waits for either the VM to signal that a block is ready to be built, +// or for the state machine to determine that a block should be built immediately due to an epoch transition. +// In the latter case, we only wait up to MaxBlockBuildingWaitTime before returning. +func (sm *StateMachine) WaitForPendingBlock(ctx context.Context, currentRoundMetadata common.ProtocolMetadata) { + if currentRoundMetadata.Seq == 0 { + // This shouldn't happen because we never build the genesis block. + // Just call the VM's WaitForPendingBlock to avoid blocking the state machine. + sm.Logger.Debug("WaitForPendingBlock called with seq 0, which is invalid; forwarding to VM to avoid blocking") + sm.BlockBuilder.WaitForPendingBlock(ctx) + return + } + + // In order to know whether we're transitioning to a new epoch, or should transition to one, + // we need to look at the previous block's metadata. + + prevBlockSeq := currentRoundMetadata.Seq - 1 + + parentBlock, finalization, err := sm.GetBlock(prevBlockSeq, currentRoundMetadata.Prev) + if err != nil { + sm.Logger.Debug("WaitForPendingBlock failed to get block", zap.Uint64("seq", prevBlockSeq), zap.Error(err)) + sm.BlockBuilder.WaitForPendingBlock(ctx) + return + } + + // In case the parent block is a Telock, we want to look at the sealing block's metadata instead. + if parentBlock.Type() == BlockTypeTelock { + sealingBlockSeq := parentBlock.Metadata.SimplexEpochInfo.SealingBlockSeq + var sealingBlock StateMachineBlock + sealingBlock, finalization, err = sm.GetBlock(sealingBlockSeq, common.Digest{}) + if err != nil { + sm.Logger.Debug("Failed retrieving sealing block for previous epoch", zap.Uint64("seq", sealingBlockSeq), zap.Error(err)) + sm.BlockBuilder.WaitForPendingBlock(ctx) + return + } + + // If the sealing block isn't finalized, we need to build a Telock immediately to extend the epoch. + if finalization == nil { + return + } + // Else, err is nil and the sealing block is finalized, + // so we can use the sealing block's metadata to determine whether we should build a block immediately or not. + parentBlock = sealingBlock + } + + currentState := parentBlock.Metadata.SimplexEpochInfo.NextState() + + // We first check if we have obvious signs that we need to build a block immediately: + var shouldObviouslyBuildBlockImmediately bool + + switch currentState { + case stateFirstSimplexBlock, stateBuildCollectingApprovals: + // In case of the first simplex block we need to persist the validator set, + // and if we're collecting approvals then we're in the process of an epoch change already. + shouldObviouslyBuildBlockImmediately = true + case stateBuildBlockEpochSealed: + // The parent is a sealing block, whose finalization we already fetched above. + // If it isn't finalized we must build a Telock immediately to extend the epoch; + // if it is finalized, we fall through to the normal epoch-transition check below. + if finalization == nil { + shouldObviouslyBuildBlockImmediately = true + } + default: // Handles stateBuildBlockNormalOp or any unknown state, don't do anything and just exit the switch. + } + + // If it's obvious that we should build a block immediately, + // we wait up to MaxBlockBuildingWaitTime and then return immediately. + if shouldObviouslyBuildBlockImmediately { + ctx, cancel := context.WithTimeout(ctx, sm.MaxBlockBuildingWaitTime) + defer cancel() + sm.BlockBuilder.WaitForPendingBlock(ctx) + return + } + + // Otherwise, we might need to build a block if we detect that we should transition to a new epoch, + // so we initialize a blockBuildingDecider and listen while waiting for the VM. + // We return when either the VM signals that a block is ready to be built, + // or that the blockBuildingDecider detects that we should transition to a new epoch. + pChainReferenceHeight := parentBlock.Metadata.SimplexEpochInfo.PChainReferenceHeight + if parentBlock.Type() == BlockTypeSealing { + // We've moved to a new epoch, so we need to use the next P-chain reference height of the sealing block, + // because the P-chain reference height of the next epoch is inherited from the P-chain reference height of the sealing block. + pChainReferenceHeight = parentBlock.Metadata.SimplexEpochInfo.NextPChainReferenceHeight + } + blockBuildingDecider := sm.createBlockBuildingDecider(pChainReferenceHeight) + _, err = blockBuildingDecider.shouldBuildBlock(ctx) + if err != nil { + sm.Logger.Debug("Error while deciding whether to build a block", zap.Error(err)) + return + } +} + // BuildBlock constructs the next block on top of the given parent block, and passes in the provided simplex metadata and blacklist. func (sm *StateMachine) BuildBlock(ctx context.Context, metadata common.ProtocolMetadata, blacklist common.Blacklist) (*StateMachineBlock, error) { // The zero sequence number is reserved for the genesis block, which should never be built. diff --git a/msm/msm_test.go b/msm/msm_test.go index ddbd0df9..4d6be02e 100644 --- a/msm/msm_test.go +++ b/msm/msm_test.go @@ -1957,3 +1957,196 @@ func TestCollectAuxiliaryInfo(t *testing.T) { }) } } + +// blockingBlockBuilder waits in WaitForPendingBlock until it is handed a pending block, the way a +// real VM waits on its mempool. The notification is consumed by a single waiter. +type blockingBlockBuilder struct { + pending chan struct{} +} + +func (b *blockingBlockBuilder) BuildBlock(context.Context, uint64) (avalanchego.VMBlock, error) { + return nil, nil +} + +func (b *blockingBlockBuilder) WaitForPendingBlock(ctx context.Context) { + select { + case <-b.pending: + case <-ctx.Done(): + } +} + +// TestMSMWaitForPendingBlock checks when WaitForPendingBlock stops waiting for the VM. It must +// return on its own if BuildBlock would have produced a block. +// In cases like the first Simplex block, an epoch transition in progress, or a Telock extending a +// sealed epoch, blocks are produced regardless of the VM. +// In other cases, the decision to produce a block is delegated to the VM. +func TestMSMWaitForPendingBlock(t *testing.T) { + const waitTime = 500 * time.Millisecond + + var ( + normal = SimplexEpochInfo{EpochNumber: 1, PChainReferenceHeight: 100} + collecting = SimplexEpochInfo{EpochNumber: 1, PChainReferenceHeight: 100, NextPChainReferenceHeight: 200} + telock = SimplexEpochInfo{EpochNumber: 1, PChainReferenceHeight: 100, NextPChainReferenceHeight: 200, SealingBlockSeq: 5} + sealing = SimplexEpochInfo{EpochNumber: 1, PChainReferenceHeight: 100, NextPChainReferenceHeight: 200, + BlockValidationDescriptor: &BlockValidationDescriptor{}, PrevSealingBlockHash: [32]byte{0xaa}} + ) + + block := func(sei SimplexEpochInfo, finalized bool) *outerBlock { + ob := &outerBlock{block: StateMachineBlock{Metadata: StateMachineMetadata{SimplexEpochInfo: sei}}} + if finalized { + ob.finalization = &common.Finalization{} + } + return ob + } + + for _, tt := range []struct { + name string + // blocks seeds the block store with parent blocks, keyed by their sequence. + blocks blockStore + // seq is the sequence of the block we are asking to build (ProtocolMetadata.Seq). + seq uint64 + // validatorSets seeds the validator set retriever, keyed by P-chain reference height. + validatorSets map[uint64]NodeBLSMappings + // vmHasBlock indicates the underlying VM has a pending block ready to build. + vmHasBlock bool + // returnsOnItsOwn is true when WaitForPendingBlock is expected to return without + // needing external cancellation, i.e. the state machine itself has a block to emit. + returnsOnItsOwn bool + }{ + { + // Never happens, we don't build the genesis block. Delegate to the VM as is. + name: "genesis sequence", + }, + { + // The parent is the pre-Simplex genesis block, so the zero block is next. + name: "first simplex block", + seq: 1, + returnsOnItsOwn: true, + }, + { + // The state is unknown, so keep waiting rather than force a block. + name: "parent block cannot be retrieved", + seq: 2, + }, + { + name: "collecting approvals for the next epoch", + seq: 2, + blocks: blockStore{1: block(collecting, false)}, + returnsOnItsOwn: true, + }, + { + // A Telock has to be emitted to extend the epoch, and it never carries an inner block. + name: "sealing block parent, not finalized", + seq: 2, + blocks: blockStore{1: block(sealing, false)}, + returnsOnItsOwn: true, + }, + { + // The new epoch is open, so this is an ordinary block again. + name: "sealing block parent, finalized", + seq: 2, + blocks: blockStore{1: block(sealing, true)}, + }, + { + // The new epoch inherits its P-chain reference height from the sealing block's + // NextPChainReferenceHeight, so that is the height whose validator set we compare + // against. Comparing against the sealed epoch's height would miss this change. + name: "sealing block parent, finalized, validator set changed again", + seq: 2, + blocks: blockStore{1: block(sealing, true)}, + validatorSets: map[uint64]NodeBLSMappings{200: {{BLSKey: []byte{9}, Weight: 1}}}, + returnsOnItsOwn: true, + }, + { + name: "Telock parent, sealing block not finalized", + seq: 8, + blocks: blockStore{7: block(telock, false), 5: block(sealing, false)}, + returnsOnItsOwn: true, + }, + { + // The Telock we build on is not finalized, but the sealing block it points at is, so + // the epoch is over. Reading the Telock's own finalization would conclude otherwise. + name: "Telock parent, sealing block finalized", + seq: 8, + blocks: blockStore{7: block(telock, false), 5: block(sealing, true)}, + }, + { + // The sealing block cannot be read, so we cannot tell whether the epoch was sealed. + // Rather than force a Telock on incomplete information, we defer to the VM and only + // return once it has a block (or the round is cancelled). + name: "Telock parent, sealing block cannot be retrieved", + seq: 8, + blocks: blockStore{7: block(telock, false)}, + }, + { + name: "normal operation, VM has nothing to build", + seq: 2, + blocks: blockStore{1: block(normal, false)}, + }, + { + name: "normal operation, VM has a pending block", + seq: 2, + blocks: blockStore{1: block(normal, false)}, + vmHasBlock: true, + returnsOnItsOwn: true, + }, + { + // The validator set changed, so a block must record the new P-chain reference height + // even if the VM has nothing to put in it. + name: "normal operation, validator set changed", + seq: 2, + blocks: blockStore{1: block(SimplexEpochInfo{EpochNumber: 1, PChainReferenceHeight: 50}, false)}, + validatorSets: map[uint64]NodeBLSMappings{50: {{BLSKey: []byte{9}, Weight: 1}}}, + returnsOnItsOwn: true, + }, + } { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + bb := &blockingBlockBuilder{pending: make(chan struct{}, 1)} + if tt.vmHasBlock { + bb.pending <- struct{}{} + } + + sm, cfg := newStateMachine(t) + sm.BlockBuilder = bb + sm.MaxBlockBuildingWaitTime = waitTime + cfg.validatorSetRetriever.resultMap = tt.validatorSets + for seq, blk := range tt.blocks { + cfg.blockStore[seq] = blk + } + + md := common.ProtocolMetadata{Seq: tt.seq} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan struct{}) + go func() { + defer close(done) + sm.WaitForPendingBlock(ctx, md) + }() + + if tt.returnsOnItsOwn { + select { + case <-done: + case <-time.After(time.Minute): + require.FailNow(t, "WaitForPendingBlock should have returned on its own") + } + return + } + + select { + case <-done: + require.FailNow(t, "WaitForPendingBlock returned with nothing to build") + case <-time.After(waitTime * 5): + } + + cancel() + select { + case <-done: + case <-time.After(time.Second * 5): + require.FailNow(t, "WaitForPendingBlock ignored cancellation") + } + }) + } +}