Revert "core, eth/downloader: implement pruning mode sync (#31414)"

This reverts commit 90d44e715d.

cmd/geth/chaincmd.go

@@ -482,7 +482,7 @@ func importHistory(ctx *cli.Context) error {
 		network = networks[0]
 	}
 
-	if err := utils.ImportHistory(chain, dir, network); err != nil {
+	if err := utils.ImportHistory(chain, db, dir, network); err != nil {
 		return err
 	}
 	fmt.Printf("Import done in %v\n", time.Since(start))

cmd/utils/cmd.go

@@ -268,9 +268,8 @@ func readList(filename string) ([]string, error) {
 }
 
 // ImportHistory imports Era1 files containing historical block information,
-// starting from genesis. The assumption is held that the provided chain
+// starting from genesis.
-// segment in Era1 file should all be canonical and verified.
+func ImportHistory(chain *core.BlockChain, db ethdb.Database, dir string, network string) error {
-func ImportHistory(chain *core.BlockChain, dir string, network string) error {
 	if chain.CurrentSnapBlock().Number.BitLen() != 0 {
 		return errors.New("history import only supported when starting from genesis")
 	}
@@ -332,6 +331,11 @@ func ImportHistory(chain *core.BlockChain, dir string, network string) error {
 				if err != nil {
 					return fmt.Errorf("error reading receipts %d: %w", it.Number(), err)
 				}
+				if status, err := chain.HeaderChain().InsertHeaderChain([]*types.Header{block.Header()}, start); err != nil {
+					return fmt.Errorf("error inserting header %d: %w", it.Number(), err)
+				} else if status != core.CanonStatTy {
+					return fmt.Errorf("error inserting header %d, not canon: %v", it.Number(), status)
+				}
 				if _, err := chain.InsertReceiptChain([]*types.Block{block}, []types.Receipts{receipts}, 2^64-1); err != nil {
 					return fmt.Errorf("error inserting body %d: %w", it.Number(), err)
 				}

cmd/utils/history_test.go

@@ -171,7 +171,7 @@ func TestHistoryImportAndExport(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unable to initialize chain: %v", err)
 	}
-	if err := ImportHistory(imported, dir, "mainnet"); err != nil {
+	if err := ImportHistory(imported, db2, dir, "mainnet"); err != nil {
 		t.Fatalf("failed to import chain: %v", err)
 	}
 	if have, want := imported.CurrentHeader(), chain.CurrentHeader(); have.Hash() != want.Hash() {

core/blockchain.go

@@ -180,9 +180,10 @@ type CacheConfig struct {
 	SnapshotNoBuild bool // Whether the background generation is allowed
 	SnapshotWait    bool // Wait for snapshot construction on startup. TODO(karalabe): This is a dirty hack for testing, nuke it
 
+	ChainHistoryMode history.HistoryMode
 	// This defines the cutoff block for history expiry.
 	// Blocks before this number may be unavailable in the chain database.
-	ChainHistoryMode history.HistoryMode
+	HistoryPruningCutoff uint64
 }
 
 // triedbConfig derives the configures for trie database.
@@ -391,10 +392,10 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, genesis *Genesis
 	bc.processor = NewStateProcessor(chainConfig, bc.hc, bc)
 
 	genesisHeader := bc.GetHeaderByNumber(0)
-	if genesisHeader == nil {
+	bc.genesisBlock = types.NewBlockWithHeader(genesisHeader)
+	if bc.genesisBlock == nil {
 		return nil, ErrNoGenesis
 	}
-	bc.genesisBlock = types.NewBlockWithHeader(genesisHeader)
 
 	bc.currentBlock.Store(nil)
 	bc.currentSnapBlock.Store(nil)
@@ -1446,6 +1447,7 @@ func (bc *BlockChain) stopWithoutSaving() {
 	// the mutex should become available quickly. It cannot be taken again after Close has
 	// returned.
 	bc.chainmu.Close()
+	bc.wg.Wait()
 }
 
 // Stop stops the blockchain service. If any imports are currently in progress
@@ -1534,36 +1536,45 @@ const (
 	SideStatTy
 )
 
-// InsertReceiptChain inserts a batch of blocks along with their receipts into
+// InsertReceiptChain attempts to complete an already existing header chain with
-// the database. Unlike InsertChain, this function does not verify the state root
+// transaction and receipt data.
-// in the blocks. It is used exclusively for snap sync. All the inserted blocks
-// will be regarded as canonical, chain reorg is not supported.
-//
-// The optional ancientLimit can also be specified and chain segment before that
-// will be directly stored in the ancient, getting rid of the chain migration.
 func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain []types.Receipts, ancientLimit uint64) (int, error) {
-	// Verify the supplied headers before insertion without lock
+	// We don't require the chainMu here since we want to maximize the
-	var headers []*types.Header
+	// concurrency of header insertion and receipt insertion.
-	for _, block := range blockChain {
+	bc.wg.Add(1)
-		headers = append(headers, block.Header())
+	defer bc.wg.Done()
 
-		// Here we also validate that blob transactions in the block do not
+	var (
-		// contain a sidecar. While the sidecar does not affect the block hash
+		ancientBlocks, liveBlocks     types.Blocks
-		// or tx hash, sending blobs within a block is not allowed.
+		ancientReceipts, liveReceipts []types.Receipts
+	)
+	// Do a sanity check that the provided chain is actually ordered and linked
+	for i, block := range blockChain {
+		if i != 0 {
+			prev := blockChain[i-1]
+			if block.NumberU64() != prev.NumberU64()+1 || block.ParentHash() != prev.Hash() {
+				log.Error("Non contiguous receipt insert",
+					"number", block.Number(), "hash", block.Hash(), "parent", block.ParentHash(),
+					"prevnumber", prev.Number(), "prevhash", prev.Hash())
+				return 0, fmt.Errorf("non contiguous insert: item %d is #%d [%x..], item %d is #%d [%x..] (parent [%x..])",
+					i-1, prev.NumberU64(), prev.Hash().Bytes()[:4],
+					i, block.NumberU64(), block.Hash().Bytes()[:4], block.ParentHash().Bytes()[:4])
+			}
+		}
+		if block.NumberU64() <= ancientLimit {
+			ancientBlocks, ancientReceipts = append(ancientBlocks, block), append(ancientReceipts, receiptChain[i])
+		} else {
+			liveBlocks, liveReceipts = append(liveBlocks, block), append(liveReceipts, receiptChain[i])
+		}
+
+		// Here we also validate that blob transactions in the block do not contain a sidecar.
+		// While the sidecar does not affect the block hash / tx hash, sending blobs within a block is not allowed.
 		for txIndex, tx := range block.Transactions() {
 			if tx.Type() == types.BlobTxType && tx.BlobTxSidecar() != nil {
 				return 0, fmt.Errorf("block #%d contains unexpected blob sidecar in tx at index %d", block.NumberU64(), txIndex)
 			}
 		}
 	}
-	if n, err := bc.hc.ValidateHeaderChain(headers); err != nil {
-		return n, err
-	}
-	// Hold the mutation lock
-	if !bc.chainmu.TryLock() {
-		return 0, errChainStopped
-	}
-	defer bc.chainmu.Unlock()
 
 	var (
 		stats = struct{ processed, ignored int32 }{}
@@ -1611,8 +1622,11 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 	// this function only accepts canonical chain data. All side chain will be reverted
 	// eventually.
 	writeAncient := func(blockChain types.Blocks, receiptChain []types.Receipts) (int, error) {
-		// Ensure genesis is in the ancient store
+		first := blockChain[0]
-		if blockChain[0].NumberU64() == 1 {
+		last := blockChain[len(blockChain)-1]
+
+		// Ensure genesis is in ancients.
+		if first.NumberU64() == 1 {
 			if frozen, _ := bc.db.Ancients(); frozen == 0 {
 				td := bc.genesisBlock.Difficulty()
 				writeSize, err := rawdb.WriteAncientBlocks(bc.db, []*types.Block{bc.genesisBlock}, []types.Receipts{nil}, []types.Receipts{nil}, td)
@@ -1728,12 +1742,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 		return 0, nil
 	}
 
-	// writeLive writes the blockchain and corresponding receipt chain to the active store.
+	// writeLive writes blockchain and corresponding receipt chain into active store.
-	//
-	// Notably, in different snap sync cycles, the supplied chain may partially reorganize
-	// existing local chain segments (reorg around the chain tip). The reorganized part
-	// will be included in the provided chain segment, and stale canonical markers will be
-	// silently rewritten. Therefore, no explicit reorg logic is needed.
 	writeLive := func(blockChain types.Blocks, receiptChain []types.Receipts) (int, error) {
 		headers := make([]*types.Header, 0, len(blockChain))
 		var (
@@ -1766,8 +1775,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 				}
 			}
 			// Write all the data out into the database
-			rawdb.WriteCanonicalHash(batch, block.Hash(), block.NumberU64())
+			rawdb.WriteBody(batch, block.Hash(), block.NumberU64(), block.Body())
-			rawdb.WriteBlock(batch, block)
 			rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i])
 
 			// Write everything belongs to the blocks into the database. So that
@@ -1800,13 +1808,9 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 		return 0, nil
 	}
 
-	// Split the supplied blocks into two groups, according to the
+	// Write downloaded chain data and corresponding receipt chain data
-	// given ancient limit.
+	if len(ancientBlocks) > 0 {
-	index := sort.Search(len(blockChain), func(i int) bool {
+		if n, err := writeAncient(ancientBlocks, ancientReceipts); err != nil {
-		return blockChain[i].NumberU64() >= ancientLimit
-	})
-	if index > 0 {
-		if n, err := writeAncient(blockChain[:index], receiptChain[:index]); err != nil {
 			if err == errInsertionInterrupted {
 				return 0, nil
 			}
@@ -1814,8 +1818,8 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 			return n, err
 		}
 	}
-	if index != len(blockChain) {
+	if len(liveBlocks) > 0 {
-		if n, err := writeLive(blockChain[index:], receiptChain[index:]); err != nil {
+		if n, err := writeLive(liveBlocks, liveReceipts); err != nil {
 			if err == errInsertionInterrupted {
 				return 0, nil
 			}
@@ -1837,6 +1841,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
 	}
 
 	log.Debug("Imported new block receipts", context...)
+
 	return 0, nil
 }
 
@@ -3347,6 +3352,7 @@ func (bc *BlockChain) InsertHeaderChain(chain []*types.Header) (int, error) {
 	if i, err := bc.hc.ValidateHeaderChain(chain); err != nil {
 		return i, err
 	}
+
 	if !bc.chainmu.TryLock() {
 		return 0, errChainStopped
 	}
@@ -3359,74 +3365,6 @@ func (bc *BlockChain) GetChainConfig() *params.ChainConfig {
 	return bc.chainConfig
 }
 
-// InsertHeadersBeforeCutoff inserts the given headers into the ancient store
-// as they are claimed older than the configured chain cutoff point. All the
-// inserted headers are regarded as canonical and chain reorg is not supported.
-func (bc *BlockChain) InsertHeadersBeforeCutoff(headers []*types.Header) (int, error) {
-	if len(headers) == 0 {
-		return 0, nil
-	}
-	// TODO(rjl493456442): Headers before the configured cutoff have already
-	// been verified by the hash of cutoff header. Theoretically, header validation
-	// could be skipped here.
-	if n, err := bc.hc.ValidateHeaderChain(headers); err != nil {
-		return n, err
-	}
-	if !bc.chainmu.TryLock() {
-		return 0, errChainStopped
-	}
-	defer bc.chainmu.Unlock()
-
-	// Initialize the ancient store with genesis block if it's empty.
-	var (
-		frozen, _ = bc.db.Ancients()
-		first     = headers[0].Number.Uint64()
-	)
-	if first == 1 && frozen == 0 {
-		_, err := rawdb.WriteAncientBlocks(bc.db, []*types.Block{bc.genesisBlock}, []types.Receipts{nil})
-		if err != nil {
-			log.Error("Error writing genesis to ancients", "err", err)
-			return 0, err
-		}
-		log.Info("Wrote genesis to ancient store")
-	} else if frozen != first {
-		return 0, fmt.Errorf("headers are gapped with the ancient store, first: %d, ancient: %d", first, frozen)
-	}
-
-	// Write headers to the ancient store, with block bodies and receipts set to nil
-	// to ensure consistency across tables in the freezer.
-	_, err := rawdb.WriteAncientHeaderChain(bc.db, headers)
-	if err != nil {
-		return 0, err
-	}
-	if err := bc.db.Sync(); err != nil {
-		return 0, err
-	}
-	// Write hash to number mappings
-	batch := bc.db.NewBatch()
-	for _, header := range headers {
-		rawdb.WriteHeaderNumber(batch, header.Hash(), header.Number.Uint64())
-	}
-	// Write head header and head snap block flags
-	last := headers[len(headers)-1]
-	rawdb.WriteHeadHeaderHash(batch, last.Hash())
-	rawdb.WriteHeadFastBlockHash(batch, last.Hash())
-	if err := batch.Write(); err != nil {
-		return 0, err
-	}
-	// Truncate the useless chain segment (zero bodies and receipts) in the
-	// ancient store.
-	if _, err := bc.db.TruncateTail(last.Number.Uint64() + 1); err != nil {
-		return 0, err
-	}
-	// Last step update all in-memory markers
-	bc.hc.currentHeader.Store(last)
-	bc.currentSnapBlock.Store(last)
-	headHeaderGauge.Update(last.Number.Int64())
-	headFastBlockGauge.Update(last.Number.Int64())
-	return 0, nil
-}
-
 // SetBlockValidatorAndProcessorForTesting sets the current validator and processor.
 // This method can be used to force an invalid blockchain to be verified for tests.
 // This method is unsafe and should only be used before block import starts.

core/blockchain_test.go

@@ -34,7 +34,6 @@ import (
 	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/consensus/beacon"
 	"github.com/ethereum/go-ethereum/consensus/ethash"
-	"github.com/ethereum/go-ethereum/core/history"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -905,6 +904,13 @@ func testFastVsFullChains(t *testing.T, scheme string) {
 	fast, _ := NewBlockChain(fastDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 	defer fast.Stop()
 
+	headers := make([]*types.Header, len(blocks))
+	for i, block := range blocks {
+		headers[i] = block.Header()
+	}
+	if n, err := fast.InsertHeaderChain(headers); err != nil {
+		t.Fatalf("failed to insert header %d: %v", n, err)
+	}
 	if n, err := fast.InsertReceiptChain(blocks, receipts, 0); err != nil {
 		t.Fatalf("failed to insert receipt %d: %v", n, err)
 	}
@@ -919,6 +925,9 @@ func testFastVsFullChains(t *testing.T, scheme string) {
 	ancient, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 	defer ancient.Stop()
 
+	if n, err := ancient.InsertHeaderChain(headers); err != nil {
+		t.Fatalf("failed to insert header %d: %v", n, err)
+	}
 	if n, err := ancient.InsertReceiptChain(blocks, receipts, uint64(len(blocks)/2)); err != nil {
 		t.Fatalf("failed to insert receipt %d: %v", n, err)
 	}
@@ -1057,6 +1066,13 @@ func testLightVsFastVsFullChainHeads(t *testing.T, scheme string) {
 	fast, _ := NewBlockChain(fastDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 	defer fast.Stop()
 
+	headers := make([]*types.Header, len(blocks))
+	for i, block := range blocks {
+		headers[i] = block.Header()
+	}
+	if n, err := fast.InsertHeaderChain(headers); err != nil {
+		t.Fatalf("failed to insert header %d: %v", n, err)
+	}
 	if n, err := fast.InsertReceiptChain(blocks, receipts, 0); err != nil {
 		t.Fatalf("failed to insert receipt %d: %v", n, err)
 	}
@@ -1071,6 +1087,9 @@ func testLightVsFastVsFullChainHeads(t *testing.T, scheme string) {
 	ancient, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 	defer ancient.Stop()
 
+	if n, err := ancient.InsertHeaderChain(headers); err != nil {
+		t.Fatalf("failed to insert header %d: %v", n, err)
+	}
 	if n, err := ancient.InsertReceiptChain(blocks, receipts, uint64(3*len(blocks)/4)); err != nil {
 		t.Fatalf("failed to insert receipt %d: %v", n, err)
 	}
@@ -1085,10 +1104,6 @@ func testLightVsFastVsFullChainHeads(t *testing.T, scheme string) {
 	// Import the chain as a light node and ensure all pointers are updated
 	lightDb := makeDb()
 	defer lightDb.Close()
-	headers := make([]*types.Header, len(blocks))
-	for i, block := range blocks {
-		headers[i] = block.Header()
-	}
 	light, _ := NewBlockChain(lightDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 	if n, err := light.InsertHeaderChain(headers); err != nil {
 		t.Fatalf("failed to insert header %d: %v", n, err)
@@ -2042,6 +2057,13 @@ func testBlockchainRecovery(t *testing.T, scheme string) {
 	defer ancientDb.Close()
 	ancient, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil, nil, nil)
 
+	headers := make([]*types.Header, len(blocks))
+	for i, block := range blocks {
+		headers[i] = block.Header()
+	}
+	if n, err := ancient.InsertHeaderChain(headers); err != nil {
+		t.Fatalf("failed to insert header %d: %v", n, err)
+	}
 	if n, err := ancient.InsertReceiptChain(blocks, receipts, uint64(3*len(blocks)/4)); err != nil {
 		t.Fatalf("failed to insert receipt %d: %v", n, err)
 	}
@@ -2070,6 +2092,82 @@ func testBlockchainRecovery(t *testing.T, scheme string) {
 	}
 }
 
+// This test checks that InsertReceiptChain will roll back correctly when attempting to insert a side chain.
+func TestInsertReceiptChainRollback(t *testing.T) {
+	testInsertReceiptChainRollback(t, rawdb.HashScheme)
+	testInsertReceiptChainRollback(t, rawdb.PathScheme)
+}
+
+func testInsertReceiptChainRollback(t *testing.T, scheme string) {
+	// Generate forked chain. The returned BlockChain object is used to process the side chain blocks.
+	tmpChain, sideblocks, canonblocks, gspec, err := getLongAndShortChains(scheme)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tmpChain.Stop()
+	// Get the side chain receipts.
+	if _, err := tmpChain.InsertChain(sideblocks); err != nil {
+		t.Fatal("processing side chain failed:", err)
+	}
+	t.Log("sidechain head:", tmpChain.CurrentBlock().Number, tmpChain.CurrentBlock().Hash())
+	sidechainReceipts := make([]types.Receipts, len(sideblocks))
+	for i, block := range sideblocks {
+		sidechainReceipts[i] = tmpChain.GetReceiptsByHash(block.Hash())
+	}
+	// Get the canon chain receipts.
+	if _, err := tmpChain.InsertChain(canonblocks); err != nil {
+		t.Fatal("processing canon chain failed:", err)
+	}
+	t.Log("canon head:", tmpChain.CurrentBlock().Number, tmpChain.CurrentBlock().Hash())
+	canonReceipts := make([]types.Receipts, len(canonblocks))
+	for i, block := range canonblocks {
+		canonReceipts[i] = tmpChain.GetReceiptsByHash(block.Hash())
+	}
+
+	// Set up a BlockChain that uses the ancient store.
+	ancientDb, err := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), "", "", false)
+	if err != nil {
+		t.Fatalf("failed to create temp freezer db: %v", err)
+	}
+	defer ancientDb.Close()
+
+	ancientChain, _ := NewBlockChain(ancientDb, DefaultCacheConfigWithScheme(scheme), gspec, nil, ethash.NewFaker(), vm.Config{}, nil)
+	defer ancientChain.Stop()
+
+	// Import the canonical header chain.
+	canonHeaders := make([]*types.Header, len(canonblocks))
+	for i, block := range canonblocks {
+		canonHeaders[i] = block.Header()
+	}
+	if _, err = ancientChain.InsertHeaderChain(canonHeaders); err != nil {
+		t.Fatal("can't import canon headers:", err)
+	}
+
+	// Try to insert blocks/receipts of the side chain.
+	_, err = ancientChain.InsertReceiptChain(sideblocks, sidechainReceipts, uint64(len(sideblocks)))
+	if err == nil {
+		t.Fatal("expected error from InsertReceiptChain.")
+	}
+	if ancientChain.CurrentSnapBlock().Number.Uint64() != 0 {
+		t.Fatalf("failed to rollback ancient data, want %d, have %d", 0, ancientChain.CurrentSnapBlock().Number)
+	}
+	if frozen, err := ancientChain.db.Ancients(); err != nil || frozen != 1 {
+		t.Fatalf("failed to truncate ancient data, frozen index is %d", frozen)
+	}
+
+	// Insert blocks/receipts of the canonical chain.
+	_, err = ancientChain.InsertReceiptChain(canonblocks, canonReceipts, uint64(len(canonblocks)))
+	if err != nil {
+		t.Fatalf("can't import canon chain receipts: %v", err)
+	}
+	if ancientChain.CurrentSnapBlock().Number.Uint64() != canonblocks[len(canonblocks)-1].NumberU64() {
+		t.Fatalf("failed to insert ancient recept chain after rollback")
+	}
+	if frozen, _ := ancientChain.db.Ancients(); frozen != uint64(len(canonblocks))+1 {
+		t.Fatalf("wrong ancients count %d", frozen)
+	}
+}
+
 // Tests that importing a very large side fork, which is larger than the canon chain,
 // but where the difficulty per block is kept low: this means that it will not
 // overtake the 'canon' chain until after it's passed canon by about 200 blocks.
@@ -2358,6 +2456,14 @@ func testInsertKnownChainData(t *testing.T, typ string, scheme string) {
 		}
 	} else if typ == "receipts" {
 		inserter = func(blocks []*types.Block, receipts []types.Receipts) error {
+			headers := make([]*types.Header, 0, len(blocks))
+			for _, block := range blocks {
+				headers = append(headers, block.Header())
+			}
+			_, err := chain.InsertHeaderChain(headers)
+			if err != nil {
+				return err
+			}
 			_, err = chain.InsertReceiptChain(blocks, receipts, 0)
 
 			return err
@@ -2538,6 +2644,14 @@ func testInsertKnownChainDataWithMerging(t *testing.T, typ string, mergeHeight i
 		}
 	} else if typ == "receipts" {
 		inserter = func(blocks []*types.Block, receipts []types.Receipts) error {
+			headers := make([]*types.Header, 0, len(blocks))
+			for _, block := range blocks {
+				headers = append(headers, block.Header())
+			}
+			i, err := chain.InsertHeaderChain(headers)
+			if err != nil {
+				return fmt.Errorf("index %d: %w", i, err)
+			}
 			_, err = chain.InsertReceiptChain(blocks, receipts, 0)
 
 			return err
@@ -4669,241 +4783,3 @@ func TestEIP7702(t *testing.T) {
 		t.Fatalf("addr2 storage wrong: expected %d, got %d", fortyTwo, actual)
 	}
 }
-
-// Tests the scenario that the synchronization target in snap sync has been changed
-// with a chain reorg at the tip. In this case the reorg'd segment should be unmarked
-// with canonical flags.
-func TestChainReorgSnapSync(t *testing.T) {
-	testChainReorgSnapSync(t, 0)
-	testChainReorgSnapSync(t, 32)
-	testChainReorgSnapSync(t, gomath.MaxUint64)
-}
-
-func testChainReorgSnapSync(t *testing.T, ancientLimit uint64) {
-	// log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true)))
-
-	// Configure and generate a sample block chain
-	var (
-		key, _  = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
-		address = crypto.PubkeyToAddress(key.PublicKey)
-		funds   = big.NewInt(1000000000000000)
-		gspec   = &Genesis{
-			Config:  params.TestChainConfig,
-			Alloc:   types.GenesisAlloc{address: {Balance: funds}},
-			BaseFee: big.NewInt(params.InitialBaseFee),
-		}
-		signer = types.LatestSigner(gspec.Config)
-		engine = beacon.New(ethash.NewFaker())
-	)
-	genDb, blocks, receipts := GenerateChainWithGenesis(gspec, engine, 32, func(i int, block *BlockGen) {
-		block.SetCoinbase(common.Address{0x00})
-
-		// If the block number is multiple of 3, send a few bonus transactions to the miner
-		if i%3 == 2 {
-			for j := 0; j < i%4+1; j++ {
-				tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, block.header.BaseFee, nil), signer, key)
-				if err != nil {
-					panic(err)
-				}
-				block.AddTx(tx)
-			}
-		}
-	})
-	chainA, receiptsA := GenerateChain(gspec.Config, blocks[len(blocks)-1], engine, genDb, 16, func(i int, gen *BlockGen) {
-		gen.SetCoinbase(common.Address{0: byte(0xa), 19: byte(i)})
-	})
-	chainB, receiptsB := GenerateChain(gspec.Config, blocks[len(blocks)-1], engine, genDb, 20, func(i int, gen *BlockGen) {
-		gen.SetCoinbase(common.Address{0: byte(0xb), 19: byte(i)})
-	})
-
-	db, _ := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), "", "", false)
-	defer db.Close()
-
-	chain, _ := NewBlockChain(db, DefaultCacheConfigWithScheme(rawdb.PathScheme), gspec, nil, beacon.New(ethash.NewFaker()), vm.Config{}, nil)
-	defer chain.Stop()
-
-	if n, err := chain.InsertReceiptChain(blocks, receipts, ancientLimit); err != nil {
-		t.Fatalf("failed to insert receipt %d: %v", n, err)
-	}
-	if n, err := chain.InsertReceiptChain(chainA, receiptsA, ancientLimit); err != nil {
-		t.Fatalf("failed to insert receipt %d: %v", n, err)
-	}
-	// If the common ancestor is below the ancient limit, rewind the chain head.
-	// It's aligned with the behavior in the snap sync
-	ancestor := blocks[len(blocks)-1].NumberU64()
-	if ancestor < ancientLimit {
-		rawdb.WriteLastPivotNumber(db, ancestor)
-		chain.SetHead(ancestor)
-	}
-	if n, err := chain.InsertReceiptChain(chainB, receiptsB, ancientLimit); err != nil {
-		t.Fatalf("failed to insert receipt %d: %v", n, err)
-	}
-	head := chain.CurrentSnapBlock()
-	if head.Hash() != chainB[len(chainB)-1].Hash() {
-		t.Errorf("head snap block #%d: header mismatch: want: %v, got: %v", head.Number, chainB[len(chainB)-1].Hash(), head.Hash())
-	}
-
-	// Iterate over all chain data components, and cross reference
-	for i := 0; i < len(blocks); i++ {
-		num, hash := blocks[i].NumberU64(), blocks[i].Hash()
-		header := chain.GetHeaderByNumber(num)
-		if header.Hash() != hash {
-			t.Errorf("block #%d: header mismatch: want: %v, got: %v", num, hash, header.Hash())
-		}
-	}
-	for i := 0; i < len(chainA); i++ {
-		num, hash := chainA[i].NumberU64(), chainA[i].Hash()
-		header := chain.GetHeaderByNumber(num)
-		if header == nil {
-			continue
-		}
-		if header.Hash() == hash {
-			t.Errorf("block #%d: unexpected canonical header: %v", num, hash)
-		}
-	}
-	for i := 0; i < len(chainB); i++ {
-		num, hash := chainB[i].NumberU64(), chainB[i].Hash()
-		header := chain.GetHeaderByNumber(num)
-		if header.Hash() != hash {
-			t.Errorf("block #%d: header mismatch: want: %v, got: %v", num, hash, header.Hash())
-		}
-	}
-}
-
-// Tests the scenario that all the inserted chain segment are with the configured
-// chain cutoff point. In this case the chain segment before the cutoff should
-// be persisted without the receipts and bodies; chain after should be persisted
-// normally.
-func TestInsertChainWithCutoff(t *testing.T) {
-	const chainLength = 64
-
-	// Configure and generate a sample block chain
-	var (
-		key, _  = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
-		address = crypto.PubkeyToAddress(key.PublicKey)
-		funds   = big.NewInt(1000000000000000)
-		gspec   = &Genesis{
-			Config:  params.TestChainConfig,
-			Alloc:   types.GenesisAlloc{address: {Balance: funds}},
-			BaseFee: big.NewInt(params.InitialBaseFee),
-		}
-		signer = types.LatestSigner(gspec.Config)
-		engine = beacon.New(ethash.NewFaker())
-	)
-	_, blocks, receipts := GenerateChainWithGenesis(gspec, engine, chainLength, func(i int, block *BlockGen) {
-		block.SetCoinbase(common.Address{0x00})
-		tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, block.header.BaseFee, nil), signer, key)
-		if err != nil {
-			panic(err)
-		}
-		block.AddTx(tx)
-	})
-
-	// Run the actual tests.
-	t.Run("cutoff-32/ancientLimit-32", func(t *testing.T) {
-		// cutoff = 32, ancientLimit = 32
-		testInsertChainWithCutoff(t, 32, 32, gspec, blocks, receipts)
-	})
-	t.Run("cutoff-32/ancientLimit-64", func(t *testing.T) {
-		// cutoff = 32, ancientLimit = 64 (entire chain in ancient)
-		testInsertChainWithCutoff(t, 32, 64, gspec, blocks, receipts)
-	})
-	t.Run("cutoff-32/ancientLimit-64", func(t *testing.T) {
-		// cutoff = 32, ancientLimit = 65 (64 blocks in ancient, 1 block in live)
-		testInsertChainWithCutoff(t, 32, 65, gspec, blocks, receipts)
-	})
-}
-
-func testInsertChainWithCutoff(t *testing.T, cutoff uint64, ancientLimit uint64, genesis *Genesis, blocks []*types.Block, receipts []types.Receipts) {
-	// log.SetDefault(log.NewLogger(log.NewTerminalHandlerWithLevel(os.Stderr, log.LevelDebug, true)))
-
-	// Add a known pruning point for the duration of the test.
-	ghash := genesis.ToBlock().Hash()
-	cutoffBlock := blocks[cutoff-1]
-	history.PrunePoints[ghash] = &history.PrunePoint{
-		BlockNumber: cutoffBlock.NumberU64(),
-		BlockHash:   cutoffBlock.Hash(),
-	}
-	defer func() {
-		delete(history.PrunePoints, ghash)
-	}()
-
-	// Enable pruning in cache config.
-	config := DefaultCacheConfigWithScheme(rawdb.PathScheme)
-	config.ChainHistoryMode = history.KeepPostMerge
-
-	db, _ := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), "", "", false)
-	defer db.Close()
-	chain, _ := NewBlockChain(db, DefaultCacheConfigWithScheme(rawdb.PathScheme), genesis, nil, beacon.New(ethash.NewFaker()), vm.Config{}, nil)
-	defer chain.Stop()
-
-	var (
-		headersBefore []*types.Header
-		blocksAfter   []*types.Block
-		receiptsAfter []types.Receipts
-	)
-	for i, b := range blocks {
-		if b.NumberU64() < cutoffBlock.NumberU64() {
-			headersBefore = append(headersBefore, b.Header())
-		} else {
-			blocksAfter = append(blocksAfter, b)
-			receiptsAfter = append(receiptsAfter, receipts[i])
-		}
-	}
-	if n, err := chain.InsertHeadersBeforeCutoff(headersBefore); err != nil {
-		t.Fatalf("failed to insert headers before cutoff %d: %v", n, err)
-	}
-	if n, err := chain.InsertReceiptChain(blocksAfter, receiptsAfter, ancientLimit); err != nil {
-		t.Fatalf("failed to insert receipt %d: %v", n, err)
-	}
-	headSnap := chain.CurrentSnapBlock()
-	if headSnap.Hash() != blocks[len(blocks)-1].Hash() {
-		t.Errorf("head snap block #%d: header mismatch: want: %v, got: %v", headSnap.Number, blocks[len(blocks)-1].Hash(), headSnap.Hash())
-	}
-	headHeader := chain.CurrentHeader()
-	if headHeader.Hash() != blocks[len(blocks)-1].Hash() {
-		t.Errorf("head header #%d: header mismatch: want: %v, got: %v", headHeader.Number, blocks[len(blocks)-1].Hash(), headHeader.Hash())
-	}
-	headBlock := chain.CurrentBlock()
-	if headBlock.Hash() != ghash {
-		t.Errorf("head block #%d: header mismatch: want: %v, got: %v", headBlock.Number, ghash, headBlock.Hash())
-	}
-
-	// Iterate over all chain data components, and cross reference
-	for i := 0; i < len(blocks); i++ {
-		num, hash := blocks[i].NumberU64(), blocks[i].Hash()
-
-		// Canonical headers should be visible regardless of cutoff
-		header := chain.GetHeaderByNumber(num)
-		if header.Hash() != hash {
-			t.Errorf("block #%d: header mismatch: want: %v, got: %v", num, hash, header.Hash())
-		}
-		tail, err := db.Tail()
-		if err != nil {
-			t.Fatalf("Failed to get chain tail, %v", err)
-		}
-		if tail != cutoffBlock.NumberU64() {
-			t.Fatalf("Unexpected chain tail, want: %d, got: %d", cutoffBlock.NumberU64(), tail)
-		}
-		// Block bodies and receipts before the cutoff should be non-existent
-		if num < cutoffBlock.NumberU64() {
-			body := chain.GetBody(hash)
-			if body != nil {
-				t.Fatalf("Unexpected block body: %d, cutoff: %d", num, cutoffBlock.NumberU64())
-			}
-			receipts := chain.GetReceiptsByHash(hash)
-			if receipts != nil {
-				t.Fatalf("Unexpected block receipts: %d, cutoff: %d", num, cutoffBlock.NumberU64())
-			}
-		} else {
-			body := chain.GetBody(hash)
-			if body == nil || len(body.Transactions) != 1 {
-				t.Fatalf("Missed block body: %d, cutoff: %d", num, cutoffBlock.NumberU64())
-			}
-			receipts := chain.GetReceiptsByHash(hash)
-			if receipts == nil || len(receipts) != 1 {
-				t.Fatalf("Missed block receipts: %d, cutoff: %d", num, cutoffBlock.NumberU64())
-			}
-		}
-	}
-}

core/error.go

@@ -31,6 +31,8 @@ var (
 
 	// ErrNoGenesis is returned when there is no Genesis Block.
 	ErrNoGenesis = errors.New("genesis not found in chain")
+
+	errSideChainReceipts = errors.New("side blocks can't be accepted as ancient chain data")
 )
 
 // List of evm-call-message pre-checking errors. All state transition messages will

core/headerchain.go

@@ -271,8 +271,7 @@ func (hc *HeaderChain) WriteHeaders(headers []*types.Header) (int, error) {
 }
 
 // writeHeadersAndSetHead writes a batch of block headers and applies the last
-// header as the chain head.
+// header as the chain head if the fork choicer says it's ok to update the chain.
-//
 // Note: This method is not concurrent-safe with inserting blocks simultaneously
 // into the chain, as side effects caused by reorganisations cannot be emulated
 // without the real blocks. Hence, writing headers directly should only be done
@@ -332,14 +331,12 @@ func (hc *HeaderChain) writeHeadersAndSetHead(headers []*types.Header, forker *F
 	return result, nil
 }
 
-// ValidateHeaderChain verifies that the supplied header chain is contiguous
-// and conforms to consensus rules.
 func (hc *HeaderChain) ValidateHeaderChain(chain []*types.Header) (int, error) {
 	// Do a sanity check that the provided chain is actually ordered and linked
 	for i := 1; i < len(chain); i++ {
 		if chain[i].Number.Uint64() != chain[i-1].Number.Uint64()+1 {
-			hash, parentHash := chain[i].Hash(), chain[i-1].Hash()
+			hash := chain[i].Hash()
-
+			parentHash := chain[i-1].Hash()
 			// Chain broke ancestry, log a message (programming error) and skip insertion
 			log.Error("Non contiguous header insert", "number", chain[i].Number, "hash", hash,
 				"parent", chain[i].ParentHash, "prevnumber", chain[i-1].Number, "prevhash", parentHash)
@@ -364,6 +361,7 @@ func (hc *HeaderChain) ValidateHeaderChain(chain []*types.Header) (int, error) {
 			return i, err
 		}
 	}
+
 	return 0, nil
 }
 

core/rawdb/accessors_chain.go

@@ -912,30 +912,6 @@ func writeAncientBlock(op ethdb.AncientWriteOp, block *types.Block, header *type
 	return nil
 }
 
-// WriteAncientHeaderChain writes the supplied headers along with nil block
-// bodies and receipts into the ancient store. It's supposed to be used for
-// storing chain segment before the chain cutoff.
-func WriteAncientHeaderChain(db ethdb.AncientWriter, headers []*types.Header) (int64, error) {
-	return db.ModifyAncients(func(op ethdb.AncientWriteOp) error {
-		for _, header := range headers {
-			num := header.Number.Uint64()
-			if err := op.AppendRaw(ChainFreezerHashTable, num, header.Hash().Bytes()); err != nil {
-				return fmt.Errorf("can't add block %d hash: %v", num, err)
-			}
-			if err := op.Append(ChainFreezerHeaderTable, num, header); err != nil {
-				return fmt.Errorf("can't append block header %d: %v", num, err)
-			}
-			if err := op.AppendRaw(ChainFreezerBodiesTable, num, nil); err != nil {
-				return fmt.Errorf("can't append block body %d: %v", num, err)
-			}
-			if err := op.AppendRaw(ChainFreezerReceiptTable, num, nil); err != nil {
-				return fmt.Errorf("can't append block %d receipts: %v", num, err)
-			}
-		}
-		return nil
-	})
-}
-
 // DeleteBlock removes all block data associated with a hash.
 func DeleteBlock(db ethdb.KeyValueWriter, hash common.Hash, number uint64) {
 	DeleteReceipts(db, hash, number)

core/rawdb/accessors_chain_test.go

@@ -535,48 +535,6 @@ func TestAncientStorage(t *testing.T) {
 	}
 }
 
-func TestWriteAncientHeaderChain(t *testing.T) {
-	db, err := NewDatabaseWithFreezer(NewMemoryDatabase(), t.TempDir(), "", false)
-	if err != nil {
-		t.Fatalf("failed to create database with ancient backend")
-	}
-	defer db.Close()
-
-	// Create a test block
-	var headers []*types.Header
-	headers = append(headers, &types.Header{
-		Number:      big.NewInt(0),
-		Extra:       []byte("test block"),
-		UncleHash:   types.EmptyUncleHash,
-		TxHash:      types.EmptyTxsHash,
-		ReceiptHash: types.EmptyReceiptsHash,
-	})
-	headers = append(headers, &types.Header{
-		Number:      big.NewInt(1),
-		Extra:       []byte("test block"),
-		UncleHash:   types.EmptyUncleHash,
-		TxHash:      types.EmptyTxsHash,
-		ReceiptHash: types.EmptyReceiptsHash,
-	})
-	// Write and verify the header in the database
-	WriteAncientHeaderChain(db, headers)
-
-	for _, header := range headers {
-		if blob := ReadHeaderRLP(db, header.Hash(), header.Number.Uint64()); len(blob) == 0 {
-			t.Fatalf("no header returned")
-		}
-		if h := ReadCanonicalHash(db, header.Number.Uint64()); h != header.Hash() {
-			t.Fatalf("no canonical hash returned")
-		}
-		if blob := ReadBodyRLP(db, header.Hash(), header.Number.Uint64()); len(blob) != 0 {
-			t.Fatalf("unexpected body returned")
-		}
-		if blob := ReadReceiptsRLP(db, header.Hash(), header.Number.Uint64()); len(blob) != 0 {
-			t.Fatalf("unexpected body returned")
-		}
-	}
-}
-
 func TestCanonicalHashIteration(t *testing.T) {
 	var cases = []struct {
 		from, to uint64

core/txindexer.go

@@ -68,11 +68,11 @@ type txIndexer struct {
 
 // newTxIndexer initializes the transaction indexer.
 func newTxIndexer(limit uint64, chain *BlockChain) *txIndexer {
-	cutoff, _ := chain.HistoryPruningCutoff()
 	indexer := &txIndexer{
 		limit:    limit,
-		cutoff: cutoff,
+		cutoff:   chain.HistoryPruningCutoff(),
 		db:       chain.db,
+		progress: make(chan chan TxIndexProgress),
 		term:     make(chan chan struct{}),
 		closed:   make(chan struct{}),
 	}

eth/backend.go

@@ -238,6 +238,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 			StateScheme:          scheme,
 			TriesInMemory:        config.TriesInMemory,
 			ChainHistoryMode:     config.HistoryMode,
+			HistoryPruningCutoff: historyPruningCutoff,
 		}
 	)
 

eth/downloader/metrics.go

@@ -25,6 +25,7 @@ import (
 var (
 	headerInMeter      = metrics.NewRegisteredMeter("eth/downloader/headers/in", nil)
 	headerReqTimer     = metrics.NewRegisteredTimer("eth/downloader/headers/req", nil)
+	headerDropMeter    = metrics.NewRegisteredMeter("eth/downloader/headers/drop", nil)
 	headerTimeoutMeter = metrics.NewRegisteredMeter("eth/downloader/headers/timeout", nil)
 
 	bodyInMeter      = metrics.NewRegisteredMeter("eth/downloader/bodies/in", nil)

eth/downloader/queue.go

@@ -72,7 +72,7 @@ type fetchResult struct {
 	Withdrawals  types.Withdrawals
 }
 
-func newFetchResult(header *types.Header, snapSync bool) *fetchResult {
+func newFetchResult(header *types.Header, fastSync bool) *fetchResult {
 	item := &fetchResult{
 		Header: header,
 	}
@@ -81,7 +81,7 @@ func newFetchResult(header *types.Header, snapSync bool) *fetchResult {
 	} else if header.WithdrawalsHash != nil {
 		item.Withdrawals = make(types.Withdrawals, 0)
 	}
-	if snapSync && !header.EmptyReceipts() {
+	if fastSync && !header.EmptyReceipts() {
 		item.pending.Store(item.pending.Load() | (1 << receiptType))
 	}
 
@@ -126,7 +126,18 @@ func (f *fetchResult) Done(kind uint) bool {
 // queue represents hashes that are either need fetching or are being fetched
 type queue struct {
 	mode SyncMode // Synchronisation mode to decide on the block parts to schedule for fetching
+
+	// Headers are "special", they download in batches, supported by a skeleton chain
 	headerHead      common.Hash                    // Hash of the last queued header to verify order
+	headerTaskPool  map[uint64]*types.Header       // Pending header retrieval tasks, mapping starting indexes to skeleton headers
+	headerTaskQueue *prque.Prque[int64, uint64]    // Priority queue of the skeleton indexes to fetch the filling headers for
+	headerPeerMiss  map[string]map[uint64]struct{} // Set of per-peer header batches known to be unavailable
+	headerPendPool  map[string]*fetchRequest       // Currently pending header retrieval operations
+	headerResults   []*types.Header                // Result cache accumulating the completed headers
+	headerHashes    []common.Hash                  // Result cache accumulating the completed header hashes
+	headerProced    int                            // Number of headers already processed from the results
+	headerOffset    uint64                         // Number of the first header in the result cache
+	headerContCh    chan bool                      // Channel to notify when header download finishes
 
 	// All data retrievals below are based on an already assembles header chain
 	blockTaskPool  map[common.Hash]*types.Header      // Pending block (body) retrieval tasks, mapping hashes to headers
@@ -153,6 +164,7 @@ type queue struct {
 func newQueue(blockCacheLimit int, thresholdInitialSize int) *queue {
 	lock := new(sync.RWMutex)
 	q := &queue{
+		headerContCh:     make(chan bool, 1),
 		blockTaskQueue:   prque.New[int64, *types.Header](nil),
 		blockWakeCh:      make(chan bool, 1),
 		receiptTaskQueue: prque.New[int64, *types.Header](nil),
@@ -174,6 +186,7 @@ func (q *queue) Reset(blockCacheLimit int, thresholdInitialSize int) {
 	q.mode = FullSync
 
 	q.headerHead = common.Hash{}
+	q.headerPendPool = make(map[string]*fetchRequest)
 
 	q.blockTaskPool = make(map[common.Hash]*types.Header)
 	q.blockTaskQueue.Reset()
@@ -196,6 +209,14 @@ func (q *queue) Close() {
 	q.lock.Unlock()
 }
 
+// PendingHeaders retrieves the number of header requests pending for retrieval.
+func (q *queue) PendingHeaders() int {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	return q.headerTaskQueue.Size()
+}
+
 // PendingBodies retrieves the number of block body requests pending for retrieval.
 func (q *queue) PendingBodies() int {
 	q.lock.Lock()
@@ -241,14 +262,54 @@ func (q *queue) Idle() bool {
 	return (queued + pending) == 0
 }
 
+// ScheduleSkeleton adds a batch of header retrieval tasks to the queue to fill
+// up an already retrieved header skeleton.
+func (q *queue) ScheduleSkeleton(from uint64, skeleton []*types.Header) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	// No skeleton retrieval can be in progress, fail hard if so (huge implementation bug)
+	if q.headerResults != nil {
+		panic("skeleton assembly already in progress")
+	}
+	// Schedule all the header retrieval tasks for the skeleton assembly
+	q.headerTaskPool = make(map[uint64]*types.Header)
+	q.headerTaskQueue = prque.New[int64, uint64](nil)
+	q.headerPeerMiss = make(map[string]map[uint64]struct{}) // Reset availability to correct invalid chains
+	q.headerResults = make([]*types.Header, len(skeleton)*MaxHeaderFetch)
+	q.headerHashes = make([]common.Hash, len(skeleton)*MaxHeaderFetch)
+	q.headerProced = 0
+	q.headerOffset = from
+	q.headerContCh = make(chan bool, 1)
+
+	for i, header := range skeleton {
+		index := from + uint64(i*MaxHeaderFetch)
+
+		q.headerTaskPool[index] = header
+		q.headerTaskQueue.Push(index, -int64(index))
+	}
+}
+
+// RetrieveHeaders retrieves the header chain assemble based on the scheduled
+// skeleton.
+func (q *queue) RetrieveHeaders() ([]*types.Header, []common.Hash, int) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+
+	headers, hashes, proced := q.headerResults, q.headerHashes, q.headerProced
+	q.headerResults, q.headerHashes, q.headerProced = nil, nil, 0
+
+	return headers, hashes, proced
+}
+
 // Schedule adds a set of headers for the download queue for scheduling, returning
 // the new headers encountered.
-func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uint64) int {
+func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uint64) []*types.Header {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
 	// Insert all the headers prioritised by the contained block number
-	var inserts int
+	inserts := make([]*types.Header, 0, len(headers))
 	for i, header := range headers {
 		// Make sure chain order is honoured and preserved throughout
 		hash := hashes[i]
@@ -279,7 +340,7 @@ func (q *queue) Schedule(headers []*types.Header, hashes []common.Hash, from uin
 				q.receiptTaskQueue.Push(header, -int64(header.Number.Uint64()))
 			}
 		}
-		inserts++
+		inserts = append(inserts, header)
 		q.headerHead = hash
 		from++
 	}
@@ -562,6 +623,10 @@ func (q *queue) Revoke(peerID string) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
+	if request, ok := q.headerPendPool[peerID]; ok {
+		q.headerTaskQueue.Push(request.From, -int64(request.From))
+		delete(q.headerPendPool, peerID)
+	}
 	if request, ok := q.blockPendPool[peerID]; ok {
 		for _, header := range request.Headers {
 			q.blockTaskQueue.Push(header, -int64(header.Number.Uint64()))

eth/downloader/resultstore.go

@@ -78,7 +78,7 @@ func (r *resultStore) SetThrottleThreshold(threshold uint64) uint64 {
 //	throttled - if true, the store is at capacity, this particular header is not prio now
 //	item      - the result to store data into
 //	err       - any error that occurred
-func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, throttled bool, item *fetchResult, err error) {
+func (r *resultStore) AddFetch(header *types.Header, fastSync bool) (stale, throttled bool, item *fetchResult, err error) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
@@ -90,7 +90,7 @@ func (r *resultStore) AddFetch(header *types.Header, snapSync bool) (stale, thro
 	}
 
 	if item == nil {
-		item = newFetchResult(header, snapSync)
+		item = newFetchResult(header, fastSync)
 		r.items[index] = item
 	}