Clamp LibNum to Number to stop forkable stalling permanently

CHANGELOG.md

@@ -41,6 +41,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `FileSource` with `FileSourceErrorOnMissingMergedBlocksFile` no longer truncates its output: on a missing file it now drains every already-read block through the ordered stream before surfacing the error, instead of calling `Shutdown()` immediately (which aborted in-flight reader goroutines and discarded blocks).
 - `JoiningSource` no longer leaves a stream silent when its cursor names a block nothing can produce. A cursor block inside the live buffer's range whose ID the buffer does not know (a corrupted or forged cursor, or one from a chain the process never saw) made the hub decline the source, and the file source it fell back to then waited for the merged-blocks file holding that block number — a whole bundle, some twenty minutes on Ethereum — before failing anyway. Such a cursor now fails immediately with `ErrResolveCursor`, which the `stream` package already surfaces as an invalid argument. Cursors on a fork the live buffer no longer holds are unaffected: the forked-blocks store is consulted before giving up, and a cursor below the live range still goes to the file source.
+- `forkable.Forkable` now always clamps a block's `lib_num` down to its own block number when it is greater (invalid; equal is still valid), logs an error, instead of letting the bad value reach `BlockInCurrentChain`/`MoveLIB`, where it moved LIB past head and silently stalled the forkable forever (healthz stays green, nothing is ever emitted again) while it kept linking incoming blocks. Mirrors `utils.ClampLibNum` in firehose-core (streamingfast/firehose-core#284), which guards the same invariant at the reader.
 
 ### Changed
 

forkable/forkable.go

@@ -664,12 +664,31 @@ func (p *Forkable) ProcessBlock(blk *pbbstream.Block, obj any) error {
 		return fmt.Errorf("invalid block ID detected on block %s (previousID: %s), bad data", blk.AsRef().String(), blk.ParentId)
 	}
 
+	zlogBlk := p.logger.With(zap.Stringer("block", blk.AsRef()))
+
+	if blk.LibNum > blk.Number {
+		// A reader bug can emit a block whose announced lib_num is higher than the block's
+		// own number. BlockInCurrentChain resolves LIB purely by walking parent links until it
+		// finds that num, so a lib_num above the chain's current head returns a ref stitching
+		// together an unrelated ID with that too-high num. MoveLIB then stores that mismatched
+		// ref, and every subsequent ReversibleSegment walk bails out as soon as it sees a real
+		// block num below that fake LIB num, without ever reaching the stored ID: the forkable
+		// stalls forever while still silently linking incoming blocks. Clamping here keeps the
+		// forkDB's invariant (lib_num <= block number) intact instead of trying to special-case
+		// every LIB consumer downstream. Mirrors utils.ClampLibNum in firehose-core (defense in
+		// depth at the reader, streamingfast/firehose-core#284).
+		zlogBlk.Error("block has lib_num greater than its own block number, clamping lib_num to block number",
+			zap.Uint64("block_num", blk.Number),
+			zap.String("block_id", blk.Id),
+			zap.Uint64("original_lib_num", blk.LibNum),
+		)
+		blk.LibNum = blk.Number
+	}
+
 	if blk.Number < p.forkDB.LIBNum() && p.lastBlockSent != nil {
 		return nil
 	}
 
-	zlogBlk := p.logger.With(zap.Stringer("block", blk.AsRef()))
-
 	// TODO: consider an `initialHeadBlockID`, triggerNewLongestChain also when the initialHeadBlockID's BlockNum == blk.Num()
 	triggersNewLongestChain := p.triggersNewLongestChain(blk)
 

forkable/forkable_test.go

@@ -25,6 +25,9 @@ import (
 	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
 )
 
 // testing cursor being applied...
@@ -1585,6 +1588,113 @@ func TestForkable_ProcessBlock_UnknownLIB(t *testing.T) {
 	}
 }
 
+// TestForkable_ProcessBlock_ClampsInvalidLibNum mirrors utils.ClampLibNum's own test in
+// firehose-core (streamingfast/firehose-core#284), the reader-side guard for the same
+// invariant: LibNum equal to Number is valid, LibNum greater than Number is not and gets
+// clamped, with an error logged carrying the block num/id and the original (bad) lib_num.
+func TestForkable_ProcessBlock_ClampsInvalidLibNum(t *testing.T) {
+	testCases := []struct {
+		name        string
+		blk         *pbbstream.Block
+		expectClamp bool
+	}{
+		{
+			name:        "lib_num less than number",
+			blk:         bTestBlock("00000002a", "00000001a"),
+			expectClamp: false,
+		},
+		{
+			name:        "lib_num equal to number is valid",
+			blk:         tb("00000002a", "00000001a", 2),
+			expectClamp: false,
+		},
+		{
+			name:        "lib_num greater than number gets clamped",
+			blk:         tb("00000002a", "00000001a", 3),
+			expectClamp: true,
+		},
+	}
+
+	for _, c := range testCases {
+		t.Run(c.name, func(t *testing.T) {
+			originalLibNum := c.blk.LibNum
+
+			core, logs := observer.New(zapcore.ErrorLevel)
+
+			fap := New(newTestForkableSink(nil, nil), WithLogger(zap.New(core)))
+			fap.forkDB = fdbLinked("00000001a")
+			fap.lastLIBSeen = fap.forkDB.libRef
+
+			require.NoError(t, fap.ProcessBlock(c.blk, nil))
+
+			if !c.expectClamp {
+				assert.Equal(t, originalLibNum, c.blk.LibNum)
+				assert.Equal(t, 0, logs.Len())
+				return
+			}
+
+			assert.Equal(t, c.blk.Number, c.blk.LibNum)
+			require.Equal(t, 1, logs.Len())
+			entry := logs.All()[0]
+			assert.Equal(t, zapcore.ErrorLevel, entry.Level)
+			assert.Equal(t, c.blk.Number, entry.ContextMap()["block_num"])
+			assert.Equal(t, c.blk.Id, entry.ContextMap()["block_id"])
+			assert.Equal(t, originalLibNum, entry.ContextMap()["original_lib_num"])
+		})
+	}
+}
+
+// TestForkable_ProcessBlock_LibNumGreaterThanNumber reproduces the sol-devnet incident
+// (2026-10-08): a reader emitted a block whose LibNum is greater than its own Number
+// (774 with LibNum 775). BlockInCurrentChain walks parent links by number only, so it
+// returns a ref carrying block 774's ID but num 775; MoveLIB stores that mismatched ref
+// as the forkDB's LIB. Every later block's ReversibleSegment walk then sees
+// curNum < f.LIBNum() before ever reaching that ID and bails out with a nil segment,
+// so computeNewLongestChain never yields a chain again: the forkable stalls forever
+// while still silently linking incoming blocks.
+//
+// WithKeptFinalBlocks(10) mirrors the production purge cutoff (LIB-10) seen in the
+// incident logs: with the default of 0, the purge right after MoveLIB wipes block
+// 774's own entry, which accidentally hides the bug (see commit message / PR
+// description for the trace).
+func TestForkable_ProcessBlock_LibNumGreaterThanNumber(t *testing.T) {
+	bstream.GetProtocolFirstStreamableBlock = 1
+	sinkHandle := newTestForkableSink(nil, nil)
+
+	fap := New(sinkHandle, WithKeptFinalBlocks(10))
+	fap.forkDB = fdbLinked("00000001a")
+	fap.lastLIBSeen = fap.forkDB.libRef
+
+	processBlocks := []*pbbstream.Block{
+		bTestBlock("00000002a", "00000001a"),
+		bTestBlock("00000003a", "00000002a"),
+		tb("00000004a", "00000003a", 5), // bad reader: LibNum (5) > Number (4)
+		bTestBlock("00000005a", "00000004a"),
+		bTestBlock("00000006a", "00000005a"),
+		bTestBlock("00000007a", "00000006a"),
+	}
+
+	for _, b := range processBlocks {
+		require.NoError(t, fap.ProcessBlock(b, b.Id))
+	}
+
+	var newBlockIDs []string
+	for _, res := range sinkHandle.results {
+		if res.step == bstream.StepNew {
+			newBlockIDs = append(newBlockIDs, res.block.ID())
+		}
+	}
+
+	assert.Equal(t, []string{
+		"00000002a",
+		"00000003a",
+		"00000004a",
+		"00000005a",
+		"00000006a",
+		"00000007a",
+	}, newBlockIDs, "blocks after the bad LibNum must keep flowing instead of stalling forever")
+}
+
 func TestForkable_ForkDBContainsPreviousPreprocessedBlockObjects(t *testing.T) {
 	var nilHandler bstream.Handler
 	p := New(nilHandler, WithExclusiveLIB(bRef("00000003a")))