diff --git a/adapters.go b/adapters.go index 4abcf515..160c8460 100644 --- a/adapters.go +++ b/adapters.go @@ -79,8 +79,15 @@ func (s *CallbackStorage) Index(ctx context.Context, block common.VerifiedBlock, } // A Telock only extends time until the epoch transition finalizes, so we never index it. + // Report the skip instead of a nil error: a nil error means the block is durable at its + // sequence, and the consensus engine advances its commit state (lastBlock, round) on it. + // Returning nil here would leave that state pointing at a block that is not, and never + // will be, in storage, permanently desynchronizing it from NumBlocks(). + // The block is deliberately left in the CachedStorage cache, since a Telock must remain + // retrievable by digest for as long as its epoch is being extended. if pb.Type() == metadata.BlockTypeTelock { - return nil + return fmt.Errorf("%w: Telocks are never persisted (seq %d, round %d)", + common.ErrBlockNotIndexed, block.BlockHeader().Seq, block.BlockHeader().Round) } if err := s.CachedStorage.Index(ctx, block, certificate); err != nil { diff --git a/adapters_test.go b/adapters_test.go index 84eb8a53..411f660c 100644 --- a/adapters_test.go +++ b/adapters_test.go @@ -110,6 +110,47 @@ func TestCachedStorageRetrieve(t *testing.T) { } } +// TestCallbackStorageReportsTelockSkip asserts that CallbackStorage reports the skip when it +// declines to persist a Telock, rather than returning a nil error. +// +// A nil error from Index means the block is durable at its sequence, and the consensus engine +// commits its in-memory state (its last committed block and its round) on that basis. Reporting +// success for a Telock, which is never persisted, would leave that state pointing at a block +// absent from storage, permanently out of step with NumBlocks(), while the block that does +// belong at the sequence can no longer be committed. +func TestCallbackStorageReportsTelockSkip(t *testing.T) { + var indexed []uint64 + cs := NewCachedStorage(NewMockStorage(t)) + storage := NewCallbackStorage(cs, nil, func(block *ParsedBlock) error { + indexed = append(indexed, block.BlockHeader().Seq) + return nil + }) + + // A normal block is persisted and the callback runs. + require.NoError(t, storage.Index(t.Context(), newTestParsedBlock(0, "normal"), common.Finalization{})) + require.Equal(t, uint64(1), storage.NumBlocks()) + require.Equal(t, []uint64{0}, indexed) + + // A Telock only extends its epoch until the sealing block finalizes, so it is never persisted. + telock := newTestParsedBlock(1, "telock") + telock.Metadata.SimplexEpochInfo.SealingBlockSeq = 1 + require.Equal(t, metadata.BlockTypeTelock, telock.Type()) + + err := storage.Index(t.Context(), telock, common.Finalization{}) + require.ErrorIs(t, err, common.ErrBlockNotIndexed, "declining to persist a Telock must be reported, not hidden behind a nil error") + require.Equal(t, uint64(1), storage.NumBlocks(), "the Telock must not be persisted") + require.Equal(t, []uint64{0}, indexed, "the post-index callback must not run for a block that was not indexed") + + // The Telock stays retrievable by digest: its epoch is still being extended. + cachedTelock := &cachedBlock{ParsedBlock: telock, cache: cs} + _, err = cachedTelock.Verify(t.Context(), common.OnlyVMVerifyOpt) + require.NoError(t, err) + require.ErrorIs(t, storage.Index(t.Context(), telock, common.Finalization{}), common.ErrBlockNotIndexed) + got, _, err := cs.Retrieve(1, telock.Digest()) + require.NoError(t, err) + require.Equal(t, telock.BlockHeader().Digest, got.BlockHeader().Digest) +} + // TestCachedStorageIndexEvictsSameSeqFork asserts that once a seq is indexed, // a zero-digest Retrieve of that seq returns the finalized block with its // finalization, even when a verified fork at the same seq was cached. diff --git a/common/api.go b/common/api.go index 51bfbc95..40d36e2a 100644 --- a/common/api.go +++ b/common/api.go @@ -50,11 +50,23 @@ type BlockBuilder interface { var ErrBlockNotFound = fmt.Errorf("block not found") +// ErrBlockNotIndexed is returned by Storage.Index when the implementation deliberately +// declined to persist the block. It is not a failure, but it is also not a commit: +// the block is not durable and NumBlocks() has not advanced, so the caller must not +// treat it as committed. +var ErrBlockNotIndexed = fmt.Errorf("block was not indexed") + type Storage interface { NumBlocks() uint64 // Retrieve returns the block and finalization at [seq]. // If [seq] the block cannot be found, returns ErrBlockNotFound. Retrieve(seq uint64) (VerifiedBlock, Finalization, error) + // Index durably persists [block] and [certificate]. + // A nil error means the block is stored at its sequence, hence NumBlocks() has advanced + // past it by the time Index returns; callers rely on NumBlocks() reflecting Index + // synchronously to know which sequence to commit next. + // An implementation that intentionally does not persist a block must return + // ErrBlockNotIndexed, never nil, otherwise the caller would wrongly consider it committed. Index(ctx context.Context, block VerifiedBlock, certificate Finalization) error } diff --git a/nonvalidator/non_validator.go b/nonvalidator/non_validator.go index ca565860..01b6900b 100644 --- a/nonvalidator/non_validator.go +++ b/nonvalidator/non_validator.go @@ -6,6 +6,7 @@ package nonvalidator import ( "bytes" "context" + "errors" "fmt" "math/rand/v2" "sync" @@ -265,6 +266,19 @@ func (n *NonValidator) newFinalizedBlockTask(block common.Block, finalization *c } if err := n.Storage.Index(n.ctx, verifiedBlock, *finalization); err != nil { + // The storage may deliberately decline to persist a block. That is not a failure, + // but the block is not committed either: nextSeqToCommit has not advanced, so we + // must not act as if this sequence is now accepted. + if errors.Is(err, common.ErrBlockNotIndexed) { + n.Logger.Info("Storage declined to index block, not committing it", zap.Uint64("Block Seq", md.Seq), zap.Stringer("Block Digest", md.Digest), zap.Error(err)) + // This block will never be persisted, so drop what we hold for its sequence and + // ask for that sequence again. Keeping it would make us reject the block that does + // belong there as a duplicate, and treat its finalization as conflicting with the + // one we hold - which halts us. + delete(n.incompleteSequences, md.Seq) + n.sequenceReplicator.ResendFinalizationRequest(md.Seq, finalization.QC.Signers()) + return md.Digest + } n.haltedError = err n.Logger.Info("Failed indexing a block and finalization", zap.Uint64("Block Seq", md.Seq), zap.Stringer("Block Digest", md.Digest), zap.Error(err)) return md.Digest diff --git a/nonvalidator/not_indexed_test.go b/nonvalidator/not_indexed_test.go new file mode 100644 index 00000000..34e8a7f9 --- /dev/null +++ b/nonvalidator/not_indexed_test.go @@ -0,0 +1,100 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package nonvalidator + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/ava-labs/simplex/common" + "github.com/ava-labs/simplex/simplex" + "github.com/ava-labs/simplex/testutil" + "github.com/stretchr/testify/require" +) + +// decliningStorage wraps a storage that never persists one specific block, reporting the skip +// with common.ErrBlockNotIndexed. In production that block is a Telock: it takes up a sequence +// in the dying epoch to extend it until the sealing block finalizes, and is superseded by the +// first block of the next epoch. +type decliningStorage struct { + common.Storage + declined common.Digest +} + +func (d *decliningStorage) Index(ctx context.Context, block common.VerifiedBlock, certificate common.Finalization) error { + if block.BlockHeader().Digest == d.declined { + return fmt.Errorf("%w: this block is never persisted", common.ErrBlockNotIndexed) + } + return d.Storage.Index(ctx, block, certificate) +} + +// TestNonValidatorRecoversFromBlockTheStorageDeclinedToIndex asserts that a block the storage +// refuses to persist does not stop the non-validator from accepting the block that does belong +// at that sequence. +// +// removeOldSequencesAndEpochs, which only runs after a real commit, is the only thing that +// clears an entry from incompleteSequences. Leaving the refused block there makes handleBlock +// drop the real block as a duplicate and makes handleFinalization treat its finalization as +// conflicting with the one being held - which halts the non-validator permanently. +func TestNonValidatorRecoversFromBlockTheStorageDeclinedToIndex(t *testing.T) { + const declinedSeq = uint64(3) + + tc := newSeededChain(t, testNodes, declinedSeq-1) + + // Two different blocks claim the same sequence: the one the storage will never persist, + // and the one that really belongs there. + unpersistable := newBlock(declinedSeq, tc.epoch, tc.digest) + unpersistable.Data = []byte("never persisted") + unpersistable.ComputeDigest() + + nv, err := NewNonValidator( + Config{ + Storage: &decliningStorage{Storage: tc, declined: unpersistable.Digest}, + Comm: testutil.NewNoopComm(tc.nodes().NodeIDs()), + Logger: testutil.MakeLogger(t, 1), + SignatureAggregatorCreator: tc.signatureAggregatorCreator, + MaxSequenceWindow: simplex.DefaultMaxRoundWindow, + ID: testNodes[0].Id, + }, + ) + require.NoError(t, err) + + nv.Start() + defer nv.Stop() + + block := blockMsg(t, unpersistable, testNodes) + require.NoError(t, nv.HandleMessage(block.msg, block.from)) + finalization := finalizationMsg(t, unpersistable, testNodes) + require.NoError(t, nv.HandleMessage(finalization.msg, finalization.from)) + + // The refused block must not be left behind as what we hold for that sequence. + require.Eventually(t, func() bool { + nv.lock.Lock() + defer nv.lock.Unlock() + _, exists := nv.incompleteSequences[declinedSeq] + return !exists + }, 30*time.Second, 10*time.Millisecond, "the block the storage refused must be dropped, not held for its sequence") + require.Equal(t, declinedSeq, tc.NumBlocks(), "the refused block must not be persisted") + + // The block that does belong at that sequence is still accepted and committed. + realBlock := tc.appendBlock() + block = blockMsg(t, realBlock, testNodes) + require.NoError(t, nv.HandleMessage(block.msg, block.from)) + finalization = finalizationMsg(t, realBlock, testNodes) + require.NoError(t, nv.HandleMessage(finalization.msg, finalization.from)) + + require.Eventually(t, func() bool { + return tc.NumBlocks() == declinedSeq+1 + }, 30*time.Second, 10*time.Millisecond, "the block that belongs at the sequence should have been committed") + + committed, _, err := tc.Retrieve(declinedSeq) + require.NoError(t, err) + require.Equal(t, realBlock.BlockHeader().Digest, committed.BlockHeader().Digest) + + nv.lock.Lock() + defer nv.lock.Unlock() + require.NoError(t, nv.haltedError, "the non-validator must not halt over a block the storage refused") +} diff --git a/simplex/epoch.go b/simplex/epoch.go index 7e3c95a6..bee25ec2 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -804,6 +804,26 @@ func (e *Epoch) handleFinalizationMessage(message *common.Finalization, from com e.Logger.Verbo("Received finalization message", zap.Stringer("from", from), zap.Uint64("round", message.Finalization.Round), zap.Uint64("seq", message.Finalization.Seq)) + // Once our epoch is sealed nothing beyond the sealing block belongs to it, + // so a finalization for a higher sequence can only be for a block of this epoch + // that will never be committed (it is superseded by the next epoch). + if e.isEpochSealed() { + e.Logger.Debug("Epoch is sealed, ignoring finalization message", + zap.Stringer("from", from), zap.Uint64("round", message.Finalization.Round), zap.Uint64("seq", message.Finalization.Seq)) + return nil + } + + // A finalization for a block of an earlier epoch cannot be committed by this epoch: + // every sequence we still have to commit belongs to our epoch or to a later one. This keeps + // a stale epoch transition artifact - a block of a sealed epoch that was never persisted - + // from claiming a round of ours; it is a layer on top of, not a substitute for, the storage + // refusing to index such a block, since e.Epoch is only as trustworthy as what we restored. + if message.Finalization.Epoch < e.Epoch { + e.Logger.Debug("Received a finalization from a previous epoch", + zap.Stringer("from", from), zap.Uint64("finalization epoch", message.Finalization.Epoch), zap.Uint64("our epoch", e.Epoch)) + return nil + } + nextSeqToCommit := e.nextSeqToCommit() // Ignore finalizations for sequences we have already committed if nextSeqToCommit > message.Finalization.Seq { @@ -1458,9 +1478,24 @@ func (e *Epoch) indexFinalizations(startRound uint64) error { finalization := *round.finalization block := round.block - if err := e.indexFinalization(block, finalization); err != nil { + committed, err := e.indexFinalization(block, finalization) + if err != nil { return err } + if !committed { + // The block was not persisted, so nextSeqToCommit did not advance and neither this + // sequence nor any sequence after it can be committed now. + // + // Drop the round: a block the storage will not persist can never be committed, so + // keeping it would only let us serve it to replicating peers as a finalized quorum + // round for this sequence (locateQuorumRecord), build our next proposal on top of it + // (metadata, via getHighestRound), and refuse the block that does belong at this + // round (storeProposal refuses a round that already has an entry). + delete(e.rounds, round.num) + e.Logger.Debug("Stopped indexing finalizations because the block was not committed", + zap.Uint64("round", round.num), zap.Uint64("seq", finalization.Finalization.Seq)) + return nil + } e.deleteRounds(round.num) // Clean up the future messages - Remove all messages we may have stored for all rounds until this round @@ -1476,10 +1511,39 @@ func (e *Epoch) indexFinalizations(startRound uint64) error { return nil } -func (e *Epoch) indexFinalization(block common.VerifiedBlock, finalization common.Finalization) error { +// indexFinalization commits [block] and [finalization] to storage. +// It returns whether the block was actually committed: a Storage may deliberately decline to +// persist a block (signalled with common.ErrBlockNotIndexed), and in that case the epoch must +// not advance its commit state. e.lastBlock and e.round would otherwise permanently diverge +// from nextSeqToCommit(), which is derived from the storage itself, leaving the epoch building +// on and advertising a block that is not in storage and can never be committed. +func (e *Epoch) indexFinalization(block common.VerifiedBlock, finalization common.Finalization) (bool, error) { + seq := finalization.Finalization.Seq + if err := e.Storage.Index(e.finishCtx, block, finalization); err != nil { - return err + if errors.Is(err, common.ErrBlockNotIndexed) { + e.Logger.Info("Storage declined to index block, not committing it", + zap.Uint64("round", finalization.Finalization.Round), + zap.Uint64("sequence", seq), + zap.Stringer("digest", finalization.Finalization.Digest), + zap.Error(err)) + return false, nil + } + return false, err } + + // A nil error from Index means the block is durable at [seq]. Enforce that invariant + // instead of trusting it, as committing in-memory state for a block the storage did not + // persist cannot be undone by any later message. + if numBlocks := e.Storage.NumBlocks(); numBlocks <= seq { + e.Logger.Warn("Storage reported a successful index but the block is not stored, not committing it", + zap.Uint64("round", finalization.Finalization.Round), + zap.Uint64("sequence", seq), + zap.Uint64("numBlocks", numBlocks), + zap.Stringer("digest", finalization.Finalization.Digest)) + return false, nil + } + e.Logger.Info("Committed block", zap.Uint64("round", finalization.Finalization.Round), zap.Uint64("sequence", finalization.Finalization.Seq), @@ -1507,7 +1571,7 @@ func (e *Epoch) indexFinalization(block common.VerifiedBlock, finalization commo // However, we may have not witnessed a notarization. // Regardless of that, we can safely progress to the round succeeding the finalization. e.progressRoundsDueToCommit(finalization.Finalization.Round + 1) - return nil + return true, nil } func (e *Epoch) maybeAssembleEmptyNotarization() error { @@ -2241,6 +2305,7 @@ func (e *Epoch) createFinalizedBlockVerificationTask(block common.Block, finaliz } // Store the verified block in rounds map so subsequent blocks can find it as a dependency + previousRoundEntry, hadRoundEntry := e.rounds[md.Round] roundEntry := NewRound(verifiedBlock) e.rounds[md.Round] = roundEntry if err := e.storeFinalization(finalization); err != nil { @@ -2251,11 +2316,33 @@ func (e *Epoch) createFinalizedBlockVerificationTask(block common.Block, finaliz zap.Uint64("seq", md.Seq), zap.Stringer("digest", md.Digest)) - if err := e.indexFinalization(verifiedBlock, *finalization); err != nil { + committed, err := e.indexFinalization(verifiedBlock, *finalization) + if err != nil { e.haltedError = err e.Logger.Error("Failed to index finalization", zap.Error(err)) return md.Digest } + if !committed { + // The storage declined to persist this replicated block, so it is not committed. + // Restore the rounds map to what it was before: keeping the block and its finalization + // there would both misrepresent it as finalized to replicating peers and prevent the + // block that does belong at this round from ever having its finalization stored. + // storeFinalization succeeded above, which means it matched the round's block header, + // so the entry we are undoing is keyed by md.Round. + if hadRoundEntry { + e.rounds[md.Round] = previousRoundEntry + } else { + delete(e.rounds, md.Round) + } + // The caller took this sequence out of the replication state before handing it to us, + // so ask for it again - we still need whichever block does belong at this sequence. + e.replicationState.ResendFinalizationRequest(md.Seq, finalization.QC.Signers()) + e.Logger.Info("Replicated finalized block was not indexed, discarding it", + zap.Uint64("round", md.Round), + zap.Uint64("seq", md.Seq), + zap.Stringer("digest", md.Digest)) + return md.Digest + } err = e.processReplicationState() if err != nil { @@ -3473,7 +3560,27 @@ func (e *Epoch) verifyQuorumRound(q common.QuorumRound) error { return err } + // A block of an earlier epoch is not part of the chain this epoch commits: every sequence we + // still have to commit belongs to our epoch or to a later one. The quorum certificates below + // only attest that the signers form a quorum of our validator set, which a quorum of a + // previous epoch may still satisfy, so without this check a replicating peer could feed us a + // block of a sealed epoch that was never persisted - a Telock, which chains onto the last + // block of that epoch and hence claims the round and sequence the first block of our epoch + // belongs to - and we would mistake it for part of our chain. Both the notarized and the + // finalized replication paths verify with OnlyVM, so neither of them would catch it. + // A quorum round rejected here may carry an empty notarization we do want; replication asks + // again, so we drop the whole round rather than rebuild the caller's message. + if q.Block != nil && q.Block.BlockHeader().Epoch < e.Epoch { + return fmt.Errorf("received a quorum round for a block of the previous epoch %d, our epoch is %d", + q.Block.BlockHeader().Epoch, e.Epoch) + } + if q.Finalization != nil { + if q.Finalization.Finalization.Epoch < e.Epoch { + return fmt.Errorf("received a finalization from a previous epoch %d, our epoch is %d", + q.Finalization.Finalization.Epoch, e.Epoch) + } + // extra check needed if we have a finalized block if err := VerifyQC(q.Finalization.QC, e.signatureAggregator.IsQuorum, e.validatorsToPKs, q.Finalization, e.validators); err != nil { return fmt.Errorf("invalid finalization: %w", err) diff --git a/simplex/epoch_not_indexed_test.go b/simplex/epoch_not_indexed_test.go new file mode 100644 index 00000000..0994f2c6 --- /dev/null +++ b/simplex/epoch_not_indexed_test.go @@ -0,0 +1,275 @@ +// Copyright (C) 2019-2025, Ava Labs, Inc. All rights reserved. +// See the file LICENSE for licensing terms. + +package simplex_test + +import ( + "context" + "fmt" + "testing" + "time" + + . "github.com/ava-labs/simplex/common" + . "github.com/ava-labs/simplex/simplex" + "github.com/ava-labs/simplex/testutil" + + "github.com/stretchr/testify/require" +) + +// decliningStorage wraps an InMemStorage and deliberately does not persist the blocks whose +// sequence is in declined. This mirrors the production storage, which never persists a Telock: +// a block that takes up a sequence in the dying epoch but is superseded by the first block of +// the next epoch. When reportSkip is set the skip is reported with ErrBlockNotIndexed, as the +// Storage contract requires; otherwise the skip is hidden behind a nil error. +type decliningStorage struct { + *testutil.InMemStorage + declined map[uint64]struct{} + reportSkip bool +} + +func (d *decliningStorage) Index(ctx context.Context, block VerifiedBlock, certificate Finalization) error { + if _, declined := d.declined[block.BlockHeader().Seq]; declined { + if d.reportSkip { + return fmt.Errorf("%w: seq %d is never persisted", ErrBlockNotIndexed, block.BlockHeader().Seq) + } + return nil + } + return d.InMemStorage.Index(ctx, block, certificate) +} + +// TestEpochDoesNotCommitBlockTheStorageDeclinedToIndex asserts that a block the storage did +// not persist is not treated as committed. +// +// A nil error from Storage.Index is the engine's only signal that a block is durable at +// Storage.NumBlocks()-1. If the epoch advances e.lastBlock and e.round for a block that is not +// in storage, then nextSeqToCommit() - which is derived from the storage - stays behind, and +// the two can never reconcile: the node builds on, and advertises as finalized, a block that no +// node can ever commit at that sequence. +// +// The epoch is driven with four blocks whose finalizations arrive in order, with the storage +// declining the seq-2 block, both when it reports the skip and when it hides it behind a nil +// error. Only the blocks before it may be committed, the round must not advance past the +// sequence that was not committed, and a replicating peer must be told that the last committed +// block is the one actually in storage. +func TestEpochDoesNotCommitBlockTheStorageDeclinedToIndex(t *testing.T) { + const ( + numBlocks = uint64(4) + declinedSeq = uint64(2) + ) + + for _, tc := range []struct { + name string + reportSkip bool + }{ + {name: "storage reports the skip", reportSkip: true}, + {name: "storage hides the skip behind a nil error", reportSkip: false}, + } { + t.Run(tc.name, func(t *testing.T) { + nodes := []NodeID{{1}, {2}, {3}, {4}, {5}, {6}, {7}, {8}, {9}, {10}} + quorum := Quorum(len(nodes)) + blacklist := Blacklist{NodeCount: uint16(len(nodes)), SuspectedNodes: SuspectedNodes{}, Updates: []BlacklistUpdate{}} + + // The node under test leads a round beyond the ones exercised here, so it never proposes. + epochNode := LeaderForRound(nodes, numBlocks+1) + + comm := &recordingComm{ + Communication: testutil.NewNoopComm(NodeIDs(nodes)), + SentMessages: make(chan *Message, 1000), + BroadcastMessages: make(chan *Message, 1000), + } + conf, _, inMemStorage := testutil.DefaultTestNodeEpochConfig(t, epochNode, comm, testutil.NewTestBlockBuilder()) + conf.ReplicationEnabled = true + storage := &decliningStorage{ + InMemStorage: inMemStorage, + declined: map[uint64]struct{}{declinedSeq: {}}, + reportSkip: tc.reportSkip, + } + conf.Storage = storage + if !tc.reportSkip { + // A storage that breaks the Index contract is reported as a warning, which is + // expected here. + conf.Logger.(*testutil.TestLogger).Silence() + } + + e, err := NewEpoch(conf) + require.NoError(t, err) + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + blocks := make([]*testutil.TestBlock, numBlocks) + var prev Digest + for i := uint64(0); i < numBlocks; i++ { + blocks[i] = testutil.NewTestBlock(ProtocolMetadata{Round: i, Seq: i, Prev: prev}, blacklist) + prev = blocks[i].BlockHeader().Digest + } + + // Deliver every block as a proposal from the node that leads its round. + for i := uint64(0); i < numBlocks; i++ { + leader := LeaderForRound(nodes, i) + require.NoError(t, e.HandleMessage(&Message{ + BlockMessage: &BlockMessage{Block: blocks[i], Vote: mustVote(t, blocks[i], leader)}, + }, leader)) + } + + sigAggr := e.SignatureAggregatorCreator(conf.Comm.Validators()) + + // Finalize the blocks preceding the declined one, they are committed as usual. + for i := uint64(0); i < declinedSeq; i++ { + finalization, _ := testutil.NewFinalizationRecord(t, sigAggr, blocks[i], nodes[:quorum]) + require.NoError(t, e.HandleMessage(&Message{Finalization: &finalization}, nodes[0])) + inMemStorage.WaitForBlockCommit(i) + } + require.Equal(t, declinedSeq, e.Metadata().Round, "the epoch should be in the round following the last committed block") + + // Finalize the declined block, and then the blocks after it. + for i := declinedSeq; i < numBlocks; i++ { + finalization, _ := testutil.NewFinalizationRecord(t, sigAggr, blocks[i], nodes[:quorum]) + require.NoError(t, e.HandleMessage(&Message{Finalization: &finalization}, nodes[0])) + } + + // The declined block is not in storage, so it was not committed, and neither is anything + // after it: the sequence it occupies is still the next sequence to commit. + inMemStorage.EnsureNoBlockCommit(t, declinedSeq) + require.Equal(t, declinedSeq, storage.NumBlocks(), "only the blocks preceding the declined one should be committed") + + // The round must not advance for a block that was not committed, otherwise the epoch + // rejects the proposal that does belong at this sequence. + require.Equal(t, declinedSeq, e.Metadata().Round, "the round must not advance past a block that was not committed") + + // A replicating peer must be told about the last block that is actually committed, + // never about the block that was not persisted - neither as our latest finalized + // sequence nor as a finalized quorum round for the sequence it claimed. + require.NoError(t, e.HandleMessage(&Message{ + ReplicationRequest: &ReplicationRequest{LatestFinalizedSeq: 1, Seqs: []uint64{declinedSeq}}, + }, nodes[1])) + response := awaitReplicationResponse(t, comm) + + require.NotNil(t, response.LatestFinalizedSeq) + require.Equal(t, declinedSeq-1, response.LatestFinalizedSeq.Finalization.Finalization.Seq, + "the epoch must advertise the last committed block as its latest finalized sequence") + require.Equal(t, blocks[declinedSeq-1].BlockHeader().Digest, response.LatestFinalizedSeq.VerifiedBlock.BlockHeader().Digest) + + for _, qr := range append(response.Data, orEmpty(response.LatestRound)...) { + if qr.Finalization == nil { + continue + } + require.NotEqual(t, blocks[declinedSeq].BlockHeader().Digest, qr.VerifiedBlock.BlockHeader().Digest, + "the epoch must never serve a block it did not persist as finalized") + } + }) + } +} + +// TestEpochRejectsFinalizationsFromPreviousEpochs asserts that a finalization for a block of an +// earlier epoch is not committed, whether it arrives as a finalization message or inside a +// replication response. +// +// Every sequence an epoch still has to commit belongs to that epoch or to a later one, so a +// finalization for an earlier epoch can only refer to a block that this epoch will never +// commit - such as a block of the previous epoch that was superseded by the epoch transition +// and was therefore never persisted. Its quorum certificate is not enough to reject it: the +// signers of a previous epoch may still form a quorum of the current validator set. +func TestEpochRejectsFinalizationsFromPreviousEpochs(t *testing.T) { + const ( + previousEpoch = uint64(1) + currentEpoch = uint64(2) + // The sequence of the first block of the current epoch, which the stale block also claims. + contestedSeq = uint64(1) + ) + + nodes := []NodeID{{1}, {2}, {3}, {4}} + quorum := Quorum(len(nodes)) + blacklist := Blacklist{NodeCount: uint16(len(nodes)), SuspectedNodes: SuspectedNodes{}, Updates: []BlacklistUpdate{}} + + comm := &recordingComm{ + Communication: testutil.NewNoopComm(NodeIDs(nodes)), + SentMessages: make(chan *Message, 1000), + } + conf, _, storage := testutil.DefaultTestNodeEpochConfig(t, nodes[0], comm, testutil.NewTestBlockBuilder()) + conf.ReplicationEnabled = true + conf.Epoch = currentEpoch + + // Seed the storage with the last block of the previous epoch, so the epoch starts where the + // previous one left off, as a freshly transitioned node does. + sealingBlock := testutil.NewTestBlock(ProtocolMetadata{Round: 0, Seq: 0, Epoch: previousEpoch}, blacklist) + sealingFinalization, _ := testutil.NewFinalizationRecord(t, &testutil.TestSignatureAggregator{N: len(nodes)}, sealingBlock, nodes[:quorum]) + require.NoError(t, storage.Index(t.Context(), sealingBlock, sealingFinalization)) + + e, err := NewEpoch(conf) + require.NoError(t, err) + e.Epoch = currentEpoch + t.Cleanup(e.Stop) + require.NoError(t, e.Start()) + + require.Equal(t, currentEpoch, e.Epoch) + roundBeforeReplay := e.Metadata().Round + + // A block of the previous epoch that claims the sequence the current epoch is about to + // commit, finalized by a quorum of the (still overlapping) validator set. + staleBlock := testutil.NewTestBlock(ProtocolMetadata{ + Round: roundBeforeReplay, + Seq: contestedSeq, + Epoch: previousEpoch, + Prev: sealingBlock.BlockHeader().Digest, + }, blacklist) + sigAggr := e.SignatureAggregatorCreator(conf.Comm.Validators()) + staleFinalization, _ := testutil.NewFinalizationRecord(t, sigAggr, staleBlock, nodes[:quorum]) + + staleNotarization, err := testutil.NewNotarization(e.Logger, sigAggr, staleBlock, nodes[:quorum]) + require.NoError(t, err) + + // Replay it through every path that could adopt it: a notarized quorum round, a finalized + // quorum round, a proposal, and a plain finalization message. + staleLeader := LeaderForRound(nodes, roundBeforeReplay) + for _, msg := range []*Message{ + {ReplicationResponse: &ReplicationResponse{Data: []QuorumRound{{Block: staleBlock, Notarization: &staleNotarization}}}}, + {ReplicationResponse: &ReplicationResponse{Data: []QuorumRound{{Block: staleBlock, Finalization: &staleFinalization}}}}, + {BlockMessage: &BlockMessage{Block: staleBlock, Vote: mustVote(t, staleBlock, staleLeader)}}, + {Finalization: &staleFinalization}, + } { + from := nodes[1] + if msg.BlockMessage != nil { + from = staleLeader + } + require.NoError(t, e.HandleMessage(msg, from)) + } + + // Storage, the commit cursor and the round are all left untouched. + storage.EnsureNoBlockCommit(t, contestedSeq) + require.Equal(t, contestedSeq, storage.NumBlocks()) + require.Equal(t, roundBeforeReplay, e.Metadata().Round) + require.Equal(t, sealingBlock.BlockHeader().Digest, e.Metadata().Prev, + "the epoch must keep building on the last committed block") +} + +func mustVote(t *testing.T, block testutil.AnyBlock, from NodeID) Vote { + t.Helper() + vote, err := testutil.NewTestVote(block, from) + require.NoError(t, err) + return *vote +} + +// awaitReplicationResponse returns the first replication response the epoch sends. +func awaitReplicationResponse(t *testing.T, comm *recordingComm) *VerifiedReplicationResponse { + t.Helper() + + timeout := time.After(30 * time.Second) + for { + select { + case msg := <-comm.SentMessages: + if msg.VerifiedReplicationResponse != nil { + return msg.VerifiedReplicationResponse + } + case <-timeout: + require.FailNow(t, "timed out waiting for a replication response") + return nil + } + } +} + +func orEmpty(qr *VerifiedQuorumRound) []VerifiedQuorumRound { + if qr == nil { + return nil + } + return []VerifiedQuorumRound{*qr} +}