bft/nonvalidator/non_validator_test.go

// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved.
// See the file LICENSE for licensing terms.

package nonvalidator

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"testing"
	"time"

	"github.com/luxfi/bft/common"
	"github.com/luxfi/bft/engine"
	"github.com/luxfi/bft/testutil"
	"github.com/stretchr/testify/require"
)

// errQC is a QuorumCertificate whose Verify always fails. Used to drive
// the "qc does not verify" code path without mutating message contents.
type errQC struct {
	common.QuorumCertificate
}

func (errQC) Verify(_ []byte, _ common.Nodes) error {
	return errors.New("qc verification failed")
}

func blockMessage(t *testing.T, block common.Block, from common.NodeID) *common.Message {
	vote, err := testutil.NewTestVote(block, from)
	require.NoError(t, err)
	return &common.Message{
		BlockMessage: &common.BlockMessage{
			Block: block,
			Vote:  *vote,
		},
	}
}

func newBlock(seq, epoch uint64, prev common.Digest) *testutil.TestBlock {
	return testutil.NewTestBlock(common.ProtocolMetadata{
		Round: seq,
		Seq:   seq,
		Epoch: epoch,
		Prev:  prev,
	}, common.Blacklist{})
}

// blockMsg wraps b in a messageInfo sent by the round leader.
func blockMsg(t *testing.T, b common.Block, nodes common.Nodes) *messageInfo {
	leader := engine.LeaderForRound(nodes.NodeIDs(), b.BlockHeader().Round)
	return &messageInfo{
		msg:  blockMessage(t, b, leader),
		from: leader,
	}
}

// finalizationMsg mints a fresh Finalization for b signed by every node in
// nodes, and wraps it in a messageInfo sent by nodes[0]. It takes the block
// (not a pre-built finalization) so callers can chain straight from
// chain.appendBlock without keeping a separate finalization variable.
func finalizationMsg(t *testing.T, b common.VerifiedBlock, nodes common.Nodes) *messageInfo {
	fin, _ := testutil.NewFinalizationRecord(t, &testutil.TestSignatureAggregator{}, b, nodes.NodeIDs())
	return &messageInfo{
		msg:  &common.Message{Finalization: &fin},
		from: nodes.NodeIDs()[0],
	}
}

// TestHandleMessages drives the non-validator with blocks and finalizations
// arriving in various orders and asserts it verifies and indexes them up to
// the expected height, including across epoch transitions.
func TestHandleMessages(t *testing.T) {
	// Each case starts from a chain seeded with seq 0 (genesis), seq 1
	// (sealing block opening epoch 1), and seq 2 (an epoch-1 block).
	tests := []struct {
		name           string
		setup          func(t *testing.T) (*testChain, []*messageInfo)
		expectedHeight uint64
	}{
		{
			name: "blocks and finalizations out of order",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3, b4, b5 := tc.appendBlock(), tc.appendBlock(), tc.appendBlock()

				// finalization for seq 3 arrives last and should kick off
				// verification for seqs 3-5.
				return tc, []*messageInfo{
					blockMsg(t, b4, testNodes),
					blockMsg(t, b3, testNodes),
					finalizationMsg(t, b4, testNodes),
					finalizationMsg(t, b5, testNodes),
					blockMsg(t, b5, testNodes),
					finalizationMsg(t, b3, testNodes),
				}
			},
			expectedHeight: 6,
		},
		{
			name: "nil block doesn't panic",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3 := tc.appendBlock()
				msg := blockMsg(t, b3, testNodes)
				msg.msg.BlockMessage.Block = nil

				return tc, []*messageInfo{msg}
			},
			expectedHeight: 3,
		},
		{
			name: "same block doesn't get verified or indexed twice",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3, b4, b5 := tc.appendBlock(), tc.appendBlock(), tc.appendBlock()

				// sending a second b4 message will try and schedule the verification task for seq 4 again
				// If the task is scheduled twice, the test will fail since storage will panic if we index the same seq twice.
				return tc, []*messageInfo{
					blockMsg(t, b4, testNodes),
					finalizationMsg(t, b4, testNodes),
					blockMsg(t, b4, testNodes),
					blockMsg(t, b3, testNodes),
					finalizationMsg(t, b5, testNodes),
					blockMsg(t, b5, testNodes),
					finalizationMsg(t, b3, testNodes),
				}
			},
			expectedHeight: 6,
		},
		{
			name: "next block followed by finalization",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3 := tc.appendBlock()

				return tc, []*messageInfo{
					blockMsg(t, b3, testNodes),
					finalizationMsg(t, b3, testNodes),
				}
			},
			expectedHeight: 4,
		},
		{
			name: "block message received from non leader",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3 := tc.appendBlock()

				blockMsg := blockMsg(t, b3, testNodes)
				blockMsg.from = testNodes.NodeIDs()[0]
				return tc, []*messageInfo{
					blockMsg,
					finalizationMsg(t, b3, testNodes),
				}
			},
			expectedHeight: 3,
		},
		{
			name: "block digest mismatch with finalization",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3 := tc.appendBlock()
				f3 := finalizationMsg(t, b3, testNodes)

				b3.Digest = common.Digest{}
				blockMsg := blockMsg(t, b3, testNodes)
				return tc, []*messageInfo{
					blockMsg,
					f3,
				}
			},
			expectedHeight: 3,
		},
		{
			name: "qc does not verify",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)
				b3 := tc.appendBlock()
				f3 := finalizationMsg(t, b3, testNodes)
				f3.msg.Finalization.QC = errQC{QuorumCertificate: f3.msg.Finalization.QC}

				return tc, []*messageInfo{
					blockMsg(t, b3, testNodes),
					f3,
				}
			},
			expectedHeight: 3,
		},
		{
			name: "multiple epochs",
			setup: func(t *testing.T) (*testChain, []*messageInfo) {
				tc := newSeededChain(t, testNodes, 2)

				var epoch3Nodes = common.Nodes{
					{Id: common.NodeID{1}, Weight: 1},
					{Id: common.NodeID{2}, Weight: 1},
					{Id: common.NodeID{3}, Weight: 1},
					{Id: common.NodeID{4}, Weight: 1},
					{Id: common.NodeID{5}, Weight: 1},
				}

				// Send the sealing block + finalization that transitions to epoch 3.
				b3 := tc.appendSealing(epoch3Nodes)
				// Blocks 4 and 5 live in epoch 3 and are finalized by the new validator set.
				b4, b5 := tc.appendBlock(), tc.appendBlock()
				// Block 6 is another sealing epoch
				b6 := tc.appendSealing(testNodes)
				// Block 7 is part of the new epoch
				b7 := tc.appendBlock()

				return tc, []*messageInfo{
					blockMsg(t, b3, testNodes),
					finalizationMsg(t, b3, testNodes),
					finalizationMsg(t, b4, epoch3Nodes),
					finalizationMsg(t, b5, epoch3Nodes),
					blockMsg(t, b4, epoch3Nodes),
					blockMsg(t, b5, epoch3Nodes),
					blockMsg(t, b6, epoch3Nodes),
					finalizationMsg(t, b6, epoch3Nodes),
					blockMsg(t, b7, testNodes),
					finalizationMsg(t, b7, testNodes),
				}
			},
			expectedHeight: 8,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tc, msgs := tt.setup(t)
			require.Equal(t, uint64(3), tc.NumBlocks())
			nv, err := NewNonValidator(
				Config{
					Storage:                    tc,
					Comm:                       testutil.NewNoopComm(tc.nodes().NodeIDs()),
					Logger:                     testutil.MakeLogger(t, 1),
					SignatureAggregatorCreator: tc.signatureAggregatorCreator,
					MaxSequenceWindow:          engine.DefaultMaxRoundWindow,
					ID:                         testNodes[0].Id,
				},
			)
			require.NoError(t, err)
			defer nv.Stop()

			for _, m := range msgs {
				require.NoError(t, nv.HandleMessage(m.msg, m.from))
			}

			tc.WaitForBlockCommit(tt.expectedHeight - 1)
			require.Equal(t, tt.expectedHeight, nv.Storage.NumBlocks())
		})
	}
}

func TestNonValidator_StopsGracefully(t *testing.T) {
	tc := newSeededChain(t, testNodes, 2)
	nv, err := NewNonValidator(
		Config{
			Storage:                    tc,
			Comm:                       testutil.NewNoopComm(tc.nodes().NodeIDs()),
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          engine.DefaultMaxRoundWindow,
			ID:                         testNodes[0].Id,
		},
	)
	require.NoError(t, err)

	nv.Start()
	nv.Stop()

	b3 := tc.appendBlock()
	block := blockMsg(t, b3, testNodes)
	require.NoError(t, nv.HandleMessage(block.msg, block.from))
	fin := finalizationMsg(t, b3, testNodes)
	require.NoError(t, nv.HandleMessage(fin.msg, fin.from))

	require.Never(t,
		func() bool {
			return nv.Storage.NumBlocks() > 3
		},
		2*time.Second,
		50*time.Millisecond,
	)
}

// TestHandleMessages_DuplicateBlock tests that when a duplicate block is received, the block is verified & indexed only once
func TestHandleMessages_DuplicateBlock(t *testing.T) {
	tc := newSeededChain(t, testNodes, 2)
	nv, err := NewNonValidator(
		Config{
			Storage:                    tc,
			Comm:                       testutil.NewNoopComm(tc.nodes().NodeIDs()),
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          engine.DefaultMaxRoundWindow,
			ID:                         testNodes[0].Id,
		},
	)
	require.NoError(t, err)

	nv.Start()
	defer nv.Stop()

	// Send the sealing block + finalization that transitions to epoch 3.
	b3 := tc.appendBlock()
	block := blockMsg(t, b3, testNodes)
	require.NoError(t, nv.HandleMessage(block.msg, block.from))
	fin := finalizationMsg(t, b3, testNodes)
	require.NoError(t, nv.HandleMessage(fin.msg, fin.from))

	tc.WaitForBlockCommit(3)

	// Storage will panic if we try indexing the same block twice
	require.NoError(t, nv.HandleMessage(block.msg, block.from))
	require.NoError(t, nv.HandleMessage(fin.msg, fin.from))
}

// TestNonValidator_RequestHighestEpochOnStart verifies that a non-validator
// starting behind the network issues a replication request for the highest
// epoch on startup.
func TestNonValidator_RequestHighestEpochOnStart(t *testing.T) {
	tc := newSeededChain(t, testNodes, 2)
	tc.indexEpochs(4, 8)

	myNodeID := common.NodeID{100}
	comm := &routerComm{
		messageQueue: &messageQueue{},
		ID:           myNodeID,
		nodes:        tc.nodes(),
	}

	nvStorage := tc.CloneUntil(2)
	nv, err := NewNonValidator(
		Config{
			Storage:                    nvStorage,
			Comm:                       comm,
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          engine.DefaultMaxRoundWindow,
			ID:                         myNodeID,
		},
	)

	require.NoError(t, err)

	nv.Start()
	defer nv.Stop()

	msg, ok := comm.messageQueue.popResponse()
	require.True(t, ok)
	require.NotNil(t, msg.msg.ReplicationRequest)
	require.Equal(t, uint64(1), msg.msg.ReplicationRequest.LatestFinalizedSeq)
}

// TestNonValidator_Bootstrap ensures a non-validator can replicate sequences given different states of the chain.
func TestNonValidator_Bootstrap(t *testing.T) {
	tests := []struct {
		name string
		// setup builds the full network chain the non-validator replicates from.
		setup             func(t *testing.T) *testChain
		maxSequenceWindow uint64
		// initialHeight is the number of blocks the non-validator's storage is
		// seeded with before replication begins.
		initialHeight uint64
		lastSeq       uint64
	}{
		{
			// Replicates multiple epochs within a single replication window.
			name: "replicates epochs",
			setup: func(t *testing.T) *testChain {
				tc := newSeededChain(t, testNodes, 2)
				tc.indexEpochs(5, 10, 20, 30, 40)
				return tc
			},
			maxSequenceWindow: 50,
			initialHeight:     2,
			lastSeq:           40,
		},
		{
			// A small window forces replication well past the max round window.
			name: "past max round window",
			setup: func(t *testing.T) *testChain {
				tc := newSeededChain(t, testNodes, 2)
				tc.indexEpochs(5, 10, 20, 30, 40, 50, 60, 80, 100)
				return tc
			},
			maxSequenceWindow: 5, // significantly lower
			initialHeight:     2,
			lastSeq:           100,
		},
		{
			// Storage starts with only the genesis block.
			name: "from genesis",
			setup: func(t *testing.T) *testChain {
				tc := newSeededChain(t, testNodes, 2)
				tc.indexEpochs(5, 10, 20)
				return tc
			},
			maxSequenceWindow: 50,
			initialHeight:     1, // genesis is the only block
			lastSeq:           20,
		},
		{
			// Replicates a chain converted from snowman to engine.
			name: "converted bft chain",
			setup: func(t *testing.T) *testChain {
				tc := newSnowToBFTChain(t, 10)
				firstBlock := tc.appendFirstBFTAfterGenesis(testNodes)
				tc.Index(context.Background(), firstBlock, tc.newFinalization(firstBlock))
				return tc
			},
			maxSequenceWindow: 50,
			initialHeight:     11, // lastSnowmanSeq + 1
			lastSeq:           11,
		},
		{
			// Converted chain that then spans several bft epochs.
			name: "converted bft chain many epochs",
			setup: func(t *testing.T) *testChain {
				tc := newSnowToBFTChain(t, 10)
				firstBlock := tc.appendFirstBFTAfterGenesis(testNodes)
				tc.Index(context.Background(), firstBlock, tc.newFinalization(firstBlock))
				tc.indexEpochs(20, 30)
				return tc
			},
			maxSequenceWindow: 50,
			initialHeight:     11, // lastSnowmanSeq + 1
			lastSeq:           30,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tc := tt.setup(t)
			myNodeID := common.NodeID{16}
			msgQueue := &messageQueue{}
			nonValidatorComm := &routerComm{
				nodes:        tc.nodes(),
				t:            tc.t,
				ID:           myNodeID,
				messageQueue: msgQueue,
			}
			epochs := newTestEpochs(tc, msgQueue, tt.maxSequenceWindow)
			epochs.start()
			defer epochs.stop()

			nvStorage := tc.CloneUntil(tt.initialHeight)
			require.Equal(t, tt.initialHeight, nvStorage.NumBlocks())

			nv, err := NewNonValidator(
				Config{
					Storage:                    nvStorage,
					Comm:                       nonValidatorComm,
					Logger:                     testutil.MakeLogger(t, 1),
					SignatureAggregatorCreator: tc.signatureAggregatorCreator,
					MaxSequenceWindow:          tt.maxSequenceWindow,
					ID:                         myNodeID,
					StartTime:                  time.Now(),
				},
			)
			require.NoError(t, err)

			nv.Start()
			defer nv.Stop()

			advanceUntil(nv, epochs, msgQueue, tt.lastSeq)
		})
	}
}

func TestNonValidator_ReplicationRequests(t *testing.T) {
	tc := newSeededChain(t, testNodes, 2)
	lastSeq := uint64(40)
	initialHeight := uint64(2)
	tc.indexEpochs(5, 10, 20, 30, lastSeq)
	maxSeqWindow := uint64(50)
	myNodeID := common.NodeID{255}
	msgQueue := &messageQueue{}
	nonValidatorComm := &routerComm{
		nodes:        tc.nodes(),
		t:            tc.t,
		ID:           myNodeID,
		messageQueue: msgQueue,
	}
	epochs := newTestEpochs(tc, msgQueue, maxSeqWindow)
	epochs.start()
	defer epochs.stop()

	nvStorage := tc.CloneUntil(initialHeight)
	startTime := time.Now()
	nv, err := NewNonValidator(
		Config{
			Storage:                    nvStorage,
			Comm:                       nonValidatorComm,
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          maxSeqWindow,
			ID:                         myNodeID,
			StartTime:                  startTime,
		},
	)
	require.NoError(t, err)

	nv.Start()
	defer nv.Stop()

	count := 0
	for {
		for msg, ok := msgQueue.popResponse(); ok; {
			// drops 25% of messages
			// TODO: we can handle a higher threshold once we implement https://github.com/ava-labs/Simplex/issues/425
			if count%4 != 0 {
				handleMessage(epochs, nv, msg)
			}
			count++
			msg, ok = msgQueue.popResponse()
		}

		// check if storage has indexed all
		if lastSeq == nv.Storage.NumBlocks()-1 {
			break
		}

		// update the time so pending replication requests time out and re-fire
		startTime = startTime.Add(engine.DefaultReplicationRequestTimeout)
		time.Sleep(50 * time.Millisecond)
		nv.AdvanceTime(startTime)
	}

	// clear in flight messages
	startTime = startTime.Add(engine.DefaultReplicationRequestTimeout)
	nv.AdvanceTime(startTime)
	time.Sleep(50 * time.Millisecond)
	msgQueue.clearResponses()

	// ensure all timeout tasks were removed: advancing time should no longer
	// cause the non-validator to emit any further requests.
	count = 0
	for {
		startTime = startTime.Add(engine.DefaultReplicationRequestTimeout)
		nv.AdvanceTime(startTime)
		time.Sleep(50 * time.Millisecond)

		msg, ok := msgQueue.popResponse()
		require.False(t, ok, fmt.Sprintf("all replication request tasks should be finished %v", msg))

		if count > 3 {
			break
		}
		count++
	}
}

// We optimistically store quorum rounds for sequences that validate an epoch(even though we have not verified the finalization)
// This test ensures, we verify the finalization when it is time to process it. If the finalization is incorrect, we need to re-request
// from the replicator.
func TestNonValidator_VerifiesFinalizationDuringReplication(t *testing.T) {
	tc := newSeededChain(t, testNodes, 2)
	epoch3Nodes := append(testNodes, common.Node{
		Id:     common.NodeID{byte(10)},
		Weight: 1,
	})
	epoch4Nodes := append(epoch3Nodes, common.Node{
		Id:     common.NodeID{byte(11)},
		Weight: 1,
	})

	startTime := time.Now()
	storage := testutil.NewInMemStorage()
	require.NoError(t, storage.Index(context.Background(), genesis, common.Finalization{}))

	nv, err := NewNonValidator(
		Config{
			Storage:                    storage,
			Comm:                       testutil.NewNoopComm(epoch4Nodes.NodeIDs()),
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          5, // significantly lower the max round window
			ID:                         testNodes.NodeIDs()[0],
			StartTime:                  startTime,
		},
	)

	require.NoError(t, err)
	nv.Start()
	defer nv.Stop()

	s3, s4 := tc.appendSealing(epoch3Nodes), tc.appendSealing(epoch4Nodes)
	f3 := tc.newFinalization(s3)
	brokenFinalization := tc.newFinalization(s4)
	brokenFinalization.Finalization.Epoch = 9000
	brokenFinalization.QC = errQC{QuorumCertificate: brokenFinalization.QC}

	brokenQR := common.QuorumRound{
		Block:        s4,
		Finalization: &brokenFinalization,
	}
	brokenReplicationResponse := &common.ReplicationResponse{
		LatestSeq: &brokenQR,
	}

	// With 6 nodes, the threshold votes for validating the highest epoch is  2. Because F + 1 = 2.
	require.NoError(t, nv.HandleMessage(
		&common.Message{
			ReplicationResponse: brokenReplicationResponse,
		},
		testNodes.NodeIDs()[2],
	))
	require.NoError(t, nv.HandleMessage(
		&common.Message{
			ReplicationResponse: brokenReplicationResponse,
		},
		testNodes.NodeIDs()[3],
	))

	// send seq 1, 2, 3
	s1, f1, err := tc.Retrieve(1)
	require.NoError(t, err)
	s2, f2, err := tc.Retrieve(2)
	require.NoError(t, err)

	require.NoError(t, nv.HandleMessage(
		&common.Message{
			ReplicationResponse: &common.ReplicationResponse{
				Data: []common.QuorumRound{
					{
						Block:        s3,
						Finalization: &f3,
					},
					{
						Block:        s1.(common.Block),
						Finalization: &f1,
					},
					{
						Block:        s2.(common.Block),
						Finalization: &f2,
					},
				},
			},
		},
		testNodes.NodeIDs()[3],
	))

	require.Never(t,
		func() bool {
			return storage.NumBlocks() >= 5
		},
		2*time.Second,
		50*time.Millisecond,
	)
}

func TestNonValidatorRejectsQuorumRoundFromNonValidator(t *testing.T) {
	validators := testNodes
	sigAggCreator := func(nodes []common.Node) common.SignatureAggregator {
		return &testutil.TestSignatureAggregator{N: len(nodes)}
	}

	// The attacker-controlled validator set the attacker wants us to adopt.
	// It is completely disjoint from the real validator set, so a block
	// finalized by it must never be committed by a correct non-validator.
	attackerSet := common.Nodes{
		{Id: common.NodeID{200}, Weight: 1},
		{Id: common.NodeID{201}, Weight: 1},
		{Id: common.NodeID{202}, Weight: 1},
	}

	makeZeroBlockQR := func(set common.Nodes) common.QuorumRound {
		block := newSealingTestBlock(1, 1, genesis.Digest, &common.SealingBlockInfo{
			ValidatorSet:         set,
			PrevSealingBlockHash: genesis.Digest,
		})
		fin, _ := testutil.NewFinalizationRecord(t, sigAggCreator(set), block, set.NodeIDs())
		return common.QuorumRound{Block: block, Finalization: &fin}
	}

	threshold := common.F(len(validators)) + 1

	tests := []struct {
		name string
		// epochSet is the validator set embedded in the first-epoch sealing block that the non-validator receives.
		// It is either a forged set or the real set.
		epochSet common.Nodes
		// senders are the peers that vouch for the sealing block.
		senders []common.NodeID
		// expectCommit is whether the non-validator should commit the block.
		expectCommit bool
	}{
		{
			name:         "forged epoch vouched for by non-validators",
			epochSet:     attackerSet,
			senders:      []common.NodeID{{100}, {101}, {102}},
			expectCommit: false,
		},
		{
			name:         "authentic epoch vouched for by validators",
			epochSet:     validators,
			senders:      validators.NodeIDs(),
			expectCommit: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.GreaterOrEqual(t, len(tt.senders), threshold)

			storage := testutil.NewInMemStorage()
			require.NoError(t, storage.Index(context.Background(), genesis, common.Finalization{}))

			nv, err := NewNonValidator(
				Config{
					Storage:                    storage,
					Comm:                       testutil.NewNoopComm(validators.NodeIDs()),
					Logger:                     testutil.MakeLogger(t, 1),
					SignatureAggregatorCreator: sigAggCreator,
					MaxSequenceWindow:          10,
					ID:                         validators.NodeIDs()[0],
					StartTime:                  time.Now(),
				},
			)
			require.NoError(t, err)
			nv.Start()
			defer nv.Stop()

			qr := makeZeroBlockQR(tt.epochSet)

			// Deliver f+1 quorum rounds, each from a distinct sender, all vouching
			// for the same sealing block.
			for i := range threshold {
				require.NoError(t, nv.HandleMessage(&common.Message{
					ReplicationResponse: &common.ReplicationResponse{
						Data: []common.QuorumRound{qr},
					},
				}, tt.senders[i]))
			}

			if tt.expectCommit {
				committed := storage.WaitForBlockCommit(1)
				require.Equal(t, qr.Block.BlockHeader().Digest, committed.BlockHeader().Digest)
				require.Equal(t, uint64(2), storage.NumBlocks())
				return
			}

			require.Never(t,
				func() bool { return storage.NumBlocks() > 1 },
				2*time.Second, 50*time.Millisecond,
				"non-validator committed a block finalized by a validator set that only non-validators vouched for (C2 fork)",
			)
		})
	}
}

func advanceUntil(nv *NonValidator, epochs *testEpochs, msgQueue *messageQueue, seq uint64) {
	startTime := nv.StartTime
	for {
		// Send any requests as responses back to the node
		for msg, ok := msgQueue.popResponse(); ok; {
			handleMessage(epochs, nv, msg)
			msg, ok = msgQueue.popResponse()
		}

		// check if storage has indexed all
		if seq == nv.Storage.NumBlocks()-1 {
			break
		}

		// update the time
		startTime = startTime.Add(engine.DefaultReplicationRequestTimeout)
		time.Sleep(50 * time.Millisecond)
		nv.AdvanceTime(startTime)
	}
}

func TestNilBlockPanicMissingBlock(t *testing.T) {
	startTime := time.Now()
	storage := testutil.NewInMemStorage()
	require.NoError(t, storage.Index(context.Background(), genesis, common.Finalization{}))

	tc := newSeededChain(t, testNodes, 2)

	nv, err := NewNonValidator(
		Config{
			Storage:                    storage,
			Comm:                       testutil.NewNoopComm(testNodes.NodeIDs()),
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          5,
			ID:                         testNodes.NodeIDs()[0],
			StartTime:                  startTime,
		},
	)
	require.NoError(t, err)
	nv.Start()
	defer nv.Stop()

	malicious := common.QuorumRound{
		Block:             nil,
		EmptyNotarization: &common.EmptyNotarization{},
		Finalization:      &common.Finalization{},
	}

	resp := &common.ReplicationResponse{
		Data: []common.QuorumRound{malicious},
	}

	require.NoError(t, nv.HandleMessage(&common.Message{ReplicationResponse: resp}, testNodes.NodeIDs()[2]))
}

// TestNonValidatorAcceptsProposalFromUnsortedValidatorSet asserts that a non-validator elects the
// same leader as the validators when the validator set it was handed is not in sorted order.
//
// engine.LeaderForRound indexes with round % len(nodes), so it is order sensitive. Validators
// sort their set in Epoch.init, so if the non-validator keeps the set in the order it arrived, the
// two disagree about who leads every round. handleBlock then drops every live proposal as "not
// from the leader of that round" and the non-validator silently degrades to replication only.
func TestNonValidatorAcceptsProposalFromUnsortedValidatorSet(t *testing.T) {
	unsortedNodes := common.Nodes{
		{Id: common.NodeID{4}, Weight: 1},
		{Id: common.NodeID{2}, Weight: 1},
		{Id: common.NodeID{3}, Weight: 1},
		{Id: common.NodeID{1}, Weight: 1},
	}

	// The order the validators run with, and therefore the leader the network actually elects.
	sortedNodes := slices.Clone(unsortedNodes)
	common.SortNodes(sortedNodes)

	// The sealing block carries the set unsorted.
	tc := newSeededChain(t, unsortedNodes, 2)
	nv, err := NewNonValidator(
		Config{
			Storage:                    tc,
			Comm:                       testutil.NewNoopComm(sortedNodes.NodeIDs()),
			Logger:                     testutil.MakeLogger(t, 1),
			SignatureAggregatorCreator: tc.signatureAggregatorCreator,
			MaxSequenceWindow:          engine.DefaultMaxRoundWindow,
			ID:                         common.NodeID{100},
		},
	)
	require.NoError(t, err)

	nv.Start()
	defer nv.Stop()

	b := tc.appendBlock()
	require.NotEqual(t,
		engine.LeaderForRound(unsortedNodes.NodeIDs(), b.BlockHeader().Round),
		engine.LeaderForRound(sortedNodes.NodeIDs(), b.BlockHeader().Round),
		"test requires a round whose leader differs between the two orderings")

	// The proposal arrives from the leader the validators elected.
	block := blockMsg(t, b, sortedNodes)
	require.NoError(t, nv.HandleMessage(block.msg, block.from))
	fin := finalizationMsg(t, b, sortedNodes)
	require.NoError(t, nv.HandleMessage(fin.msg, fin.from))

	require.Eventually(t, func() bool {
		return tc.NumBlocks() == b.BlockHeader().Seq+1
	}, 5*time.Second, 10*time.Millisecond,
		"non-validator never committed the block, having dropped the proposal because it ordered the validator set differently than the validators")
}