From 524923258adb8da4e8f352c9887199654c7aea60 Mon Sep 17 00:00:00 2001 From: Damilola Edwards Date: Wed, 29 Jul 2026 16:52:02 +0100 Subject: [PATCH] Stop dropped block headers from being reported as reorgs When a block's header didn't arrive within the 500ms wait, the loop skipped updating lastBlock along with the reorg check, leaving it pointing at an older block. The next block's parent root was then compared against that stale reference instead of the block that was actually skipped, so a completely healthy linear chain could be reported as a reorg purely because one header fetch was slow. lastBlock now always advances to the current block. Only the reorg comparison itself is skipped when a header is missing, since there is no way to check a block's parent without its header, and a warning is logged so a run of missed headers is still visible. --- pkg/tasks/check_consensus_reorgs/task.go | 13 +- pkg/tasks/check_consensus_reorgs/task_test.go | 187 ++++++++++++++++++ 2 files changed, 196 insertions(+), 4 deletions(-) create mode 100644 pkg/tasks/check_consensus_reorgs/task_test.go diff --git a/pkg/tasks/check_consensus_reorgs/task.go b/pkg/tasks/check_consensus_reorgs/task.go index 6a17899b..c73c827a 100644 --- a/pkg/tasks/check_consensus_reorgs/task.go +++ b/pkg/tasks/check_consensus_reorgs/task.go @@ -96,11 +96,16 @@ func (t *Task) Execute(ctx context.Context) error { select { case block := <-blockSubscription.Channel(): blockHeader := block.AwaitHeader(ctx, 500*time.Millisecond) - if blockHeader == nil { - break - } - if lastBlock != nil && !bytes.Equal(blockHeader.Message.ParentRoot[:], lastBlock.Root[:]) { + switch { + case blockHeader == nil: + // The header didn't arrive in time, so we can't tell what + // this block's parent is and can't check it for a reorg. + // lastBlock still advances to this block below: comparing + // the *next* block's parent against a stale, older head + // would otherwise report a reorg that never happened. + t.logger.Warnf("timed out waiting for header of block %v [0x%x], skipping reorg check for it", block.Slot, block.Root) + case lastBlock != nil && !bytes.Equal(blockHeader.Message.ParentRoot[:], lastBlock.Root[:]): // chain reorg t.processChainReorg(lastBlock, block) diff --git a/pkg/tasks/check_consensus_reorgs/task_test.go b/pkg/tasks/check_consensus_reorgs/task_test.go new file mode 100644 index 00000000..78dd1360 --- /dev/null +++ b/pkg/tasks/check_consensus_reorgs/task_test.go @@ -0,0 +1,187 @@ +package checkconsensusreorgs + +import ( + "bytes" + "context" + "io" + "testing" + "time" + + "github.com/ethpandaops/assertoor/pkg/clients" + "github.com/ethpandaops/assertoor/pkg/clients/consensus" + "github.com/ethpandaops/assertoor/pkg/db" + "github.com/ethpandaops/assertoor/pkg/events" + "github.com/ethpandaops/assertoor/pkg/helper" + "github.com/ethpandaops/assertoor/pkg/logger" + "github.com/ethpandaops/assertoor/pkg/names" + "github.com/ethpandaops/assertoor/pkg/txmgr" + "github.com/ethpandaops/assertoor/pkg/types" + "github.com/ethpandaops/go-eth2-client/spec/phase0" + "github.com/sirupsen/logrus" +) + +type fakeServices struct { + pool *clients.ClientPool +} + +func (f *fakeServices) Database() *db.Database { return nil } +func (f *fakeServices) ClientPool() *clients.ClientPool { return f.pool } +func (f *fakeServices) WalletManager() *txmgr.Spamoor { return nil } +func (f *fakeServices) ValidatorNames() *names.ValidatorNames { return nil } +func (f *fakeServices) EventBus() *events.EventBus { return nil } + +type fakeSchedulerRunner struct { + services types.TaskServices +} + +func (f *fakeSchedulerRunner) GetServices() types.TaskServices { return f.services } +func (f *fakeSchedulerRunner) GetTestRunID() uint64 { return 1 } +func (f *fakeSchedulerRunner) GetTestRunCtx() context.Context { return context.Background() } +func (f *fakeSchedulerRunner) ParseTaskOptions(_ helper.IRawMessage) (*types.TaskOptions, error) { + return nil, nil +} +func (f *fakeSchedulerRunner) ExecuteTask(_ context.Context, _ types.TaskIndex, _ func(ctx context.Context, cancelFn context.CancelFunc, taskIndex types.TaskIndex)) error { + return nil +} +func (f *fakeSchedulerRunner) TestResultPath() (string, error) { return "", nil } +func (f *fakeSchedulerRunner) GetTaskState(_ types.TaskIndex) types.TaskState { return nil } +func (f *fakeSchedulerRunner) GetTaskCount() uint64 { return 0 } +func (f *fakeSchedulerRunner) GetAllTasks() []types.TaskIndex { return nil } +func (f *fakeSchedulerRunner) GetRootTasks() []types.TaskIndex { return nil } +func (f *fakeSchedulerRunner) GetAllCleanupTasks() []types.TaskIndex { return nil } +func (f *fakeSchedulerRunner) GetRootCleanupTasks() []types.TaskIndex { return nil } + +func header(slot phase0.Slot, parent phase0.Root) *phase0.SignedBeaconBlockHeader { + return &phase0.SignedBeaconBlockHeader{ + Message: &phase0.BeaconBlockHeader{ + Slot: slot, + ParentRoot: parent, + }, + } +} + +// ensureHeader sets a block's header the same way the real client +// ingestion path does, so AwaitHeader resolves immediately instead of +// waiting out its timeout. +func ensureHeader(t *testing.T, block *consensus.Block, h *phase0.SignedBeaconBlockHeader) { + t.Helper() + + if err := block.EnsureHeader(func() (*phase0.SignedBeaconBlockHeader, error) { + return h, nil + }); err != nil { + t.Fatalf("failed setting header: %v", err) + } +} + +func newTestTask(t *testing.T) (*Task, *consensus.BlockCache) { + t.Helper() + + log := logrus.New() + log.SetOutput(io.Discard) + + pool, err := clients.NewClientPool(log) + if err != nil { + t.Fatalf("failed constructing client pool: %v", err) + } + + taskCtx := &types.TaskContext{ + Scheduler: &fakeSchedulerRunner{services: &fakeServices{pool: pool}}, + Logger: logger.NewLogger(&logger.ScopeOptions{ + Parent: log, + }), + SetResult: func(types.TaskResult) {}, + ReportProgress: func(float64, string) {}, + } + + task := &Task{ + ctx: taskCtx, + logger: taskCtx.Logger.GetLogger(), + config: Config{}, + } + + return task, pool.GetConsensusPool().GetBlockCache() +} + +// handleBlock is Execute()'s per-block handling (the arm of its select +// statement), copied verbatim minus the channel receive itself, which +// can't be driven from this package: it is fired by an unexported method +// deep in the consensus client's ingestion pipeline, reachable in +// production only through a live beacon client goroutine. +func handleBlock(t *testing.T, task *Task, lastBlock **consensus.Block, block *consensus.Block, checkCount *int) { + t.Helper() + + blockHeader := block.AwaitHeader(context.Background(), 500*time.Millisecond) + + switch { + case blockHeader == nil: + case *lastBlock != nil && !bytes.Equal(blockHeader.Message.ParentRoot[:], (*lastBlock).Root[:]): + task.processChainReorg(*lastBlock, block) + } + + *checkCount++ + *lastBlock = block +} + +// TestNoPhantomReorgOnDroppedHeader is the regression test for the phantom +// reorg bug: a linear chain A -> B -> C, where B's header is never +// delivered in time, must not be counted as a reorg. +func TestNoPhantomReorgOnDroppedHeader(t *testing.T) { + task, blockCache := newTestTask(t) + + var rootA, rootB, rootC phase0.Root + rootA[0] = 0xAA + rootB[0] = 0xBB + rootC[0] = 0xCC + + blockA, _ := blockCache.AddBlock(rootA, 100) + ensureHeader(t, blockA, header(100, phase0.Root{})) + + blockB, _ := blockCache.AddBlock(rootB, 101) + // blockB's header is deliberately never set. + + blockC, _ := blockCache.AddBlock(rootC, 102) + ensureHeader(t, blockC, header(102, rootB)) // C's real parent is B: a linear chain. + + var lastBlock *consensus.Block + + checkCount := 0 + for _, block := range []*consensus.Block{blockA, blockB, blockC} { + handleBlock(t, task, &lastBlock, block, &checkCount) + } + + if task.totalReorgs != 0 { + t.Fatalf("expected no reorgs on a linear chain with one dropped header, got %d", task.totalReorgs) + } +} + +// TestRealReorgStillDetected guards against the fix over-correcting: a +// genuine reorg, where every header is available, must still be counted. +func TestRealReorgStillDetected(t *testing.T) { + task, blockCache := newTestTask(t) + + var rootA, rootB, rootC phase0.Root + rootA[0] = 0xAA + rootB[0] = 0xBB + rootC[0] = 0xCC + + blockA, _ := blockCache.AddBlock(rootA, 100) + ensureHeader(t, blockA, header(100, phase0.Root{})) + + blockB, _ := blockCache.AddBlock(rootB, 101) + ensureHeader(t, blockB, header(101, rootA)) + + // blockC forks off blockA directly, not blockB: a genuine reorg. + blockC, _ := blockCache.AddBlock(rootC, 102) + ensureHeader(t, blockC, header(102, rootA)) + + var lastBlock *consensus.Block + + checkCount := 0 + for _, block := range []*consensus.Block{blockA, blockB, blockC} { + handleBlock(t, task, &lastBlock, block, &checkCount) + } + + if task.totalReorgs != 1 { + t.Fatalf("expected exactly 1 reorg for a genuine fork, got %d", task.totalReorgs) + } +}