From 8c3e1a5717485e7f219cc66e9687fbab0f3ee921 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Mon, 7 Sep 2026 06:07:51 +0800 Subject: [PATCH 1/5] refactor(core): use errors.Is for sentinel error comparisons in blockchain.go Replace direct == / != comparisons against sentinel errors (consensus.ErrPrunedAncestor, consensus.ErrFutureBlock, consensus.ErrUnknownAncestor, ErrKnownBlock, ErrStopPreparingBlock) with errors.Is, which also matches wrapped errors. --- core/blockchain.go | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 54f5291c8895..0e6b50fbe7ab 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1806,12 +1806,12 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] block, err := it.next() switch { // First block is pruned, insert as sidechain and reorg only if TD grows enough - case err == consensus.ErrPrunedAncestor: + case errors.Is(err, consensus.ErrPrunedAncestor): return bc.insertSidechain(block, it) // First block is future, shove it (and all children) to the future queue (unknown ancestor) - case err == consensus.ErrFutureBlock || (err == consensus.ErrUnknownAncestor && bc.futureBlocks.Contains(it.first().ParentHash())): - for block != nil && (it.index == 0 || err == consensus.ErrUnknownAncestor) { + case errors.Is(err, consensus.ErrFutureBlock) || (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(it.first().ParentHash())): + for block != nil && (it.index == 0 || errors.Is(err, consensus.ErrUnknownAncestor)) { if err := bc.addFutureBlock(block); err != nil { return it.index, events, coalescedLogs, err } @@ -1827,11 +1827,11 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] // 1. We did a roll-back, and should now do a re-import // 2. The block is stored as a sidechain, and is lying about it's stateroot, and passes a stateroot // from the canonical chain, which has not been verified. - case err == ErrKnownBlock: + case errors.Is(err, ErrKnownBlock): // Skip all known blocks that behind us current := bc.CurrentBlock().Number.Uint64() - for block != nil && err == ErrKnownBlock && current >= block.NumberU64() { + for block != nil && errors.Is(err, ErrKnownBlock) && current >= block.NumberU64() { stats.ignored++ block, err = it.next() } @@ -1934,13 +1934,13 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] } // Any blocks remaining here? The only ones we care about are the future ones - if block != nil && err == consensus.ErrFutureBlock { + if block != nil && errors.Is(err, consensus.ErrFutureBlock) { if err := bc.addFutureBlock(block); err != nil { return it.index, events, coalescedLogs, err } block, err = it.next() - for ; block != nil && err == consensus.ErrUnknownAncestor; block, err = it.next() { + for ; block != nil && errors.Is(err, consensus.ErrUnknownAncestor); block, err = it.next() { if err := bc.addFutureBlock(block); err != nil { return it.index, events, coalescedLogs, err } @@ -2064,7 +2064,7 @@ func (bc *BlockChain) insertSidechain(block *types.Block, it *insertIterator) (i // ones. Any other errors means that the block is invalid, and should not be written // to disk. err := consensus.ErrPrunedAncestor - for ; block != nil && (err == consensus.ErrPrunedAncestor); block, err = it.next() { + for ; block != nil && (errors.Is(err, consensus.ErrPrunedAncestor)); block, err = it.next() { // Check the canonical state root for that number if number := block.NumberU64(); current >= number { if canonical := bc.GetBlockByNumber(number); canonical != nil && canonical.Root() == block.Root() { @@ -2225,13 +2225,13 @@ func (bc *BlockChain) getResultBlock(block *types.Block, verifiedM2 bool) (*Resu bstart := time.Now() err := bc.validator.ValidateBody(block) switch { - case err == ErrKnownBlock: + case errors.Is(err, ErrKnownBlock): // Block and state both already known. However if the current block is below // this number we did a rollback and we should reimport it nonetheless. if bc.CurrentBlock().Number.Uint64() >= block.NumberU64() { return nil, ErrKnownBlock } - case err == consensus.ErrPrunedAncestor: + case errors.Is(err, consensus.ErrPrunedAncestor): // Block competing with the canonical chain, store in the db, but don't process // until the competitor TD goes above the canonical TD currentBlock := bc.CurrentBlock() @@ -2286,7 +2286,7 @@ func (bc *BlockChain) getResultBlock(block *types.Block, verifiedM2 bool) (*Resu receipts, logs, usedGas, err := bc.processor.ProcessBlockNoValidator(calculatedBlock, statedb, tradingState, bc.vmConfig, feeCapacity) process := time.Since(bstart) if err != nil { - if err != ErrStopPreparingBlock { + if !errors.Is(err, ErrStopPreparingBlock) { bc.reportBlock(block, receipts, err) } return nil, err From 33e8da5d27878619d07d0608b7f4ca23c073a4d0 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Sun, 6 Sep 2026 01:25:37 +0800 Subject: [PATCH 2/5] refactor(core): drop dead stats accounting from insertChain abort paths stats is a function-local insertStats and stats.report is only called from the main import loop, so every stats write on a path that returns before reaching the loop is dead and never logged: - stats.queued += it.processed() on the first-block future path is not only unreachable for reporting but off by one: it.processed() returns it.index+1, which over-counts by one both when the whole batch is drained (it.index == len(chain) at exhaustion) and when the loop aborts on a non-exempt error (it.index points at the block that was never queued). Both queued counters date back to ff435e081. - stats.queued++ in the tail future loop (which also skipped the block that triggered the queueing) and stats.ignored += it.remaining() on both future paths are written then dropped on return. - stats.ignored += len(it.chain) in the first-block error abort has the same write-then-return shape. Remove all of them. Drop the insertStats.queued field and the "queued" log context in report() that could never fire once the counters are gone, matching upstream geth whose insertStats no longer carries a queued field, and remove the now-unused insertIterator helpers processed() and remaining(). The live stats.ignored accounting (the ErrKnownBlock skip loop that falls through to the import loop and the InsertReceiptChain stats) is untouched. --- core/blockchain.go | 7 ------- core/blockchain_insert.go | 21 ++++----------------- 2 files changed, 4 insertions(+), 24 deletions(-) diff --git a/core/blockchain.go b/core/blockchain.go index 0e6b50fbe7ab..c45fbc015347 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1817,10 +1817,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] } block, err = it.next() } - stats.queued += it.processed() - stats.ignored += it.remaining() - - // If there are any still remaining, mark as ignored return it.index, events, coalescedLogs, err // First block (and state) is known @@ -1839,7 +1835,6 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] // Some other error occurred, abort case err != nil: - stats.ignored += len(it.chain) bc.reportBlock(block, nil, err) return it.index, events, coalescedLogs, err } @@ -1944,10 +1939,8 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] if err := bc.addFutureBlock(block); err != nil { return it.index, events, coalescedLogs, err } - stats.queued++ } } - stats.ignored += it.remaining() // Append a single chain head event if we've progressed the chain if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() { diff --git a/core/blockchain_insert.go b/core/blockchain_insert.go index b376b9836364..d5ccd15d64ca 100644 --- a/core/blockchain_insert.go +++ b/core/blockchain_insert.go @@ -27,10 +27,10 @@ import ( // insertStats tracks and reports on block insertion. type insertStats struct { - queued, processed, ignored int - usedGas uint64 - lastIndex int - startTime mclock.AbsTime + processed, ignored int + usedGas uint64 + lastIndex int + startTime mclock.AbsTime } // statsReportLimit is the time limit during import and export after which we @@ -65,9 +65,6 @@ func (st *insertStats) report(chain []*types.Block, index int, cache common.Stor } context = append(context, []interface{}{"dirty", cache}...) - if st.queued > 0 { - context = append(context, []interface{}{"queued", st.queued}...) - } if st.ignored > 0 { context = append(context, []interface{}{"ignored", st.ignored}...) } @@ -154,13 +151,3 @@ func (it *insertIterator) previous() *types.Header { func (it *insertIterator) first() *types.Block { return it.chain[0] } - -// remaining returns the number of remaining blocks. -func (it *insertIterator) remaining() int { - return len(it.chain) - it.index -} - -// processed returns the number of processed blocks. -func (it *insertIterator) processed() int { - return it.index + 1 -} From 5d666921f908c196a012129e7e4a621d25454985 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Mon, 7 Sep 2026 07:36:19 +0800 Subject: [PATCH 3/5] test(consensus): pin the timestamp check before the parent lookup Add regression tests asserting that header verification checks the timestamp first and answers ErrFutureBlock for blocks too far ahead, without requiring their parent to be resolvable. XDPoS v1/v2 (under full verification) order the checks this way; these tests pin the behavior the future-batch tail queueing (queueFutureTail) relies on: children of a future block surface as ErrFutureBlock, not ErrUnknownAncestor. No production code is changed. --- .../engines/engine_v1/verify_header_test.go | 117 ++++++++++++++++++ .../engine_v2_tests/verify_header_test.go | 78 ++++++++++++ 2 files changed, 195 insertions(+) create mode 100644 consensus/XDPoS/engines/engine_v1/verify_header_test.go diff --git a/consensus/XDPoS/engines/engine_v1/verify_header_test.go b/consensus/XDPoS/engines/engine_v1/verify_header_test.go new file mode 100644 index 000000000000..01af1dc9c68a --- /dev/null +++ b/consensus/XDPoS/engines/engine_v1/verify_header_test.go @@ -0,0 +1,117 @@ +package engine_v1 + +import ( + "math/big" + "testing" + "time" + + "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/consensus" + "github.com/XinFinOrg/XDPoSChain/consensus/XDPoS/utils" + "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/params" + "github.com/stretchr/testify/assert" +) + +// futureTestChainReader mimics a chain that holds no verifiable ancestors: it +// hands out no headers, so any parent lookup has to fall back to the batch of +// headers passed to the engine. +type futureTestChainReader struct { + consensus.ChainReader // nil; methods other than the overrides below are only reachable under a regressed engine ordering + config *params.ChainConfig +} + +func (r *futureTestChainReader) Config() *params.ChainConfig { + return r.config +} + +func (r *futureTestChainReader) GetHeader(common.Hash, uint64) *types.Header { + return nil +} + +func (r *futureTestChainReader) GetHeaderByNumber(uint64) *types.Header { + return nil +} + +func (r *futureTestChainReader) GetHeaderByHash(common.Hash) *types.Header { + return nil +} + +func (r *futureTestChainReader) GetBlock(common.Hash, uint64) *types.Block { + return nil +} + +func (r *futureTestChainReader) GetBlockByNumber(uint64) *types.Block { + return nil +} + +func (r *futureTestChainReader) CurrentHeader() *types.Header { + return nil +} + +// futureTestHeader builds a header that passes all standalone v1 checks so that +// only the timestamp/parent ordering decides its fate. +func futureTestHeader(config *params.ChainConfig, number int64, parentHash common.Hash, timestamp uint64) *types.Header { + header := &types.Header{ + Number: big.NewInt(number), + ParentHash: parentHash, + Difficulty: big.NewInt(1), + GasLimit: 1200000000, + Time: timestamp, + Extra: make([]byte, utils.ExtraVanity+utils.ExtraSeal), + UncleHash: utils.UncleHash, + } + if config.IsEIP1559(header.Number) { + header.BaseFee = params.BaseFeeForBlock(config, header.Number) + } + return header +} + +// TestFutureTimestampCheckPrecedesParentLookup pins the engine premise that the +// insertChain future-batch handling relies on: the timestamp check runs before +// the parent lookup, so a header whose parent is in the same batch and whose +// timestamp is in the future surfaces as ErrFutureBlock, never as +// ErrUnknownAncestor. If the checks are ever reordered, children of a future +// block stop being classified as future blocks and insertChain treats a valid +// delivery as an invalid chain, which makes the downloader drop the peer. +func TestFutureTimestampCheckPrecedesParentLookup(t *testing.T) { + config := *params.TestXDPoSMockChainConfig + xdpos := *config.XDPoS + xdpos.SkipV1Validation = false // the timestamp check is part of full v1 validation + config.XDPoS = &xdpos + + engine := New(&config, nil) + reader := &futureTestChainReader{config: &config} + + now := uint64(time.Now().Unix()) + block1 := futureTestHeader(&config, 1, common.Hash{}, now) + // Timestamp in the future and parent not resolvable from the reader: with + // the checks in the wrong order this header would answer ErrUnknownAncestor. + block2 := futureTestHeader(&config, 2, block1.Hash(), now+10000) + + // Single-header path: the parent (block1) is neither in a batch nor in the + // database, so the parent lookup fails unless the timestamp check wins. + err := engine.VerifyHeader(reader, block2, true) + assert.Equal(t, consensus.ErrFutureBlock, err) + + // Batch path: the parent (block1) is in the same batch, so the parent + // lookup can always succeed and must not mask the future classification. + headers := []*types.Header{block1, block2} + abort := make(chan struct{}) + results := make(chan error, len(headers)) + engine.VerifyHeaders(reader, headers, []bool{true, true}, abort, results) + for i := 0; i < len(headers); i++ { + select { + case result := <-results: + if i == 0 { + // block1 has no verifiable ancestor on the stub reader and no + // seal to recover; only the future classification of block2 is + // under test. + continue + } + assert.Equal(t, consensus.ErrFutureBlock, result) + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for verify result %d", i) + } + } +} diff --git a/consensus/tests/engine_v2_tests/verify_header_test.go b/consensus/tests/engine_v2_tests/verify_header_test.go index 9db05ee6fae2..b3e5b6b582d9 100644 --- a/consensus/tests/engine_v2_tests/verify_header_test.go +++ b/consensus/tests/engine_v2_tests/verify_header_test.go @@ -687,3 +687,81 @@ func TestVerifyHeadersDoesNotFabricateBatchBlocksForHookPenalty(t *testing.T) { } } } + +// TestFutureTimestampCheckPrecedesParentLookup pins the engine premise that the +// insertChain future-batch handling relies on: the timestamp check runs before +// the parent lookup, so a header whose parent is in the same batch and whose +// timestamp is in the future surfaces as ErrFutureBlock, never as +// ErrUnknownAncestor. If the checks are ever reordered, children of a future +// block stop being classified as future blocks and insertChain treats a valid +// delivery as an invalid chain, which makes the downloader drop the peer. +func TestFutureTimestampCheckPrecedesParentLookup(t *testing.T) { + skipLongInShortMode(t) + b, err := json.Marshal(params.TestXDPoSMockChainConfig) + assert.Nil(t, err) + configString := string(b) + + var config params.ChainConfig + err = json.Unmarshal([]byte(configString), &config) + assert.Nil(t, err) + // Block 901 is the first v2 block with round of 1 + blockchain, _, block910, signer, signFn, _ := PrepareXDCTestBlockChainForV2Engine(t, 910, &config, nil) + adaptor := blockchain.Engine().(*XDPoS.XDPoS) + + // Build blocks 911 and 912 in memory only; neither is written into the DB. + block911 := CreateBlock( + blockchain, + blockchain.Config(), + block910, + 911, + int64(911)-config.XDPoS.V2.SwitchBlock.Int64(), + signer.Hex(), + signer, + signFn, + nil, + nil, + "", + ) + block912 := CreateBlock( + blockchain, + blockchain.Config(), + block911, + 912, + int64(912)-config.XDPoS.V2.SwitchBlock.Int64(), + signer.Hex(), + signer, + signFn, + nil, + nil, + "", + ) + + // Re-timestamp 912 into the future. The hash changes so the QC no longer + // matches, which is fine: the timestamp check also precedes QC verification. + futureHeader := block912.Header() + futureHeader.Time = uint64(time.Now().Unix() + 10000) + + // Batch path: the parent (911) is in the same batch, so the parent lookup + // can always succeed and must not mask the future classification. + headers := []*types.Header{block911.Header(), futureHeader} + fullVerifies := []bool{true, true} + _, results := adaptor.VerifyHeaders(blockchain, headers, fullVerifies) + for i := 0; i < len(headers); i++ { + select { + case result := <-results: + if i == 0 { + assert.Nil(t, result) + continue + } + assert.Equal(t, consensus.ErrFutureBlock, result) + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for verify result %d", i) + } + } + + // Single-header path: the parent (911) is neither in a batch nor in the DB, + // so an engine that checks the parent before the timestamp would answer + // ErrUnknownAncestor here instead of ErrFutureBlock. + err = adaptor.VerifyHeader(blockchain, futureHeader, true) + assert.Equal(t, consensus.ErrFutureBlock, err) +} From 8dcc7a2abd320280e34add77b1b2a209a85bd965 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Mon, 7 Sep 2026 16:00:00 +0800 Subject: [PATCH 4/5] fix(eth/downloader): only handle the proposed block when the tail is canonical Problem importBlockResults fed the last fetched block to the proposed-block handler as long as the batch import returned nil, taking a nil error to mean the last block reached the canonical chain. A nil InsertChain does not imply canonicality: a fork batch writes its blocks as side-chain entries that a mere existence check cannot tell from canonicality, and processQC overwrites highestQuorumCert before its own existence check (GetHeaderByHash), so a master node can end up voting for a block that is not in the canonical chain. Changes - Gate the proposed-block handler on storage and canonicality: the batch tail must be present as a full block and be the canonical block at its own height, via a new GetCanonicalHash method on the downloader's BlockChain interface. The canonical hash is keyed by height, so it is immune to a concurrent import advancing the head past the tail between the insertion and this read (a CurrentBlock comparison would spuriously skip the handler in that window); only a reorg changes it, and a reorged-away block must not reach the handler. HasBlock covers the fast sync header phase, where the header chain marks a height canonical before its body is imported. - downloadTester gains storeBlock/GetCanonicalHash, a canonical number-to-hash table advanced the way HeaderChain.WriteHeader and insertChain advance theirs, plus parkTailOnce and extendTailAfterInsert hooks so the tests can present batches that never reached the chain or were later reorganized away. Both hooks are consumed by the single InsertChain call they apply to. GetCanonicalHash picks the head by total difficulty, the way the real chain resolves a reorg, instead of by insertion order. HasBlock now also requires the stored block to match the requested height, like the real chain. Tests - TestImportBlockResultsProposedBlockHandler covers six scenarios: a batch that never reached the chain must not trigger the handler, a fully imported batch triggers it exactly once on the batch tail, a stored fork batch re-delivered after the local chain grew past it must not trigger it again, a concurrent import advancing the head past the canonical tail must not skip the handler (height-keyed lookup), a heavier fork stays canonical once a lighter chain is stored after it, and a height that the fast-sync header phase made canonical without a body must not trigger it either. --- eth/downloader/downloader.go | 20 +- eth/downloader/downloader_test.go | 392 ++++++++++++++++++++++++++++-- 2 files changed, 382 insertions(+), 30 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 05cd1bd8fe78..0bf39643550f 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -212,6 +212,9 @@ type BlockChain interface { // InsertChain inserts a batch of blocks into the local chain. InsertChain(types.Blocks) (int, error) + // GetCanonicalHash retrieves the hash of the canonical block at the given height. + GetCanonicalHash(uint64) common.Hash + // InterruptInsert disables or enables chain insertion. InterruptInsert(on bool) @@ -1625,11 +1628,20 @@ func (d *Downloader) importBlockResults(results []*fetchResult) error { } return fmt.Errorf("%w: %v", errInvalidChain, err) } + // A nil InsertChain does not mean the tail reached the canonical chain: + // a fork batch is stored as side entries and a parked tail is not stored + // at all. Comparing by height keeps a concurrent import that advances + // the head past the tail from skipping the handler. + tail := blocks[len(blocks)-1] if d.handleProposedBlock != nil { - header := blocks[len(blocks)-1].Header() - err := d.handleProposedBlock(header) - if err != nil { - log.Info("[downloader] handle proposed block has error", "err", err, "block hash", header.Hash(), "number", header.Number) + hasBlock := d.blockchain.HasBlock(tail.Hash(), tail.NumberU64()) + canonical := d.blockchain.GetCanonicalHash(tail.NumberU64()) == tail.Hash() + if hasBlock && canonical { + if err := d.handleProposedBlock(tail.Header()); err != nil { + log.Info("[downloader] handle proposed block has error", "err", err, "block hash", tail.Hash(), "number", tail.Number()) + } + } else { + log.Debug("[downloader] skipped proposed block handler", "block hash", tail.Hash(), "number", tail.Number(), "hasBlock", hasBlock, "canonical", canonical) } } return nil diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index f7195caf0dbb..b2f7be6562bb 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -67,13 +67,28 @@ type downloadTester struct { peerDb ethdb.Database // Database of the peers containing all data peers map[string]*downloadTesterPeer - ownHashes []common.Hash // Hash chain belonging to the tester - ownHeaders map[common.Hash]*types.Header // Headers belonging to the tester - ownBlocks map[common.Hash]*types.Block // Blocks belonging to the tester - ownReceipts map[common.Hash]types.Receipts // Receipts belonging to the tester - ownChainTd map[common.Hash]*big.Int // Total difficulties of the blocks in the local chain + ownHashes []common.Hash // Hash chain belonging to the tester + ownHeaders map[common.Hash]*types.Header // Headers belonging to the tester + ownBlocks map[common.Hash]*types.Block // Blocks belonging to the tester + ownReceipts map[common.Hash]types.Receipts // Receipts belonging to the tester + ownChainTd map[common.Hash]*big.Int // Total difficulties of the blocks in the local chain + ownCanonical map[uint64]common.Hash // Canonical number-to-hash table, mirroring the real chain's marker writes insertHeaderChainHook func([]*types.Header) error + // parkTailOnce makes the next InsertChain park the batch tail, the way + // insertChain parks future blocks: everything but the tail is written. + // Consumed by that one call. + parkTailOnce bool + // extendTailAfterInsert, when set, is called after the next InsertChain + // with the batch tail and stores the child it returns, simulating a + // concurrent import landing before importBlockResults reads the head. + // Consumed by that one call. + extendTailAfterInsert func(tail *types.Block) *types.Block + + // proposedCalls and lastProposedHeader record handleProposedBlock calls + // for assertions; read them together via proposedState. + proposedCalls int + lastProposedHeader *types.Header // headHeaderCap, when non-zero, caps the height reported by CurrentHeader. // It models the real chain, where importing blocks moves the header head @@ -103,14 +118,15 @@ func newTester() *downloadTester { // shared test genesis. func newTesterWithGenesis(genesis *types.Block, peerDb ethdb.Database) *downloadTester { tester := &downloadTester{ - genesis: genesis, - peerDb: peerDb, - peers: make(map[string]*downloadTesterPeer), - ownHashes: []common.Hash{genesis.Hash()}, - ownHeaders: map[common.Hash]*types.Header{genesis.Hash(): genesis.Header()}, - ownBlocks: map[common.Hash]*types.Block{genesis.Hash(): genesis}, - ownReceipts: map[common.Hash]types.Receipts{genesis.Hash(): nil}, - ownChainTd: map[common.Hash]*big.Int{genesis.Hash(): genesis.Difficulty()}, + genesis: genesis, + peerDb: peerDb, + peers: make(map[string]*downloadTesterPeer), + ownHashes: []common.Hash{genesis.Hash()}, + ownHeaders: map[common.Hash]*types.Header{genesis.Hash(): genesis.Header()}, + ownBlocks: map[common.Hash]*types.Block{genesis.Hash(): genesis}, + ownReceipts: map[common.Hash]types.Receipts{genesis.Hash(): nil}, + ownChainTd: map[common.Hash]*big.Int{genesis.Hash(): genesis.Difficulty()}, + ownCanonical: map[uint64]common.Hash{0: genesis.Hash()}, } tester.stateDb = rawdb.NewMemoryDatabase() tester.triedb = trie.NewDatabase(tester.stateDb) @@ -156,9 +172,13 @@ func (dl *downloadTester) HasHeader(hash common.Hash, number uint64) bool { return dl.GetHeaderByHash(hash) != nil } -// HasBlock checks if a block is present in the testers canonical chain. +// HasBlock checks whether a full block is present at the requested height. func (dl *downloadTester) HasBlock(hash common.Hash, number uint64) bool { - return dl.GetBlockByHash(hash) != nil + dl.lock.RLock() + defer dl.lock.RUnlock() + + block := dl.ownBlocks[hash] + return block != nil && block.NumberU64() == number } // HasFastBlock checks if a block is present in the testers canonical chain. @@ -281,6 +301,10 @@ func (dl *downloadTester) InsertHeaderChain(headers []*types.Header, checkFreq i hashes = append(hashes, hash) } hashes = append(hashes, headers[len(headers)-1].Hash()) + // Anchor of the WriteHeader comparison: the current head header. A nil + // total difficulty means the entry was rolled back. + _, headHash := dl.canonicalHead(false) + headTd := dl.ownChainTd[headHash] // Do a full insert if pre-checks passed for i, header := range headers { hash := hashes[i] @@ -296,6 +320,14 @@ func (dl *downloadTester) InsertHeaderChain(headers []*types.Header, checkFreq i td := dl.getTd(header.ParentHash) dl.ownChainTd[hash] = new(big.Int).Add(td, header.Difficulty) + + // Mirror WriteHeader: a header with a higher total difficulty than + // the head takes over the canonical table, before any body exists. + if headTd == nil || dl.ownChainTd[hash].Cmp(headTd) > 0 { + dl.canonicalize(hash, header.Number.Uint64()) + dl.removeAbove(header.Number.Uint64()) + headTd = dl.ownChainTd[hash] + } } return len(headers), nil } @@ -305,23 +337,109 @@ func (dl *downloadTester) InsertChain(blocks types.Blocks) (i int, err error) { dl.lock.Lock() defer dl.lock.Unlock() + if dl.parkTailOnce { + dl.parkTailOnce = false + // Park the tail the way insertChain does with future blocks. + for i, block := range blocks[:len(blocks)-1] { + if err := dl.storeBlock(block); err != nil { + return i, fmt.Errorf("InsertChain: %v", err) + } + } + return len(blocks) - 1, nil + } + for i, block := range blocks { - if parent, ok := dl.ownBlocks[block.ParentHash()]; !ok { - return i, fmt.Errorf("InsertChain: unknown parent at position %d / %d", i, len(blocks)) - } else if _, err := dl.stateDb.Get(parent.Root().Bytes()); err != nil { - return i, fmt.Errorf("InsertChain: unknown parent state %x: %v", parent.Root(), err) + if err := dl.storeBlock(block); err != nil { + return i, fmt.Errorf("InsertChain: %v", err) } - if _, ok := dl.ownHeaders[block.Hash()]; !ok { - dl.ownHashes = append(dl.ownHashes, block.Hash()) - dl.ownHeaders[block.Hash()] = block.Header() + } + if hook := dl.extendTailAfterInsert; hook != nil { + dl.extendTailAfterInsert = nil + if child := hook(blocks[len(blocks)-1]); child != nil { + if err := dl.storeBlock(child); err != nil { + return len(blocks), fmt.Errorf("InsertChain: %v", err) + } } - dl.ownBlocks[block.Hash()] = block - dl.stateDb.Put(block.Root().Bytes(), []byte{0x00}) - dl.ownChainTd[block.Hash()] = new(big.Int).Add(dl.ownChainTd[block.ParentHash()], block.Difficulty()) } return len(blocks), nil } +// storeBlock records a block in the tester's chain, validating its parent. +// It must be called with dl.lock held for writing. +func (dl *downloadTester) storeBlock(block *types.Block) error { + parent, ok := dl.ownBlocks[block.ParentHash()] + if !ok { + return fmt.Errorf("unknown parent %s", block.ParentHash()) + } + if _, err := dl.stateDb.Get(parent.Root().Bytes()); err != nil { + return fmt.Errorf("unknown parent state %x: %v", parent.Root(), err) + } + if _, ok := dl.ownHeaders[block.Hash()]; !ok { + dl.ownHashes = append(dl.ownHashes, block.Hash()) + dl.ownHeaders[block.Hash()] = block.Header() + } + dl.ownBlocks[block.Hash()] = block + dl.stateDb.Put(block.Root().Bytes(), []byte{0x00}) + dl.ownChainTd[block.Hash()] = new(big.Int).Add(dl.ownChainTd[block.ParentHash()], block.Difficulty()) + // Mirror insertChain: only a block heavier than the head block takes + // over the canonical table; a lighter one is written as a side entry. + _, headHash := dl.canonicalHead(true) + if td := dl.ownChainTd[headHash]; td == nil || dl.ownChainTd[block.Hash()].Cmp(td) > 0 { + dl.canonicalize(block.Hash(), block.NumberU64()) + } + return nil +} + +// canonicalize makes hash the canonical entry at number and rewires the +// segment below it, mirroring the marker writes of WriteHeader and +// insertChain. Caller must hold dl.lock for writing. +func (dl *downloadTester) canonicalize(hash common.Hash, number uint64) { + for n, h := number, hash; n > 0 && dl.ownCanonical[n] != h; n-- { + dl.ownCanonical[n] = h + header := dl.ownHeaders[h] + if header == nil { + break + } + h = header.ParentHash + } +} + +// removeAbove drops canonical entries above number, the stale-marker cleanup +// of WriteHeader. Caller must hold dl.lock for writing. +func (dl *downloadTester) removeAbove(number uint64) { + for n := number + 1; ; n++ { + if _, ok := dl.ownCanonical[n]; !ok { + break + } + delete(dl.ownCanonical, n) + } +} + +// canonicalHead returns the highest canonical entry. With requireBlock, +// entries without a stored block are skipped, giving the head block that +// insertChain reorganizes against; without it, the head header that +// WriteHeader compares against. Caller must hold dl.lock. +func (dl *downloadTester) canonicalHead(requireBlock bool) (uint64, common.Hash) { + headNum, headHash := uint64(0), dl.ownCanonical[0] + for n, h := range dl.ownCanonical { + if n > headNum && (!requireBlock || dl.ownBlocks[h] != nil) { + headNum, headHash = n, h + } + } + return headNum, headHash +} + +// GetCanonicalHash returns the hash of the canonical block at the given +// height, or the zero hash when the height is unknown. Entries are advanced +// by header and block inserts that overtake the head's total difficulty, +// mirroring WriteHeader and insertChain. +func (dl *downloadTester) GetCanonicalHash(number uint64) common.Hash { + dl.lock.RLock() + defer dl.lock.RUnlock() + + return dl.ownCanonical[number] +} + // InsertReceiptChain injects a new batch of receipts into the simulated chain. func (dl *downloadTester) InsertReceiptChain(blocks types.Blocks, receipts []types.Receipts) (i int, err error) { dl.lock.Lock() @@ -375,11 +493,23 @@ func (dl *downloadTester) dropPeer(id string) { dl.downloader.UnregisterPeer(id) } -// an empty handleProposedBlock function +// handleProposedBlock records the invocation and its argument for assertions. func (dl *downloadTester) handleProposedBlock(header *types.Header) error { + dl.lock.Lock() + dl.proposedCalls++ + dl.lastProposedHeader = header + dl.lock.Unlock() return nil } +// proposedState returns the invocation count and the header of the most +// recent handleProposedBlock call. +func (dl *downloadTester) proposedState() (int, *types.Header) { + dl.lock.RLock() + defer dl.lock.RUnlock() + return dl.proposedCalls, dl.lastProposedHeader +} + // Config retrieves the blockchain's chain configuration. func (dl *downloadTester) Config() *params.ChainConfig { if dl.configOverride != nil { @@ -2328,3 +2458,213 @@ func TestDownloaderUnregisterPeerNeverRegistered(t *testing.T) { t.Fatalf("unregistering a never-registered peer error mismatch: got %v want %v", err, errNotRegistered) } } + +// makeTestBlock builds an empty block on top of parent; the seed goes into +// the extra data, so chains from the same parent with different seeds are +// distinct forks. +func makeTestBlock(parent *types.Block, seed byte) *types.Block { + header := &types.Header{ + ParentHash: parent.Hash(), + Number: new(big.Int).Add(parent.Number(), common.Big1), + Difficulty: common.Big1, + GasLimit: params.GenesisGasLimit, + Time: parent.Header().Time + 10, + Extra: []byte{seed}, + } + return types.NewBlockWithHeader(header).WithBody(types.Body{}) +} + +// extendTestChain builds n blocks on top of parent, seeding their extra data +// with seed, seed+1 and so on. parent is not included. +func extendTestChain(parent *types.Block, n int, seed byte) []*types.Block { + chain := make([]*types.Block, 0, n) + for i := 0; i < n; i++ { + block := makeTestBlock(parent, seed+byte(i)) + chain = append(chain, block) + parent = block + } + return chain +} + +// assertProposedBlock checks the handleProposedBlock call count and, when +// wantTail is non-nil, the block the most recent call saw. +func assertProposedBlock(t *testing.T, dl *downloadTester, wantCalls int, wantTail *types.Block) { + t.Helper() + calls, got := dl.proposedState() + if calls != wantCalls { + t.Fatalf("handleProposedBlock ran %d times, want %d", calls, wantCalls) + } + if wantTail == nil { + return + } + if got == nil || got.Hash() != wantTail.Hash() { + t.Fatalf("handleProposedBlock ran on %v, want %v", got, wantTail.Hash()) + } +} + +// TestImportBlockResultsProposedBlockHandler checks that the handler only +// runs for a batch tail that is both stored and canonical, since a nil +// InsertChain also covers a parked tail, a fork batch stored as side +// entries, and a fast-sync height made canonical before its body arrives. +func TestImportBlockResultsProposedBlockHandler(t *testing.T) { + toResults := func(blocks []*types.Block) []*fetchResult { + results := make([]*fetchResult, 0, len(blocks)) + for _, block := range blocks { + results = append(results, &fetchResult{Header: block.Header(), Transactions: types.Transactions{}}) + } + return results + } + + t.Run("batch tail parked in the future queue", func(t *testing.T) { + // A parked tail is neither stored nor canonical, so the handler must + // not run. + dl := newTester() + defer dl.terminate() + + local := extendTestChain(dl.genesis, 4, 0) + if _, err := dl.InsertChain(local); err != nil { + t.Fatalf("failed to set up the local chain: %v", err) + } + // The batch continues the local head, like a queued batch does. + batch := extendTestChain(local[len(local)-1], 4, 16) + tail := batch[len(batch)-1] + dl.parkTailOnce = true + before, _ := dl.proposedState() + if err := dl.downloader.importBlockResults(toResults(batch)); err != nil { + t.Fatalf("failed to import the queued batch: %v", err) + } + assertProposedBlock(t, dl, before, nil) + if dl.GetBlockByHash(tail.Hash()) != nil { + t.Fatalf("queued batch tail unexpectedly stored") + } + if got := dl.GetCanonicalHash(tail.NumberU64()); got != (common.Hash{}) { + t.Fatalf("parked tail height unexpectedly canonical: %v", got) + } + }) + + t.Run("imported batch", func(t *testing.T) { + // A fully written batch triggers the handler exactly once. + dl := newTester() + defer dl.terminate() + + batch := extendTestChain(dl.genesis, 4, 32) + before, _ := dl.proposedState() + if err := dl.downloader.importBlockResults(toResults(batch)); err != nil { + t.Fatalf("failed to import the batch: %v", err) + } + assertProposedBlock(t, dl, before+1, batch[len(batch)-1]) + if dl.GetBlockByHash(batch[len(batch)-1].Hash()) == nil { + t.Fatalf("imported batch tail not stored") + } + }) + + t.Run("stored side-chain batch re-delivered", func(t *testing.T) { + // A stored fork batch that is no longer the head must not reach the + // handler: storage is not canonicality. + dl := newTester() + defer dl.terminate() + + fork := extendTestChain(dl.genesis, 4, 48) + // Freshly stored, the fork is ahead of the local chain, so it + // becomes the head and the handler firing once is expected. + if err := dl.downloader.importBlockResults(toResults(fork)); err != nil { + t.Fatalf("failed to store the fork batch: %v", err) + } + // Growing the local chain past the fork turns its blocks into side + // entries. + local := extendTestChain(dl.genesis, 5, 64) + if _, err := dl.InsertChain(local); err != nil { + t.Fatalf("failed to set up the local chain: %v", err) + } + tail := fork[len(fork)-1] + if dl.GetBlockByHash(tail.Hash()) == nil { + t.Fatalf("fork batch tail not stored") + } + // Re-delivering the stored fork must not reach the handler again. + before, _ := dl.proposedState() + if err := dl.downloader.importBlockResults(toResults(fork)); err != nil { + t.Fatalf("failed to re-import the fork batch: %v", err) + } + assertProposedBlock(t, dl, before, nil) + }) + + t.Run("head advanced past the canonical tail", func(t *testing.T) { + // The tail stays canonical at its own height even once the head + // moved past it, so the handler must still run. + dl := newTester() + defer dl.terminate() + + base := extendTestChain(dl.genesis, 4, 80) + if _, err := dl.InsertChain(base); err != nil { + t.Fatalf("failed to set up the local chain: %v", err) + } + batch := extendTestChain(base[len(base)-1], 4, 96) + dl.extendTailAfterInsert = func(tail *types.Block) *types.Block { + // A concurrent import of the tail's child landing before + // importBlockResults reads the head. + return makeTestBlock(tail, 200) + } + before, _ := dl.proposedState() + if err := dl.downloader.importBlockResults(toResults(batch)); err != nil { + t.Fatalf("failed to import the batch: %v", err) + } + assertProposedBlock(t, dl, before+1, batch[len(batch)-1]) + }) + + t.Run("heavier fork stays canonical", func(t *testing.T) { + // The head goes to the highest total difficulty, not to the last + // insert, so a lighter chain stored later must not displace the + // heavier fork. + dl := newTester() + defer dl.terminate() + + fork := extendTestChain(dl.genesis, 6, 112) + if err := dl.downloader.importBlockResults(toResults(fork)); err != nil { + t.Fatalf("failed to store the fork batch: %v", err) + } + local := extendTestChain(dl.genesis, 2, 128) + if _, err := dl.InsertChain(local); err != nil { + t.Fatalf("failed to set up the local chain: %v", err) + } + // The fork is still canonical, so re-delivering it reaches the + // handler again. + before, _ := dl.proposedState() + if err := dl.downloader.importBlockResults(toResults(fork)); err != nil { + t.Fatalf("failed to re-import the fork batch: %v", err) + } + assertProposedBlock(t, dl, before+1, fork[len(fork)-1]) + }) + + t.Run("fast-sync canonical tail without body", func(t *testing.T) { + // The header phase makes the tail's height canonical before its body + // is imported, so HasBlock has to gate the handler too. + dl := newTester() + defer dl.terminate() + + batch := extendTestChain(dl.genesis, 4, 144) + headers := make([]*types.Header, len(batch)) + for i, block := range batch { + headers[i] = block.Header() + } + if _, err := dl.InsertHeaderChain(headers, 0); err != nil { + t.Fatalf("failed to set up the header phase: %v", err) + } + // The tail height is canonical already, but its body is missing. + tail := batch[len(batch)-1] + if got := dl.GetCanonicalHash(tail.NumberU64()); got != tail.Hash() { + t.Fatalf("tail height not canonical after the header phase: have %v, want %v", got, tail.Hash()) + } + if dl.HasBlock(tail.Hash(), tail.NumberU64()) { + t.Fatalf("tail block unexpectedly stored before the body phase") + } + dl.parkTailOnce = true + before, _ := dl.proposedState() + if err := dl.downloader.importBlockResults(toResults(batch)); err != nil { + t.Fatalf("failed to import the queued batch: %v", err) + } + assertProposedBlock(t, dl, before, nil) + if dl.HasBlock(tail.Hash(), tail.NumberU64()) { + t.Fatalf("queued batch tail unexpectedly stored") + } + }) +} From 4862bd049fa0e88084e105e42bdb20706bee00c3 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Sun, 6 Sep 2026 19:08:01 +0800 Subject: [PATCH 5/5] fix(core,eth): queue all future blocks of a batch instead of failing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Problem ------- XDPoS engines check the header timestamp before the parent lookup with zero tolerance, so the children of a future block also fail verification with ErrFutureBlock rather than ErrUnknownAncestor. The insertChain future-block loops only accepted ErrUnknownAncestor, so a batch whose first or middle block was in the future stopped at its second block and returned the error, which the downloader wraps into errInvalidChain and answers by dropping the peer on a valid delivery. Before #2534 the same scenario silently dropped the tail and failed one batch later with a bogus bad-block report. Changes ------- * Accept ErrFutureBlock in both insertChain future-block loops so the whole tail enters the futureBlocks queue and the import reports success; the queued blocks are imported by procFutureBlocks once their timestamps are reached. Queueing is bounded by addFutureBlock's maxTimeFutureBlocks (30s) window: a tail whose timestamps span past it aborts the batch with the enqueue error (non-sentinel, so the downloader still drops the peer) — the same visible outcome as before — while the in-window prefix stays queued. * Gate HandleProposedBlock on canonicality instead of block existence (importBlockResults now requires the tail to be the current head). A nil InsertChain no longer implies the last block reached the chain, and a queued tail can sit at or below the head height (re-delivered history on a side chain); feeding such a block into processQC overwrites highestQuorumCert before its block-existence check, so a master node could end up voting for a block that is not in the chain. * In the fetcher, move the sign hook, the proposed-block handling and the final broadcast behind a "block exists in the chain" gate. * Evict parked blocks that fail to import: procFutureBlocks used to keep every failed block, so garbage delivered inside the future window could fill the queue permanently and be re-verified and re-reported as bad blocks on every futureBlocksLoop tick forever. A block is now evicted unless it is still in the future or its parent is itself parked; blocks are sorted by number, so evicting a failed parent cascades to its queued children within the same pass. * Collapse the duplicated enqueue loops of the two insertChain future paths into the shared queueFutureTail helper, and drop the dead insertStats accounting on the abort paths (stats.report is only called from the main import loop). Tests ----- Future-block regression tests live in core/blockchain_futureblocks_test.go. --- .../engine_v2_tests/future_block_test.go | 85 +++ core/blockchain.go | 126 +++- core/blockchain_futureblocks_test.go | 556 ++++++++++++++++++ core/blockchain_test.go | 140 +++++ eth/fetcher/block_fetcher.go | 67 +++ eth/fetcher/block_fetcher_test.go | 175 ++++++ 6 files changed, 1125 insertions(+), 24 deletions(-) create mode 100644 consensus/tests/engine_v2_tests/future_block_test.go create mode 100644 core/blockchain_futureblocks_test.go diff --git a/consensus/tests/engine_v2_tests/future_block_test.go b/consensus/tests/engine_v2_tests/future_block_test.go new file mode 100644 index 000000000000..2f39e048e509 --- /dev/null +++ b/consensus/tests/engine_v2_tests/future_block_test.go @@ -0,0 +1,85 @@ +package engine_v2_tests + +import ( + "testing" + "time" + + "github.com/XinFinOrg/XDPoSChain/core/rawdb" + "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/params" + "github.com/stretchr/testify/assert" +) + +// TestInsertChainQueuesFutureBatchEndToEnd combines the two halves the +// future-batch fix relies on: the real XDPoS engine classifying children of a +// future block as ErrFutureBlock (timestamp check precedes the parent lookup), +// and insertChain consuming those results by queueing the whole tail instead +// of failing the import. The core stub-based tests pin the loop mechanics and +// the engine tests pin the classification; neither covers their composition, +// which is the exact online scenario the fix targets. +// +// Deliberately not skipped in -short mode: the whole run (910-block harness +// build included) stays well under a second, and this is the only test that +// covers the composition. +func TestInsertChainQueuesFutureBatchEndToEnd(t *testing.T) { + config := params.TestXDPoSMockChainConfig + blockchain, _, tip910, signer, signFn, _ := PrepareXDCTestBlockChainForV2Engine(t, 910, config, nil) + t.Cleanup(blockchain.Stop) + db := blockchain.ChainDb() + + // Build blocks 911..915 in memory only; none is written into the DB. + switchBlock := config.XDPoS.V2.SwitchBlock.Int64() + blocks := make([]*types.Block, 0, 5) + current := tip910 + for n := 911; n <= 915; n++ { + block := CreateBlock( + blockchain, + blockchain.Config(), + current, + n, + int64(n)-switchBlock, + signer.Hex(), + signer, + signFn, + nil, + nil, + "", + ) + blocks = append(blocks, block) + current = block + } + + // Re-timestamp the tail (912..915) into the near future. The hash changes + // so the QC no longer matches, which is fine: the timestamp check also + // precedes QC verification. Each re-stamped header also breaks the link to + // its built child, so the parents are re-chained along the clone line. Park + // the tail 15s inside addFutureBlock's 30s acceptance window: addFutureBlock + // re-reads time.Now() at enqueue time, so the 15s of slack also absorbs a + // wall-clock step back between stamping here and the enqueue check (a 25s + // offset would leave only 5s and fail on a ~6s NTP step back), and it stays + // far beyond the 100ms futureBlocksLoop period, so the queue cannot drain + // before the assertions run. + now := uint64(time.Now().Unix()) + var batch types.Blocks + batch = append(batch, blocks[0]) // 911 keeps a past timestamp and imports + parent := blocks[0] + for _, block := range blocks[1:] { + header := block.Header() + header.ParentHash = parent.Hash() + header.Time = now + 15 + parent = types.NewBlockWithHeader(header).WithBody(*block.Body()) + batch = append(batch, parent) + } + + n, err := blockchain.InsertChain(batch) + assert.Nil(t, err, "a future tail must not fail the import, the downloader would drop the peer") + assert.Equal(t, len(batch), n) + + // The prefix imported, the tail did not reach the database. + assert.Equal(t, blocks[0].NumberU64(), blockchain.CurrentBlock().Number.Uint64()) + assert.Equal(t, blocks[0].Hash(), blockchain.CurrentBlock().Hash()) + for _, block := range batch[1:] { + assert.Nil(t, blockchain.GetBlockByNumber(block.NumberU64()), "future block %d must stay queued", block.NumberU64()) + assert.Nil(t, rawdb.ReadBadBlock(db, block.Hash()), "future block %d must not be reported as bad block", block.NumberU64()) + } +} diff --git a/core/blockchain.go b/core/blockchain.go index c45fbc015347..de3365f8f212 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -1268,6 +1268,14 @@ func (bc *BlockChain) insertStopped() bool { return bc.procInterrupt.Load() } +// proposedBlockHandler is the consensus hook invoked when the future-block +// queue imports a block that advances the canonical head, letting the engine +// process the new head (QC handling, voting). XDPoS implements it; the small +// interface keeps procFutureBlocks testable with a stub engine. +type proposedBlockHandler interface { + HandleProposedBlock(chain consensus.ChainReader, header *types.Header) error +} + func (bc *BlockChain) procFutureBlocks() { capacity := bc.futureBlocks.Len() if capacity == 0 { @@ -1283,17 +1291,44 @@ func (bc *BlockChain) procFutureBlocks() { types.BlockBy(types.Number).Sort(blocks) // Insert one by one as chain insertion needs contiguous ancestry between blocks + var lastCanon *types.Block for i := range blocks { - _, err := bc.InsertChain(blocks[i : i+1]) - // let consensus engine handle the last block (e.g. for voting) - if i == len(blocks)-1 && err == nil { - engine, ok := bc.Engine().(*XDPoS.XDPoS) - if ok { - header := blocks[i].Header() - err = engine.HandleProposedBlock(bc, header) - if err != nil { - log.Info("[procFutureBlocks] handle proposed block has error", "err", err, "block hash", header.Hash(), "number", header.Number) - } + if _, err := bc.InsertChain(blocks[i : i+1]); err != nil { + // Retryable failures keep the block parked: it is still in the future + // (ErrFutureBlock), or its parent is itself parked in the queue + // (ErrUnknownAncestor) and may import on a later pass. Everything + // else — most notably an ErrUnknownAncestor whose parent is neither + // in the chain nor in the queue — can never succeed on a retry, so + // the block is evicted instead of being re-verified and re-reported + // as a bad block on every futureBlocksLoop tick. Blocks are sorted + // by number, so evicting a failed parent makes the Contains check + // of its queued children fail within the same pass and the whole + // orphaned chain drains. + if errors.Is(err, consensus.ErrFutureBlock) || + (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(blocks[i].ParentHash())) { + continue + } + bc.futureBlocks.Remove(blocks[i].Hash()) + continue + } + // Only a write that advanced the canonical head qualifies for the + // engine hook below: known blocks that are skipped return a nil + // error without importing, and side-chain writes must not be + // treated as the head. Comparing hashes also keeps a same-height + // fork from being mistaken for the canonical head. + if bc.CurrentBlock().Hash() == blocks[i].Hash() { + lastCanon = blocks[i] + } + } + // Let the consensus engine handle the highest imported canonical block + // (e.g. for voting). The sorted queue tail cannot be the target: it may + // have failed to import while lower blocks advanced the head, or sit on + // a side branch. + if lastCanon != nil { + if engine, ok := bc.Engine().(proposedBlockHandler); ok { + header := lastCanon.Header() + if err := engine.HandleProposedBlock(bc, header); err != nil { + log.Info("[procFutureBlocks] handle proposed block has error", "err", err, "block hash", header.Hash(), "number", header.Number) } } } @@ -1707,6 +1742,12 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types. return status, nil } +// isQueueableImportErr reports whether a batch tail block should be parked in the +// future queue instead of failing the import. +func isQueueableImportErr(err error) bool { + return errors.Is(err, consensus.ErrUnknownAncestor) || errors.Is(err, consensus.ErrFutureBlock) +} + // addFutureBlock checks if the block is within the max allowed window to get // accepted for future processing, and returns an error if the block is too far // ahead and was not added. @@ -1719,11 +1760,38 @@ func (bc *BlockChain) addFutureBlock(block *types.Block) error { return nil } +// queueFutureTail parks the batch tail in the future queue, starting at the +// given block whose verification error err is queueable, and stops at the +// first block that fails verification with a non-queueable error or at the end +// of the batch. XDPoS v1/v2 (under full verification) check the timestamp +// before the parent lookup (engine_v1/engine.go, engine_v2/verifyHeader.go), +// so children of a future block surface as ErrFutureBlock. Engines that +// resolve the parent first (e.g. ethash VerifyHeader), or XDPoS without +// fullVerify, answer ErrUnknownAncestor instead; the loop accepts both. +// +// It returns the block and error that stopped the queueing — the caller +// decides whether to report them as bad or ignore them — and a non-nil abort +// error when addFutureBlock rejected the enqueue and the import must fail +// whole. +func (bc *BlockChain) queueFutureTail(it *insertIterator, block *types.Block, err error) (*types.Block, error, error) { + for block != nil && isQueueableImportErr(err) { + if aerr := bc.addFutureBlock(block); aerr != nil { + return block, err, aerr + } + block, err = it.next() + } + return block, err, nil +} + // InsertChain attempts to insert the given batch of blocks in to the canonical // chain or, otherwise, create a fork. If an error is returned it will return // the index number of the failing block as well an error describing what went // wrong. // +// A nil error does not imply every block was written: a tail failing with +// ErrFutureBlock/ErrUnknownAncestor is parked in the future queue and +// processed later. +// // After insertion is done, all accumulated events will be fired. func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) { // Sanity check that we have something meaningful to import @@ -1811,11 +1879,17 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] // First block is future, shove it (and all children) to the future queue (unknown ancestor) case errors.Is(err, consensus.ErrFutureBlock) || (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(it.first().ParentHash())): - for block != nil && (it.index == 0 || errors.Is(err, consensus.ErrUnknownAncestor)) { - if err := bc.addFutureBlock(block); err != nil { - return it.index, events, coalescedLogs, err - } - block, err = it.next() + stopped, err, abortErr := bc.queueFutureTail(it, block, err) + if abortErr != nil { + return it.index, events, coalescedLogs, abortErr + } + // The queueing stopped at a block that failed verification with a + // non-queueable error. Record the reject like the tail path below; + // future, known and pruned-ancestor errors are legitimate states, + // not invalid blocks. + if err != nil && stopped != nil && !errors.Is(err, ErrKnownBlock) && + !errors.Is(err, consensus.ErrFutureBlock) && !errors.Is(err, consensus.ErrPrunedAncestor) { + bc.reportBlock(stopped, nil, err) } return it.index, events, coalescedLogs, err @@ -1930,18 +2004,22 @@ func (bc *BlockChain) insertChain(chain types.Blocks, verifySeals bool) (int, [] // Any blocks remaining here? The only ones we care about are the future ones if block != nil && errors.Is(err, consensus.ErrFutureBlock) { - if err := bc.addFutureBlock(block); err != nil { - return it.index, events, coalescedLogs, err + var abortErr error + block, err, abortErr = bc.queueFutureTail(it, block, err) + if abortErr != nil { + return it.index, events, coalescedLogs, abortErr + } + // The queueing stopped at a block that failed verification with a + // non-queueable error. Record the reject like the first-block failure + // path; future, known and pruned-ancestor errors are legitimate + // states, not invalid blocks. + if err != nil && block != nil && !errors.Is(err, ErrKnownBlock) && + !errors.Is(err, consensus.ErrFutureBlock) && !errors.Is(err, consensus.ErrPrunedAncestor) { + bc.reportBlock(block, nil, err) } - block, err = it.next() + return it.index, events, coalescedLogs, err - for ; block != nil && errors.Is(err, consensus.ErrUnknownAncestor); block, err = it.next() { - if err := bc.addFutureBlock(block); err != nil { - return it.index, events, coalescedLogs, err - } - } } - // Append a single chain head event if we've progressed the chain if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() { log.Debug("New ChainHeadEvent ", "number", lastCanon.NumberU64(), "hash", lastCanon.Hash()) diff --git a/core/blockchain_futureblocks_test.go b/core/blockchain_futureblocks_test.go new file mode 100644 index 000000000000..52be1178502d --- /dev/null +++ b/core/blockchain_futureblocks_test.go @@ -0,0 +1,556 @@ +package core + +import ( + "errors" + "fmt" + "math/big" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/consensus" + "github.com/XinFinOrg/XDPoSChain/consensus/ethash" + "github.com/XinFinOrg/XDPoSChain/core/rawdb" + "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/core/vm" + "github.com/XinFinOrg/XDPoSChain/crypto" + "github.com/XinFinOrg/XDPoSChain/params" +) + +// futureVerifyEngine fails header verification for a single block number, or for +// every block from failFrom on (0 disables the range), so a batch can be made +// to fail in the middle instead of at its first block. failFrom is atomic +// because the chain's future-block loop calls VerifyHeaders concurrently. +type futureVerifyEngine struct { + consensus.Engine + failNumber uint64 + failFrom atomic.Uint64 + failErr error + + // handleMu guards handleCalls: the chain's background future-block loop + // invokes HandleProposedBlock concurrently with the test goroutine. + handleMu sync.Mutex + handleCalls []*types.Header +} + +func (e *futureVerifyEngine) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { + abort := make(chan struct{}) + results := make(chan error, len(headers)) + go func() { + for _, header := range headers { + var err error + number := header.Number.Uint64() + failFrom := e.failFrom.Load() + if number == e.failNumber || (failFrom != 0 && number >= failFrom) { + err = e.failErr + } + select { + case <-abort: + return + case results <- err: + } + } + }() + return abort, results +} + +// HandleProposedBlock records the compensated header so tests can assert which +// block the future-block loop hands to the consensus engine after an import. +// It also makes futureVerifyEngine satisfy the proposedBlockHandler interface +// that procFutureBlocks asserts on the chain engine. +func (e *futureVerifyEngine) HandleProposedBlock(chain consensus.ChainReader, header *types.Header) error { + e.handleMu.Lock() + defer e.handleMu.Unlock() + e.handleCalls = append(e.handleCalls, header) + return nil +} + +func (e *futureVerifyEngine) lastHandled() *types.Header { + e.handleMu.Lock() + defer e.handleMu.Unlock() + if len(e.handleCalls) == 0 { + return nil + } + return e.handleCalls[len(e.handleCalls)-1] +} + +func (e *futureVerifyEngine) handleSnapshot() []*types.Header { + e.handleMu.Lock() + defer e.handleMu.Unlock() + return append([]*types.Header(nil), e.handleCalls...) +} + +// TestInsertChainQueuesMidBatchFutureBlocks verifies that a batch rejected as +// future mid-way has its whole tail queued instead of failing the import: +// children of a future block fail the timestamp check before the parent lookup +// so they surface as ErrFutureBlock too, and returning that error would make +// the downloader drop the peer on a valid delivery. +func TestInsertChainQueuesMidBatchFutureBlocks(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + ) + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 5, nil) + + engine := &futureVerifyEngine{Engine: ethash.NewFaker(), failErr: consensus.ErrFutureBlock} + engine.failFrom.Store(3) + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + if n, err := chain.InsertChain(blocks); err != nil || n != len(blocks) { + t.Fatalf("failed to insert into chain: index %d err %v", n, err) + } + if want := uint64(2); chain.CurrentBlock().Number.Uint64() != want { + t.Fatalf("unexpected head number: have %d want %d", chain.CurrentBlock().Number.Uint64(), want) + } + for _, block := range blocks[2:] { + if !chain.futureBlocks.Contains(block.Hash()) { + t.Fatalf("block %d not queued as future block", block.NumberU64()) + } + if bad := rawdb.ReadBadBlock(db, block.Hash()); bad != nil { + t.Fatalf("future block %d recorded as bad block", block.NumberU64()) + } + } +} + +// TestInsertChainQueuesFutureBatchFromFirstBlock verifies that a batch whose +// first block is already in the future is queued in full instead of failing at +// its second block, which made the downloader drop the delivering peer. +func TestInsertChainQueuesFutureBatchFromFirstBlock(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + ) + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 5, nil) + + engine := &futureVerifyEngine{Engine: ethash.NewFaker(), failErr: consensus.ErrFutureBlock} + engine.failFrom.Store(1) + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + if n, err := chain.InsertChain(blocks); err != nil || n != len(blocks) { + t.Fatalf("failed to insert into chain: index %d err %v", n, err) + } + if want := uint64(0); chain.CurrentBlock().Number.Uint64() != want { + t.Fatalf("unexpected head number: have %d want %d", chain.CurrentBlock().Number.Uint64(), want) + } + for _, block := range blocks { + if !chain.futureBlocks.Contains(block.Hash()) { + t.Fatalf("block %d not queued as future block", block.NumberU64()) + } + } +} + +// TestInsertChainQueuesWrappedFutureError guards the errors.Is comparisons on +// the future-block paths: header verification may hand back a %w-wrapped +// ErrFutureBlock, and bare == comparisons would then abort the import (or +// skip the tail queueing) instead of parking the batch for procFutureBlocks. +func TestInsertChainQueuesWrappedFutureError(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + ) + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 5, nil) + + wrapped := fmt.Errorf("verification: %w", consensus.ErrFutureBlock) + if !errors.Is(wrapped, consensus.ErrFutureBlock) { + t.Fatalf("test setup: error must wrap ErrFutureBlock") + } + + // failFrom=1 rejects the whole batch at its first block (first-block queue + // path), failFrom=3 rejects it mid-way after blocks 1 and 2 imported (tail + // queue path). Either way the wrapped sentinel must still park the tail. + for _, tc := range []struct { + name string + failFrom uint64 + head uint64 + queued int // number of tail blocks that must end up in the future queue + }{ + {name: "first-block", failFrom: 1, head: 0, queued: 5}, + {name: "mid-batch", failFrom: 3, head: 2, queued: 3}, + } { + t.Run(tc.name, func(t *testing.T) { + engine := &futureVerifyEngine{Engine: ethash.NewFaker(), failErr: wrapped} + engine.failFrom.Store(tc.failFrom) + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + if n, err := chain.InsertChain(blocks); err != nil || n != len(blocks) { + t.Fatalf("failed to insert into chain: index %d err %v", n, err) + } + if head := chain.CurrentBlock().Number.Uint64(); head != tc.head { + t.Fatalf("unexpected head number: have %d want %d", head, tc.head) + } + for i, block := range blocks { + if queued, want := chain.futureBlocks.Contains(block.Hash()), i >= len(blocks)-tc.queued; queued != want { + t.Fatalf("block %d queued = %v, want %v", block.NumberU64(), queued, want) + } + if bad := rawdb.ReadBadBlock(db, block.Hash()); bad != nil { + t.Fatalf("block %d recorded as bad block", block.NumberU64()) + } + } + }) + } +} + +// TestInsertChainFutureBlocksBeyondEnqueueWindow pins the maxTimeFutureBlocks +// boundary of the tail queueing: blocks are only parked while their timestamps +// stay within maxTimeFutureBlocks (30s) of the wall clock. Once the tail spans +// past the window, addFutureBlock rejects the first out-of-window block and +// InsertChain fails with that non-sentinel error — the same visible outcome as +// before queueing was extended — while the in-window prefix stays queued for +// procFutureBlocks. +func TestInsertChainFutureBlocksBeyondEnqueueWindow(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + now = uint64(time.Now().Unix()) + ) + // Simulate a 2s block period with the batch starting 18s ahead of the + // clock: blocks 0..5 sit inside the 30s enqueue window, blocks 6.. jump + // past it by a wide margin. The last in-window block stops 2s short of + // the boundary, so a backward wall-clock step (e.g. an NTP correction) + // between the now capture and addFutureBlock's own time.Now read cannot + // push it out of the window; the out-of-window tail would need a 30s + // forward jump to sneak in, which cannot happen within a test run. + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 12, func(i int, gen *BlockGen) { + if i < 6 { + gen.header.Time = now + 18 + uint64(2*i) + } else { + gen.header.Time = now + 60 + uint64(i-6) + } + }) + + engine := &futureVerifyEngine{Engine: ethash.NewFaker(), failErr: consensus.ErrFutureBlock} + engine.failFrom.Store(1) + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + n, err := chain.InsertChain(blocks) + if n != 6 { + t.Fatalf("unexpected failing index: have %d want 6", n) + } + if err == nil || !strings.Contains(err.Error(), "future block timestamp") || errors.Is(err, consensus.ErrFutureBlock) { + t.Fatalf("unexpected error for block %d: %v", n, err) + } + if head := chain.CurrentBlock().Number.Uint64(); head != 0 { + t.Fatalf("unexpected head number: have %d want 0", head) + } + for i, block := range blocks { + if queued, want := chain.futureBlocks.Contains(block.Hash()), i < 6; queued != want { + t.Fatalf("block %d queued = %v, want %v", block.NumberU64(), queued, want) + } + if bad := rawdb.ReadBadBlock(db, block.Hash()); bad != nil { + t.Fatalf("block %d recorded as bad block", block.NumberU64()) + } + } +} + +// TestInsertChainProcFutureBlocksResumesImport proves the queued tail is not +// dropped: once headers verify again, the queued blocks import and the head +// advances to the batch tip. +func TestInsertChainProcFutureBlocksResumesImport(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + ) + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 5, nil) + + engine := &futureVerifyEngine{Engine: ethash.NewFaker(), failErr: consensus.ErrFutureBlock} + engine.failFrom.Store(3) + + chain, err := NewBlockChain(rawdb.NewMemoryDatabase(), nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + if _, err := chain.InsertChain(blocks); err != nil { + t.Fatalf("failed to insert into chain: %v", err) + } + if want := uint64(2); chain.CurrentBlock().Number.Uint64() != want { + t.Fatalf("unexpected head number: have %d want %d", chain.CurrentBlock().Number.Uint64(), want) + } + + // The background future-block loop may drain the queue first; the explicit + // call is then a no-op and the head assertion still holds. + engine.failFrom.Store(0) + // The 100ms futureBlocksLoop also drains the queue; retry until it settles. + for i := 0; i < 100 && chain.CurrentBlock().Number.Uint64() != 5; i++ { + chain.procFutureBlocks() + time.Sleep(10 * time.Millisecond) + } + if want := uint64(5); chain.CurrentBlock().Number.Uint64() != want { + t.Fatalf("unexpected head number: have %d want %d", chain.CurrentBlock().Number.Uint64(), want) + } +} + +// TestProcFutureBlocksHandlesHighestImportedCanonicalBlock verifies that the +// consensus-engine compensation after a future-queue drain targets the highest +// block that actually advanced the canonical head. The historical code only +// compensated the sorted tail of the queue when its own import succeeded, so a +// tail that failed to import (bad block, or still in the future) silently +// skipped the engine hook for the lower blocks that did import, dropping their +// processQC and vote handling. +func TestProcFutureBlocksHandlesHighestImportedCanonicalBlock(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + ) + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 5, nil) + + for _, tc := range []struct { + name string + failNumber uint64 // block number that must fail to import (0 = none) + failErr error + head uint64 // canonical head after the drain, and hook target + }{ + {name: "tail-rejected", failNumber: 5, failErr: errors.New("simulated bad tail block"), head: 4}, + {name: "tail-still-future", failNumber: 5, failErr: consensus.ErrFutureBlock, head: 4}, + {name: "whole-queue-imported", head: 5}, + } { + t.Run(tc.name, func(t *testing.T) { + engine := &futureVerifyEngine{Engine: ethash.NewFaker(), failNumber: tc.failNumber, failErr: tc.failErr} + + chain, err := NewBlockChain(rawdb.NewMemoryDatabase(), nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + // Park the whole batch as future blocks, as a failed import would have. + for _, block := range blocks { + chain.futureBlocks.Add(block.Hash(), block) + } + // The background 100ms future-block loop drains the queue too, so keep + // prodding until the hook has fired for the expected head. + deadline := time.Now().Add(2 * time.Second) + for chain.CurrentBlock().Number.Uint64() != tc.head && time.Now().Before(deadline) { + chain.procFutureBlocks() + time.Sleep(10 * time.Millisecond) + } + if last := engine.lastHandled(); last == nil || last.Number.Uint64() != tc.head { + t.Fatalf("consensus hook not called for imported head %d", tc.head) + } + if head := chain.CurrentBlock().Number.Uint64(); head != tc.head { + t.Fatalf("unexpected head number: have %d want %d", head, tc.head) + } + wantHash := blocks[tc.head-1].Hash() + for _, handled := range engine.handleSnapshot() { + if handled.Number.Uint64() > tc.head { + t.Fatalf("consensus hook called for block %d beyond the imported head %d", handled.Number.Uint64(), tc.head) + } + if handled.Number.Uint64() == tc.head && handled.Hash() != wantHash { + t.Fatalf("consensus hook called for the wrong block at height %d", tc.head) + } + } + }) + } +} + +// mutableErrEngine is a verification stub whose injected error can be switched +// at runtime, modelling a future block whose timestamp expires between the +// original park in the future queue and the queue retry. +type mutableErrEngine struct { + consensus.Engine + verifyErr atomic.Value // error +} + +func (e *mutableErrEngine) setErr(err error) { e.verifyErr.Store(err) } + +func (e *mutableErrEngine) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { + abort := make(chan struct{}) + results := make(chan error, len(headers)) + err, _ := e.verifyErr.Load().(error) + go func() { + for range headers { + select { + case <-abort: + return + case results <- err: + } + } + }() + return abort, results +} + +// TestProcFutureBlocksEvictsUnretryableFutureBlocks pins the eviction rule of +// the future-block drain: a queued block is retried only while it is still in +// the future, or while its parent is itself parked in the queue. A poison batch +// of garbage blocks — delivered while its timestamps sat inside the future +// window and therefore parked in full — must drain from the queue once those +// timestamps expire, instead of being re-verified and re-reported as bad +// blocks on every futureBlocksLoop tick forever. +func TestProcFutureBlocksEvictsUnretryableFutureBlocks(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + now = uint64(time.Now().Unix()) + ) + + // buildPoisonChain returns a contiguous run of count blocks starting at + // height 1 whose root's parent is an unresolvable hash: once their + // timestamps expire, the root is a permanent orphan and every child hangs + // off a still-parked parent, so no block of the chain can ever import. + buildPoisonChain := func(count int) types.Blocks { + blocks := make(types.Blocks, 0, count) + parent := common.Hash{0xde, 0xad} + for i := 0; i < count; i++ { + header := &types.Header{ + ParentHash: parent, + Number: big.NewInt(int64(i + 1)), + GasLimit: params.GenesisGasLimit, + Time: now + 10, + Difficulty: common.Big1, + } + block := types.NewBlockWithHeader(header) + blocks = append(blocks, block) + parent = block.Hash() + } + return blocks + } + + for _, tc := range []struct { + name string + poison bool // garbage chain with an unresolvable root parent + count int // blocks delivered in a single batch + expire bool // switch the verification error to ErrUnknownAncestor + }{ + {name: "small-orphaned-chain-drains", poison: true, count: 5, expire: true}, + {name: "still-future-retained", count: 5}, + } { + t.Run(tc.name, func(t *testing.T) { + var blocks types.Blocks + if tc.poison { + blocks = buildPoisonChain(tc.count) + } else { + _, blocks, _ = GenerateChainWithGenesis(gspec, ethash.NewFaker(), tc.count, nil) + } + + engine := &mutableErrEngine{Engine: ethash.NewFaker()} + engine.setErr(consensus.ErrFutureBlock) // every timestamp sits inside the future window + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + // Deliver the batch while every block is future: the import reports + // success with the whole batch parked in the queue. + if n, err := chain.InsertChain(blocks); err != nil || n != len(blocks) { + t.Fatalf("failed to deliver batch: index %d err %v", n, err) + } + if have := chain.futureBlocks.Len(); have != tc.count { + t.Fatalf("unexpected queue length after delivery: have %d want %d", have, tc.count) + } + + if !tc.expire { + // The blocks are still in the future: retrying must keep them + // parked, and must not report them as bad. + deadline := time.Now().Add(300 * time.Millisecond) + for time.Now().Before(deadline) { + chain.procFutureBlocks() + time.Sleep(10 * time.Millisecond) + } + if have := chain.futureBlocks.Len(); have != tc.count { + t.Fatalf("future blocks evicted while still in the future: have %d want %d", have, tc.count) + } + for _, block := range blocks { + if bad := rawdb.ReadBadBlock(db, block.Hash()); bad != nil { + t.Fatalf("future block %d reported as bad", block.NumberU64()) + } + } + return + } + + // Phase 2: the timestamps expired, so verification now fails on the + // unresolvable ancestor. The queue must drain instead of retrying + // and re-reporting the poison forever. + engine.setErr(consensus.ErrUnknownAncestor) + deadline := time.Now().Add(2 * time.Second) + for chain.futureBlocks.Len() > 0 && time.Now().Before(deadline) { + chain.procFutureBlocks() + time.Sleep(10 * time.Millisecond) + } + if have := chain.futureBlocks.Len(); have != 0 { + t.Fatalf("poisoned queue did not drain: %d blocks still queued", have) + } + if head := chain.CurrentBlock().Number.Uint64(); head != 0 { + t.Fatalf("garbage block imported, head at %d", head) + } + // Every parked block was reported as bad once, during the drain. + for _, block := range blocks { + if bad := rawdb.ReadBadBlock(db, block.Hash()); bad == nil { + t.Fatalf("garbage block %d not reported as bad", block.NumberU64()) + } + } + }) + } +} diff --git a/core/blockchain_test.go b/core/blockchain_test.go index dc747e5e2190..c8064dd521da 100644 --- a/core/blockchain_test.go +++ b/core/blockchain_test.go @@ -22,6 +22,7 @@ import ( "math/rand" "strings" "sync" + "sync/atomic" "testing" "time" @@ -2890,3 +2891,142 @@ func TestDeleteCreateRevert(t *testing.T) { t.Fatalf("block %d: failed to insert into chain: %v", n, err) } } + +// failVerifyEngine fails header verification for a single block number, or for +// every block from failFrom on (0 disables the range), so a batch can be made +// to fail in the middle instead of at its first block. failFrom is atomic +// because the chain's future-block loop calls VerifyHeaders concurrently. +type failVerifyEngine struct { + consensus.Engine + failNumber uint64 + failFrom atomic.Uint64 + failErr error + + // failErrAt overrides the failing error for individual block numbers. + // Set up before the chain starts and read-only afterwards. + failErrAt map[uint64]error +} + +func (e *failVerifyEngine) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) { + abort := make(chan struct{}) + results := make(chan error, len(headers)) + go func() { + for _, header := range headers { + var err error + number := header.Number.Uint64() + failFrom := e.failFrom.Load() + if custom, ok := e.failErrAt[number]; ok { + err = custom + } else if number == e.failNumber || (failFrom != 0 && number >= failFrom) { + err = e.failErr + } + select { + case <-abort: + return + case results <- err: + } + } + }() + return abort, results +} + +// errNonQueueableTest is a verification error that is neither queueable nor +// one of the legitimate skip states (future, known, pruned ancestor). +var errNonQueueableTest = errors.New("non-queueable verification failure") + +// TestInsertChainReportsNonQueueableBlockAfterFuturePrefix verifies that a +// batch stopping at a non-queueable verification error after a future prefix +// records the reject like the tail path, instead of returning the error +// without a bad-block record. +func TestInsertChainReportsNonQueueableBlockAfterFuturePrefix(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + now = uint64(time.Now().Unix()) + ) + // Block 1 is future and parks, block 2 fails with a non-queueable error. + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 2, func(i int, gen *BlockGen) { + gen.header.Time = now + 10 + }) + + engine := &failVerifyEngine{Engine: ethash.NewFaker(), failErr: consensus.ErrFutureBlock} + engine.failFrom.Store(1) + engine.failErrAt = map[uint64]error{2: errNonQueueableTest} + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + n, err := chain.InsertChain(blocks) + if n != 1 { + t.Fatalf("unexpected failing index: have %d want 1", n) + } + if !errors.Is(err, errNonQueueableTest) { + t.Fatalf("unexpected error for block %d: %v", n, err) + } + if bad := rawdb.ReadBadBlock(db, blocks[0].Hash()); bad != nil { + t.Fatalf("future block %d recorded as bad block", blocks[0].NumberU64()) + } + if bad := rawdb.ReadBadBlock(db, blocks[1].Hash()); bad == nil { + t.Fatalf("non-queueable block %d not recorded as bad block", blocks[1].NumberU64()) + } +} + +// TestInsertChainReportsNonQueueableBlockAfterFutureTail verifies that a +// batch whose tail parks a future block and then stops at a non-queueable +// verification error records the reject and propagates the error instead of +// silently swallowing it. +func TestInsertChainReportsNonQueueableBlockAfterFutureTail(t *testing.T) { + var ( + key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291") + address = crypto.PubkeyToAddress(key.PublicKey) + funds = big.NewInt(1000000000000000) + gspec = &Genesis{ + Alloc: types.GenesisAlloc{address: {Balance: funds}}, + BaseFee: big.NewInt(params.InitialBaseFee), + Config: params.TestChainConfig, + } + now = uint64(time.Now().Unix()) + ) + // Block 1 imports, block 2 is future and parks, block 3 fails with a + // non-queueable error. + _, blocks, _ := GenerateChainWithGenesis(gspec, ethash.NewFaker(), 3, func(i int, gen *BlockGen) { + gen.header.Time = now + 10 + }) + + engine := &failVerifyEngine{Engine: ethash.NewFaker()} + engine.failErrAt = map[uint64]error{2: consensus.ErrFutureBlock, 3: errNonQueueableTest} + + db := rawdb.NewMemoryDatabase() + chain, err := NewBlockChain(db, nil, gspec, engine, vm.Config{}) + if err != nil { + t.Fatalf("failed to create tester chain: %v", err) + } + defer chain.Stop() + + n, err := chain.InsertChain(blocks) + if n != 2 { + t.Fatalf("unexpected failing index: have %d want 2", n) + } + if !errors.Is(err, errNonQueueableTest) { + t.Fatalf("unexpected error for block %d: %v", n, err) + } + if bad := rawdb.ReadBadBlock(db, blocks[1].Hash()); bad != nil { + t.Fatalf("future block %d recorded as bad block", blocks[1].NumberU64()) + } + if bad := rawdb.ReadBadBlock(db, blocks[2].Hash()); bad == nil { + t.Fatalf("non-queueable block %d not recorded as bad block", blocks[2].NumberU64()) + } + if !chain.futureBlocks.Contains(blocks[1].Hash()) { + t.Fatalf("future block %d not parked", blocks[1].NumberU64()) + } +} diff --git a/eth/fetcher/block_fetcher.go b/eth/fetcher/block_fetcher.go index caea7b6a3483..6a95c2bf9370 100644 --- a/eth/fetcher/block_fetcher.go +++ b/eth/fetcher/block_fetcher.go @@ -45,6 +45,15 @@ const ( blockLimit = 64 // Maximum number of unique blocks a peer may have delivered ) +const ( + // parkedImportTimeout bounds the wait for the future-block loop to import + // a block that was parked in the future queue before its signature is + // dropped. The enqueue window in core is maxTimeFutureBlocks (30s), so a + // threefold margin covers clock skew and queue congestion. + parkedImportTimeout = 90 * time.Second + parkedImportPollInterval = 100 * time.Millisecond // same cadence as the future-block loop +) + // IsPlausibleAnnouncement reports whether a block announcement at the given // number is within the fetcher's plausibility window of the current chain // height. Untrusted announced numbers must be gated on this check before being @@ -767,6 +776,33 @@ func (f *BlockFetcher) insert(peer string, block *types.Block) { return } + // Signing and consensus handling require the block to actually be in + // the chain: the import can report success while only parking the + // block (or its ancestors) in the future queue. Signing an unimported + // block, or voting on it, would corrupt the consensus state. Relaying + // is safe either way — a future block is simply parked again by the + // receivers — so the broadcast below must not depend on whether the + // block was parked locally. + if f.getBlock(block.Hash()) == nil { + log.Debug("Block parked in the future queue, deferring signing", "peer", peer, "number", block.Number(), "hash", hash) + // The future-block loop imports the block once its timestamp + // arrives; create the signature transaction then. The vote is + // not compensated: procFutureBlocks feeds the imported block + // to the consensus engine itself. Only nodes with a signing + // hook (XDPoS) compensate; without one there is nothing to + // wait for, so no waiter is spawned. + if f.signHook != nil { + go f.compensateSignHook(block) + } + if isM2 { + blockBroadcastOutTimer.UpdateSince(block.ReceivedAt) + go f.broadcastBlock(block, true) + } else { + blockAnnounceOutTimer.UpdateSince(block.ReceivedAt) + go f.broadcastBlock(block, false) + } + return + } if f.signHook != nil { if err := f.signHook(block); err != nil { log.Error("Can't sign the imported block", "err", err) @@ -789,6 +825,37 @@ func (f *BlockFetcher) insert(peer string, block *types.Block) { }() } +// compensateSignHook waits for the future-block loop to import a block that +// was parked in the future queue at delivery time and then runs the signing +// hook that insert had to skip. Voting is not compensated: procFutureBlocks +// feeds the imported block to the consensus engine itself. Giving up after +// parkedImportTimeout only loses the signature of a block that was never +// imported (evicted, rejected or reorged away). +func (f *BlockFetcher) compensateSignHook(block *types.Block) { + hash := block.Hash() + deadline := time.NewTimer(parkedImportTimeout) + defer deadline.Stop() + ticker := time.NewTicker(parkedImportPollInterval) + defer ticker.Stop() + + for f.getBlock(hash) == nil { + select { + case <-f.quit: + return + case <-deadline.C: + log.Warn("Parked block was not imported in time, dropping its signature", "number", block.Number(), "hash", hash) + return + case <-ticker.C: + } + } + + if f.signHook != nil { + if err := f.signHook(block); err != nil { + log.Error("Can't sign the imported block", "err", err) + } + } +} + // forgetHash removes all traces of a block announcement from the fetcher's // internal state. func (f *BlockFetcher) forgetHash(hash common.Hash) { diff --git a/eth/fetcher/block_fetcher_test.go b/eth/fetcher/block_fetcher_test.go index ca5550674d08..68f063dc47b1 100644 --- a/eth/fetcher/block_fetcher_test.go +++ b/eth/fetcher/block_fetcher_test.go @@ -940,3 +940,178 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) { } verifyImportDone(t, imported) } + +// Tests that the consensus handler is skipped and the signing hook deferred +// when the import reported success without the block actually reaching the +// chain (e.g. the block or one of its ancestors was parked in the future +// queue): signing and voting on an unimported block would corrupt consensus +// state, while relaying it is safe and happens regardless. +func TestUnimportedBlockSkipsConsensusHandling(t *testing.T) { + block := types.NewBlockWithHeader(&types.Header{ + ParentHash: genesis.Hash(), + Number: common.Big1, + Difficulty: common.Big1, + GasLimit: params.GenesisGasLimit, + }).WithBody(types.Body{}) + + // Simulate insertBlock queueing the block in the future queue and still + // reporting success: neither the sign hook nor the consensus handler may run. + tester := newTester() + testSkipConsensusHandlingForUnimportedBlock(t, tester, block) + + // A block that actually lands in the chain keeps its consensus handling. + tester = newTester() + testConsensusHandlingForImportedBlock(t, tester, block) +} + +// testSkipConsensusHandlingForUnimportedBlock injects a block whose import +// reports success without storing anything and asserts that neither the +// signing hook nor the consensus handler runs for it: both are consensus +// actions reserved for blocks that actually reached the chain. +func testSkipConsensusHandlingForUnimportedBlock(t *testing.T, tester *fetcherTester, block *types.Block) { + imported := make(chan *types.Block, 1) + tester.fetcher.insertBlock = func(block *types.Block) error { + imported <- block + return nil + } + signed := make(chan *types.Block, 1) + tester.fetcher.signHook = func(block *types.Block) error { + signed <- block + return nil + } + proposed := make(chan *types.Header, 1) + tester.fetcher.handleProposedBlock = func(header *types.Header) error { + proposed <- header + return nil + } + tester.fetcher.Enqueue("test", block) + + // Wait for the import attempt to finish before checking the hooks. + select { + case <-imported: + case <-time.After(5 * time.Second): + t.Fatalf("import of the unimported block never finished") + } + select { + case b := <-signed: + t.Fatalf("sign hook ran for unimported block %v", b.Hash()) + case <-time.After(300 * time.Millisecond): + } + select { + case header := <-proposed: + t.Fatalf("consensus handler ran for unimported block %v", header.Hash()) + case <-time.After(300 * time.Millisecond): + } +} + +// testConsensusHandlingForImportedBlock injects a block that reaches the +// chain and asserts both the signing hook and the consensus handler run for it. +func testConsensusHandlingForImportedBlock(t *testing.T, tester *fetcherTester, block *types.Block) { + signed := make(chan *types.Block, 1) + tester.fetcher.signHook = func(block *types.Block) error { + signed <- block + return nil + } + proposed := make(chan *types.Header, 1) + tester.fetcher.handleProposedBlock = func(header *types.Header) error { + proposed <- header + return nil + } + tester.fetcher.Enqueue("test", block) + + // The signing hook runs before the consensus handler. + select { + case b := <-signed: + if b.Hash() != block.Hash() { + t.Fatalf("sign hook ran for block %v, want %v", b.Hash(), block.Hash()) + } + case <-time.After(5 * time.Second): + t.Fatalf("sign hook never ran for the imported block") + } + select { + case header := <-proposed: + if header.Hash() != block.Hash() { + t.Fatalf("consensus handler ran for block %v, want %v", header.Hash(), block.Hash()) + } + case <-time.After(5 * time.Second): + t.Fatalf("consensus handler never ran for the imported block") + } +} + +// TestParkedBlockSignsAfterFutureImport verifies that a block parked in the +// future queue at delivery time still gets its signature transaction once +// the future-block loop imports it, while the consensus handler stays +// reserved for procFutureBlocks. +func TestParkedBlockSignsAfterFutureImport(t *testing.T) { + block := types.NewBlockWithHeader(&types.Header{ + ParentHash: genesis.Hash(), + Number: common.Big1, + Difficulty: common.Big1, + GasLimit: params.GenesisGasLimit, + }).WithBody(types.Body{}) + + tester := newTester() + defer tester.fetcher.Stop() + + imported := make(chan *types.Block, 1) + tester.fetcher.insertBlock = func(block *types.Block) error { + imported <- block + return nil // block parked, nothing stored + } + signed := make(chan *types.Block, 1) + tester.fetcher.signHook = func(block *types.Block) error { + signed <- block + return nil + } + proposed := make(chan *types.Header, 1) + tester.fetcher.handleProposedBlock = func(header *types.Header) error { + proposed <- header + return nil + } + broadcasts := make(chan *types.Block, 1) + tester.fetcher.broadcastBlock = func(block *types.Block, propagate bool) { + broadcasts <- block + } + tester.fetcher.Enqueue("test", block) + + // Wait for the import attempt: the block is parked, so the signing hook + // and the consensus handler must not run yet. + select { + case <-imported: + case <-time.After(5 * time.Second): + t.Fatalf("import of the parked block never finished") + } + select { + case b := <-signed: + t.Fatalf("sign hook ran before the block was imported: %v", b.Hash()) + case <-time.After(300 * time.Millisecond): + } + // Relaying does not wait for the import. + select { + case b := <-broadcasts: + if b.Hash() != block.Hash() { + t.Fatalf("broadcast ran for block %v, want %v", b.Hash(), block.Hash()) + } + case <-time.After(5 * time.Second): + t.Fatalf("parked block was never broadcast") + } + + // Simulate the future-block loop importing the parked block. + tester.insertChain(types.Blocks{block}) + + // The deferred signing hook now runs for the imported block. + select { + case b := <-signed: + if b.Hash() != block.Hash() { + t.Fatalf("sign hook ran for block %v, want %v", b.Hash(), block.Hash()) + } + case <-time.After(5 * time.Second): + t.Fatalf("sign hook never ran for the imported parked block") + } + // The consensus handler stays reserved for procFutureBlocks. + select { + case header := <-proposed: + t.Fatalf("consensus handler ran for parked block %v", header.Hash()) + case <-time.After(300 * time.Millisecond): + } +}