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 00000000000..01af1dc9c68 --- /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/future_block_test.go b/consensus/tests/engine_v2_tests/future_block_test.go new file mode 100644 index 00000000000..2f39e048e50 --- /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/consensus/tests/engine_v2_tests/verify_header_test.go b/consensus/tests/engine_v2_tests/verify_header_test.go index 9db05ee6fae..b3e5b6b582d 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) +} diff --git a/core/blockchain.go b/core/blockchain.go index 54f5291c889..de3365f8f21 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 @@ -1806,32 +1874,34 @@ 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) { - if err := bc.addFutureBlock(block); err != nil { - return it.index, events, coalescedLogs, err - } - block, err = it.next() + case errors.Is(err, consensus.ErrFutureBlock) || (errors.Is(err, consensus.ErrUnknownAncestor) && bc.futureBlocks.Contains(it.first().ParentHash())): + 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) } - 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 // 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() } @@ -1839,7 +1909,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 } @@ -1934,21 +2003,23 @@ 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 err := bc.addFutureBlock(block); err != nil { - return it.index, events, coalescedLogs, err + if block != nil && errors.Is(err, consensus.ErrFutureBlock) { + 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 && err == consensus.ErrUnknownAncestor; block, err = it.next() { - 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() { log.Debug("New ChainHeadEvent ", "number", lastCanon.NumberU64(), "hash", lastCanon.Hash()) @@ -2064,7 +2135,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 +2296,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 +2357,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 diff --git a/core/blockchain_futureblocks_test.go b/core/blockchain_futureblocks_test.go new file mode 100644 index 00000000000..52be1178502 --- /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_insert.go b/core/blockchain_insert.go index b376b983636..d5ccd15d64c 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 -} diff --git a/core/blockchain_test.go b/core/blockchain_test.go index dc747e5e219..c8064dd521d 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/downloader/downloader.go b/eth/downloader/downloader.go index 05cd1bd8fe7..0bf39643550 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 f7195caf0db..b2f7be6562b 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") + } + }) +} diff --git a/eth/fetcher/block_fetcher.go b/eth/fetcher/block_fetcher.go index caea7b6a348..6a95c2bf937 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 ca5550674d0..68f063dc47b 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): + } +}