Don't treat a block the storage never persisted as committed

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 {

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.

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
 }
 

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

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")
+}

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)