From c7fb8fdccfb8a576786312cf1e545c33c1e05535 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 15:41:19 +0000 Subject: [PATCH 1/2] Don't treat a block the storage never persisted as committed EpochAwareStorage/CallbackStorage.Index returned nil - success - for a block it deliberately did not persist (a Telock, which only extends the dying epoch until the sealing block finalizes). A nil error from Storage.Index is the engine's only signal that a block is durable at Storage.NumBlocks()-1, so Epoch.indexFinalization treated the skip as a commit: it set e.lastBlock to the block and advanced e.round via progressRoundsDueToCommit, while nextSeqToCommit() - which is Storage.NumBlocks() - stayed behind. The two cursors can never reconcile afterwards. The node builds proposals on a phantom parent that honest nodes reject, advertises the never-persisted block to replicating peers as its latest finalized block, and rejects the proposal that does belong at that sequence. It happens without an attacker in the dying epoch, since a Telock legitimately collects a finalization whenever the sealing finalization lags, and it can be induced in a freshly transitioned node by replaying the old-epoch Telock block and its finalization QC in a ReplicationResponse. - Storage.Index now documents its contract and CallbackStorage reports the skip with the new common.ErrBlockNotIndexed instead of nil. The block is deliberately left in the CachedStorage cache, as a Telock must stay retrievable by digest while its epoch is being extended. - indexFinalization reports whether the block was committed and leaves lastBlock and the round untouched when it was not, both for a reported skip and - defensively - for a storage that reports success without advancing NumBlocks. Its callers stop the commit loop and, on the replication path, restore the rounds map so the block that does belong at that round can still be finalized. - The non-validator likewise no longer halts, nor treats the sequence as accepted, when the storage declines to persist a block. Old-epoch artifacts also reached the commit path with no epoch binding: VerifyQC only attests that the signers form a quorum of the current validator set, which a quorum of the previous epoch still satisfies when the sets overlap. A replayed Telock chains onto the last block of its epoch, so it claims the round and sequence the first block of the new epoch belongs to. Blocks and finalizations of an earlier epoch are now rejected in verifyQuorumRound and handleFinalizationMessage, and a sealed epoch stops accepting finalization messages, since nothing beyond its sealing block belongs to it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GpZGbLPZnb9sV7EVcYi3dB --- adapters.go | 9 +- adapters_test.go | 41 +++++ common/api.go | 10 ++ nonvalidator/non_validator.go | 8 + simplex/epoch.go | 102 +++++++++++- simplex/epoch_not_indexed_test.go | 258 ++++++++++++++++++++++++++++++ 6 files changed, 422 insertions(+), 6 deletions(-) create mode 100644 simplex/epoch_not_indexed_test.go 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..76a68510 100644 --- a/common/api.go +++ b/common/api.go @@ -50,11 +50,21 @@ 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. + // 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..9633a3de 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,13 @@ 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)) + 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/simplex/epoch.go b/simplex/epoch.go index 7e3c95a6..085028cb 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -804,6 +804,25 @@ 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. + // Without this check, a stale epoch transition artifact - a block of a sealed epoch that + // was never persisted - could be replayed to us and be mistaken for a finalized block. + 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 +1477,18 @@ 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. Stop, and leave the round + // in place so that the block that does belong at this sequence can still be committed. + 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 +1504,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 +1564,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 +2298,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 +2309,28 @@ 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. + if hadRoundEntry { + e.rounds[md.Round] = previousRoundEntry + } else { + delete(e.rounds, md.Round) + } + 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 +3548,24 @@ 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. + 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..d3f53437 --- /dev/null +++ b/simplex/epoch_not_indexed_test.go @@ -0,0 +1,258 @@ +// 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. + require.NoError(t, e.HandleMessage(&Message{ + ReplicationRequest: &ReplicationRequest{LatestFinalizedSeq: 1}, + }, nodes[1])) + latestFinalized := awaitLatestFinalizedSeq(t, comm) + require.Equal(t, declinedSeq-1, latestFinalized.Finalization.Finalization.Seq, + "the epoch must advertise the last committed block as its latest finalized sequence") + require.Equal(t, blocks[declinedSeq-1].BlockHeader().Digest, latestFinalized.VerifiedBlock.BlockHeader().Digest) + }) + } +} + +// 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 +} + +// awaitLatestFinalizedSeq returns the LatestFinalizedSeq of the first replication response +// the epoch sends. +func awaitLatestFinalizedSeq(t *testing.T, comm *recordingComm) *VerifiedQuorumRound { + t.Helper() + + timeout := time.After(30 * time.Second) + for { + select { + case msg := <-comm.SentMessages: + if msg.VerifiedReplicationResponse != nil && msg.VerifiedReplicationResponse.LatestFinalizedSeq != nil { + return msg.VerifiedReplicationResponse.LatestFinalizedSeq + } + case <-timeout: + require.FailNow(t, "timed out waiting for a replication response with a latest finalized sequence") + return nil + } + } +} From 8e7e6fa07cf9aa0d260127f7a8d0ac66d52e9d6d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 16:07:40 +0000 Subject: [PATCH 2/2] Discard the block the storage refused instead of holding on to it Refusing to commit a block the storage did not persist is not enough on its own: whatever the engine still holds for that sequence has to go too, or it takes the place of the block that does belong there. - The non-validator left the refused block in incompleteSequences, which only a real commit ever clears. handleBlock then dropped the block that belongs at that sequence as a duplicate, and handleFinalization treated its finalization as conflicting and set haltedError - a permanent halt where before the fix the state was cleared as a side effect of the false commit. Drop the entry and re-request the sequence. - indexFinalizations kept the round, which let the refused block be served to replicating peers as a finalized quorum round for its sequence, be used as the parent of our next proposal, and make storeProposal refuse the real block for that round. Drop the round, as the replication path already did. - The replication path now re-requests the sequence as well, since the caller removed it from the replication state before handing it over. Also note in the Storage contract that NumBlocks() must reflect Index synchronously, and stop the epoch-binding comments from reading as though they were the boundary rather than a layer above it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GpZGbLPZnb9sV7EVcYi3dB --- common/api.go | 4 +- nonvalidator/non_validator.go | 6 ++ nonvalidator/not_indexed_test.go | 100 ++++++++++++++++++++++++++++++ simplex/epoch.go | 27 ++++++-- simplex/epoch_not_indexed_test.go | 39 ++++++++---- 5 files changed, 158 insertions(+), 18 deletions(-) create mode 100644 nonvalidator/not_indexed_test.go diff --git a/common/api.go b/common/api.go index 76a68510..40d36e2a 100644 --- a/common/api.go +++ b/common/api.go @@ -62,7 +62,9 @@ type Storage interface { // 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. + // 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 9633a3de..01b6900b 100644 --- a/nonvalidator/non_validator.go +++ b/nonvalidator/non_validator.go @@ -271,6 +271,12 @@ func (n *NonValidator) newFinalizedBlockTask(block common.Block, finalization *c // 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 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 085028cb..bee25ec2 100644 --- a/simplex/epoch.go +++ b/simplex/epoch.go @@ -814,9 +814,10 @@ func (e *Epoch) handleFinalizationMessage(message *common.Finalization, from com } // 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. - // Without this check, a stale epoch transition artifact - a block of a sealed epoch that - // was never persisted - could be replayed to us and be mistaken for a finalized block. + // 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)) @@ -1483,8 +1484,14 @@ func (e *Epoch) indexFinalizations(startRound uint64) error { } 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. Stop, and leave the round - // in place so that the block that does belong at this sequence can still be committed. + // 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 @@ -2320,11 +2327,16 @@ func (e *Epoch) createFinalizedBlockVerificationTask(block common.Block, finaliz // 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), @@ -3554,7 +3566,10 @@ func (e *Epoch) verifyQuorumRound(q common.QuorumRound) error { // 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. + // 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) diff --git a/simplex/epoch_not_indexed_test.go b/simplex/epoch_not_indexed_test.go index d3f53437..0994f2c6 100644 --- a/simplex/epoch_not_indexed_test.go +++ b/simplex/epoch_not_indexed_test.go @@ -137,14 +137,25 @@ func TestEpochDoesNotCommitBlockTheStorageDeclinedToIndex(t *testing.T) { 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. + // 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}, + ReplicationRequest: &ReplicationRequest{LatestFinalizedSeq: 1, Seqs: []uint64{declinedSeq}}, }, nodes[1])) - latestFinalized := awaitLatestFinalizedSeq(t, comm) - require.Equal(t, declinedSeq-1, latestFinalized.Finalization.Finalization.Seq, + 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, latestFinalized.VerifiedBlock.BlockHeader().Digest) + 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") + } }) } } @@ -238,21 +249,27 @@ func mustVote(t *testing.T, block testutil.AnyBlock, from NodeID) Vote { return *vote } -// awaitLatestFinalizedSeq returns the LatestFinalizedSeq of the first replication response -// the epoch sends. -func awaitLatestFinalizedSeq(t *testing.T, comm *recordingComm) *VerifiedQuorumRound { +// 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 && msg.VerifiedReplicationResponse.LatestFinalizedSeq != nil { - return msg.VerifiedReplicationResponse.LatestFinalizedSeq + if msg.VerifiedReplicationResponse != nil { + return msg.VerifiedReplicationResponse } case <-timeout: - require.FailNow(t, "timed out waiting for a replication response with a latest finalized sequence") + 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} +}