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