From 7a0bb9899af52f59b953f37e760eddc39cb7cddd Mon Sep 17 00:00:00 2001 From: Matthieu Vachon Date: Thu, 8 Oct 2026 18:09:26 -0400 Subject: [PATCH] Clamp LibNum to Number in Forkable.ProcessBlock to stop permanent stalls A reader bug can announce a block's LIB num higher than its own num (sol-devnet incident: block 774 with LibNum 775). BlockInCurrentChain resolves LIB by walking parent links by number only, so a too-high LibNum makes it return a ref pairing an unrelated block ID with that num; MoveLIB stores it, and every later ReversibleSegment walk then bails out before ever reaching the stored ID, silently freezing the forkable's head forever while it keeps linking incoming blocks. Clamp the block's LibNum down to its own Number on entry and log an error, mirroring utils.ClampLibNum in firehose-core (streamingfast/firehose-core#284) for symmetry across both defenses. --- CHANGELOG.md | 1 + forkable/forkable.go | 23 +++++++- forkable/forkable_test.go | 110 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 132 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d4f6d3..7fbdc74 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/forkable/forkable.go b/forkable/forkable.go index 7f6c80b..3f93405 100644 --- a/forkable/forkable.go +++ b/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) diff --git a/forkable/forkable_test.go b/forkable/forkable_test.go index 3a73a36..e67f8b1 100644 --- a/forkable/forkable_test.go +++ b/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")))