diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go
index e30572e0a1..d2dd2798d7 100644
--- a/accounts/abi/bind/backends/simulated.go
+++ b/accounts/abi/bind/backends/simulated.go
@@ -70,7 +70,7 @@ type SimulatedBackend struct {
func NewSimulatedBackendWithDatabase(database ethdb.Database, alloc core.GenesisAlloc, gasLimit uint64) *SimulatedBackend {
genesis := core.Genesis{Config: params.AllEthashProtocolChanges, GasLimit: gasLimit, Alloc: alloc}
genesis.MustCommit(database)
- blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := core.NewBlockChain(database, nil, genesis.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
backend := &SimulatedBackend{
database: database,
diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go
index 49e6a05949..20fb5aa995 100644
--- a/cmd/geth/chaincmd.go
+++ b/cmd/geth/chaincmd.go
@@ -69,6 +69,7 @@ It expects the genesis file as argument.`,
utils.GCModeFlag,
utils.CacheDatabaseFlag,
utils.CacheGCFlag,
+ utils.TxLookupLimitFlag,
},
Category: "BLOCKCHAIN COMMANDS",
Description: `
@@ -136,6 +137,7 @@ The export-preimages command export hash preimages to an RLP encoded stream`,
utils.FakePoWFlag,
utils.TestnetFlag,
utils.RinkebyFlag,
+ utils.TxLookupLimitFlag,
},
Category: "BLOCKCHAIN COMMANDS",
Description: `
@@ -234,7 +236,7 @@ func importChain(ctx *cli.Context) error {
stack := makeFullNode(ctx)
defer stack.Close()
- chain, db := utils.MakeChain(ctx, stack)
+ chain, db := utils.MakeChain(ctx, stack, false)
defer db.Close()
// Start periodically gathering memory profiles
@@ -324,7 +326,7 @@ func exportChain(ctx *cli.Context) error {
stack := makeFullNode(ctx)
defer stack.Close()
- chain, _ := utils.MakeChain(ctx, stack)
+ chain, _ := utils.MakeChain(ctx, stack, true)
start := time.Now()
var err error
@@ -399,7 +401,7 @@ func copyDb(ctx *cli.Context) error {
stack := makeFullNode(ctx)
defer stack.Close()
- chain, chainDb := utils.MakeChain(ctx, stack)
+ chain, chainDb := utils.MakeChain(ctx, stack, false)
syncMode := *utils.GlobalTextMarshaler(ctx, utils.SyncModeFlag.Name).(*downloader.SyncMode)
var syncBloom *trie.SyncBloom
@@ -507,7 +509,7 @@ func dump(ctx *cli.Context) error {
stack := makeFullNode(ctx)
defer stack.Close()
- chain, chainDb := utils.MakeChain(ctx, stack)
+ chain, chainDb := utils.MakeChain(ctx, stack, true)
defer chainDb.Close()
for _, arg := range ctx.Args() {
var block *types.Block
@@ -546,7 +548,7 @@ func inspect(ctx *cli.Context) error {
node, _ := makeConfigNode(ctx)
defer node.Close()
- _, chainDb := utils.MakeChain(ctx, node)
+ _, chainDb := utils.MakeChain(ctx, node, true)
defer chainDb.Close()
return rawdb.InspectDatabase(chainDb)
diff --git a/cmd/geth/main.go b/cmd/geth/main.go
index 019eee605f..2a72f10f85 100644
--- a/cmd/geth/main.go
+++ b/cmd/geth/main.go
@@ -94,6 +94,7 @@ var (
utils.SyncModeFlag,
utils.ExitWhenSyncedFlag,
utils.GCModeFlag,
+ utils.TxLookupLimitFlag,
utils.LightServeFlag,
utils.LightLegacyServFlag,
utils.LightIngressFlag,
diff --git a/cmd/geth/retesteth.go b/cmd/geth/retesteth.go
index b6aa3706b2..7bf0aa9908 100644
--- a/cmd/geth/retesteth.go
+++ b/cmd/geth/retesteth.go
@@ -398,7 +398,7 @@ func (api *RetestethAPI) SetChainParams(ctx context.Context, chainParams ChainPa
}
engine := &NoRewardEngine{inner: inner, rewardsOn: chainParams.SealEngine != "NoReward"}
- blockchain, err := core.NewBlockChain(ethDb, nil, chainConfig, engine, vm.Config{}, nil)
+ blockchain, err := core.NewBlockChain(ethDb, nil, chainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
return false, err
}
diff --git a/cmd/geth/usage.go b/cmd/geth/usage.go
index af195425b1..d7aed0ce8c 100644
--- a/cmd/geth/usage.go
+++ b/cmd/geth/usage.go
@@ -80,6 +80,7 @@ var AppHelpFlagGroups = []flagGroup{
utils.SyncModeFlag,
utils.ExitWhenSyncedFlag,
utils.GCModeFlag,
+ utils.TxLookupLimitFlag,
utils.EthStatsURLFlag,
utils.IdentityFlag,
utils.LightKDFFlag,
diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
index 42424bcb52..9de8efc21e 100644
--- a/cmd/utils/flags.go
+++ b/cmd/utils/flags.go
@@ -215,6 +215,11 @@ var (
Usage: `Blockchain garbage collection mode ("full", "archive")`,
Value: "full",
}
+ TxLookupLimitFlag = cli.Int64Flag{
+ Name: "txlookuplimit",
+ Usage: "Number of recent blocks to index transactions-by-hash in (default = index all blocks)",
+ Value: 0,
+ }
LightKDFFlag = cli.BoolFlag{
Name: "lightkdf",
Usage: "Reduce key-derivation RAM & CPU usage at some expense of KDF strength",
@@ -1423,7 +1428,11 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
CheckExclusive(ctx, DeveloperFlag, TestnetFlag, RinkebyFlag, GoerliFlag)
CheckExclusive(ctx, LightLegacyServFlag, LightServeFlag, SyncModeFlag, "light")
CheckExclusive(ctx, DeveloperFlag, ExternalSignerFlag) // Can't use both ephemeral unlocked and external signer
-
+ CheckExclusive(ctx, GCModeFlag, "archive", TxLookupLimitFlag)
+ // todo(rjl493456442) make it available for les server
+ // Ancient tx indices pruning is not available for les server now
+ // since light client relies on the server for transaction status query.
+ CheckExclusive(ctx, LightLegacyServFlag, LightServeFlag, TxLookupLimitFlag)
var ks *keystore.KeyStore
if keystores := stack.AccountManager().Backends(keystore.KeyStoreType); len(keystores) > 0 {
ks = keystores[0].(*keystore.KeyStore)
@@ -1459,6 +1468,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
if ctx.GlobalIsSet(CacheNoPrefetchFlag.Name) {
cfg.NoPrefetch = ctx.GlobalBool(CacheNoPrefetchFlag.Name)
}
+ if ctx.GlobalIsSet(TxLookupLimitFlag.Name) {
+ cfg.TxLookupLimit = ctx.GlobalUint64(TxLookupLimitFlag.Name)
+ }
if ctx.GlobalIsSet(CacheFlag.Name) || ctx.GlobalIsSet(CacheTrieFlag.Name) {
cfg.TrieCleanCache = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheTrieFlag.Name) / 100
}
@@ -1684,7 +1696,7 @@ func MakeGenesis(ctx *cli.Context) *core.Genesis {
}
// MakeChain creates a chain manager from set command line flags.
-func MakeChain(ctx *cli.Context, stack *node.Node) (chain *core.BlockChain, chainDb ethdb.Database) {
+func MakeChain(ctx *cli.Context, stack *node.Node, readOnly bool) (chain *core.BlockChain, chainDb ethdb.Database) {
var err error
chainDb = MakeChainDatabase(ctx, stack)
config, _, err := core.SetupGenesisBlock(chainDb, MakeGenesis(ctx))
@@ -1724,7 +1736,12 @@ func MakeChain(ctx *cli.Context, stack *node.Node) (chain *core.BlockChain, chai
cache.TrieDirtyLimit = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheGCFlag.Name) / 100
}
vmcfg := vm.Config{EnablePreimageRecording: ctx.GlobalBool(VMEnableDebugFlag.Name)}
- chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil)
+ var limit *uint64
+ if ctx.GlobalIsSet(TxLookupLimitFlag.Name) && !readOnly {
+ l := ctx.GlobalUint64(TxLookupLimitFlag.Name)
+ limit = &l
+ }
+ chain, err = core.NewBlockChain(chainDb, cache, config, engine, vmcfg, nil, limit)
if err != nil {
Fatalf("Can't create BlockChain: %v", err)
}
diff --git a/consensus/clique/clique_test.go b/consensus/clique/clique_test.go
index 710f448055..49313374fc 100644
--- a/consensus/clique/clique_test.go
+++ b/consensus/clique/clique_test.go
@@ -54,7 +54,7 @@ func TestReimportMirroredState(t *testing.T) {
genesis := genspec.MustCommit(db)
// Generate a batch of blocks, each properly signed
- chain, _ := core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil)
+ chain, _ := core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, nil)
defer chain.Stop()
blocks, _ := core.GenerateChain(params.AllCliqueProtocolChanges, genesis, engine, db, 3, func(i int, block *core.BlockGen) {
@@ -88,7 +88,7 @@ func TestReimportMirroredState(t *testing.T) {
db = rawdb.NewMemoryDatabase()
genspec.MustCommit(db)
- chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil)
+ chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, nil)
defer chain.Stop()
if _, err := chain.InsertChain(blocks[:2]); err != nil {
@@ -101,7 +101,7 @@ func TestReimportMirroredState(t *testing.T) {
// Simulate a crash by creating a new chain on top of the database, without
// flushing the dirty states out. Insert the last block, trigerring a sidechain
// reimport.
- chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil)
+ chain, _ = core.NewBlockChain(db, nil, params.AllCliqueProtocolChanges, engine, vm.Config{}, nil, nil)
defer chain.Stop()
if _, err := chain.InsertChain(blocks[2:]); err != nil {
diff --git a/consensus/clique/snapshot_test.go b/consensus/clique/snapshot_test.go
index fc08722efd..3890fc51dd 100644
--- a/consensus/clique/snapshot_test.go
+++ b/consensus/clique/snapshot_test.go
@@ -448,7 +448,7 @@ func TestClique(t *testing.T) {
batches[len(batches)-1] = append(batches[len(batches)-1], block)
}
// Pass all the headers through clique and ensure tallying succeeds
- chain, err := core.NewBlockChain(db, nil, &config, engine, vm.Config{}, nil)
+ chain, err := core.NewBlockChain(db, nil, &config, engine, vm.Config{}, nil, nil)
if err != nil {
t.Errorf("test %d: failed to create test chain: %v", i, err)
continue
diff --git a/core/bench_test.go b/core/bench_test.go
index d7a5e11c2f..0f4cabd837 100644
--- a/core/bench_test.go
+++ b/core/bench_test.go
@@ -175,7 +175,7 @@ func benchInsertChain(b *testing.B, disk bool, gen func(int, *BlockGen)) {
// Time the insertion of the new chain.
// State and blocks are stored in the same DB.
- chainman, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ chainman, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer chainman.Stop()
b.ReportAllocs()
b.ResetTimer()
@@ -287,7 +287,7 @@ func benchReadChain(b *testing.B, full bool, count uint64) {
if err != nil {
b.Fatalf("error opening database at %v: %v", dir, err)
}
- chain, err := NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
+ chain, err := NewBlockChain(db, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
if err != nil {
b.Fatalf("error creating chain: %v", err)
}
diff --git a/core/block_validator_test.go b/core/block_validator_test.go
index 06e2ba1a4f..dfb37b88cf 100644
--- a/core/block_validator_test.go
+++ b/core/block_validator_test.go
@@ -42,7 +42,7 @@ func TestHeaderVerification(t *testing.T) {
headers[i] = block.Header()
}
// Run the header checker for blocks one-by-one, checking for both valid and invalid nonces
- chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
+ chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
defer chain.Stop()
for i := 0; i < len(blocks); i++ {
@@ -106,11 +106,11 @@ func testHeaderConcurrentVerification(t *testing.T, threads int) {
var results <-chan error
if valid {
- chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
+ chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
_, results = chain.engine.VerifyHeaders(chain, headers, seals)
chain.Stop()
} else {
- chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), vm.Config{}, nil)
+ chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeFailer(uint64(len(headers)-1)), vm.Config{}, nil, nil)
_, results = chain.engine.VerifyHeaders(chain, headers, seals)
chain.Stop()
}
@@ -173,7 +173,7 @@ func testHeaderConcurrentAbortion(t *testing.T, threads int) {
defer runtime.GOMAXPROCS(old)
// Start the verifications and immediately abort
- chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), vm.Config{}, nil)
+ chain, _ := NewBlockChain(testdb, nil, params.TestChainConfig, ethash.NewFakeDelayer(time.Millisecond), vm.Config{}, nil, nil)
defer chain.Stop()
abort, results := chain.engine.VerifyHeaders(chain, headers, seals)
diff --git a/core/blockchain.go b/core/blockchain.go
index 9fb02b1482..6fd5adf066 100644
--- a/core/blockchain.go
+++ b/core/blockchain.go
@@ -138,6 +138,13 @@ type BlockChain struct {
triegc *prque.Prque // Priority queue mapping block numbers to tries to gc
gcproc time.Duration // Accumulates canonical block processing for trie dumping
+ // txLookupLimit is the maximum number of blocks from head whose tx indices
+ // are reserved:
+ // * 0: means no limit and regenerate any missing indexes
+ // * N: means N block limit [HEAD-N+1, HEAD] and delete extra indexes
+ // * nil: disable tx reindexer/deleter, but still index new blocks
+ txLookupLimit uint64
+
hc *HeaderChain
rmLogsFeed event.Feed
chainFeed event.Feed
@@ -181,7 +188,7 @@ type BlockChain struct {
// NewBlockChain returns a fully initialised block chain using information
// available in the database. It initialises the default Ethereum Validator and
// Processor.
-func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool) (*BlockChain, error) {
+func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *params.ChainConfig, engine consensus.Engine, vmConfig vm.Config, shouldPreserve func(block *types.Block) bool, txLookupLimit *uint64) (*BlockChain, error) {
if cacheConfig == nil {
cacheConfig = &CacheConfig{
TrieCleanLimit: 256,
@@ -234,8 +241,16 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
bc.currentFastBlock.Store(nilBlock)
// Initialize the chain with ancient data if it isn't empty.
+ var txIndexBlock uint64
+
if bc.empty() {
rawdb.InitDatabaseFromFreezer(bc.db)
+ // If ancient database is not empty, reconstruct all missing
+ // indices in the background.
+ frozen, _ := bc.db.Ancients()
+ if frozen > 0 {
+ txIndexBlock = frozen
+ }
}
if err := bc.loadLastState(); err != nil {
@@ -294,6 +309,10 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
}
// Take ownership of this particular state
go bc.update()
+ if txLookupLimit != nil {
+ bc.txLookupLimit = *txLookupLimit
+ go bc.maintainTxIndex(txIndexBlock)
+ }
return bc, nil
}
@@ -1093,8 +1112,23 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
}
// Flush data into ancient database.
size += rawdb.WriteAncientBlock(bc.db, block, receiptChain[i], bc.GetTd(block.Hash(), block.NumberU64()))
- rawdb.WriteTxLookupEntries(batch, block)
+ // Write tx indices if any condition is satisfied:
+ // * If user requires to reserve all tx indices(txlookuplimit=0)
+ // * If all ancient tx indices are required to be reserved(txlookuplimit is even higher than ancientlimit)
+ // * If block number is large enough to be regarded as a recent block
+ // It means blocks below the ancientLimit-txlookupLimit won't be indexed.
+ //
+ // But if the `TxIndexTail` is not nil, e.g. Geth is initialized with
+ // an external ancient database, during the setup, blockchain will start
+ // a background routine to re-indexed all indices in [ancients - txlookupLimit, ancients)
+ // range. In this case, all tx indices of newly imported blocks should be
+ // generated.
+ if bc.txLookupLimit == 0 || ancientLimit <= bc.txLookupLimit || block.NumberU64() >= ancientLimit-bc.txLookupLimit {
+ rawdb.WriteTxLookupEntries(batch, block)
+ } else if rawdb.ReadTxIndexTail(bc.db) != nil {
+ rawdb.WriteTxLookupEntries(batch, block)
+ }
stats.processed++
}
// Flush all tx-lookup index data.
@@ -1168,7 +1202,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
// Write all the data out into the database
rawdb.WriteBody(batch, block.Hash(), block.NumberU64(), block.Body())
rawdb.WriteReceipts(batch, block.Hash(), block.NumberU64(), receiptChain[i])
- rawdb.WriteTxLookupEntries(batch, block)
+ rawdb.WriteTxLookupEntries(batch, block) // Always write tx indices for live blocks, we assume they are needed
stats.processed++
if batch.ValueSize() >= ethdb.IdealBatchSize {
@@ -1188,7 +1222,7 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
updateHead(blockChain[len(blockChain)-1])
return 0, nil
}
- // Write downloaded chain data and corresponding receipt chain data.
+ // Write downloaded chain data and corresponding receipt chain data
if len(ancientBlocks) > 0 {
if n, err := writeAncient(ancientBlocks, ancientReceipts); err != nil {
if err == errInsertionInterrupted {
@@ -1197,6 +1231,19 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
return n, err
}
}
+ // Write the tx index tail (block number from where we index) before write any live blocks
+ if len(liveBlocks) > 0 && liveBlocks[0].NumberU64() == ancientLimit+1 {
+ // The tx index tail can only be one of the following two options:
+ // * 0: all ancient blocks have been indexed
+ // * ancient-limit: the indices of blocks before ancient-limit are ignored
+ if tail := rawdb.ReadTxIndexTail(bc.db); tail == nil {
+ if bc.txLookupLimit == 0 || ancientLimit <= bc.txLookupLimit {
+ rawdb.WriteTxIndexTail(bc.db, 0)
+ } else {
+ rawdb.WriteTxIndexTail(bc.db, ancientLimit-bc.txLookupLimit)
+ }
+ }
+ }
if len(liveBlocks) > 0 {
if n, err := writeLive(liveBlocks, liveReceipts); err != nil {
if err == errInsertionInterrupted {
@@ -1220,6 +1267,18 @@ func (bc *BlockChain) InsertReceiptChain(blockChain types.Blocks, receiptChain [
return 0, nil
}
+// SetTxLookupLimit is responsible for updating the txlookup limit to the
+// original one stored in db if the new mismatches with the old one.
+func (bc *BlockChain) SetTxLookupLimit(limit uint64) {
+ bc.txLookupLimit = limit
+}
+
+// TxLookupLimit retrieves the txlookup limit used by blockchain to prune
+// stale transaction indices.
+func (bc *BlockChain) TxLookupLimit() uint64 {
+ return bc.txLookupLimit
+}
+
var lastWrite uint64
// writeBlockWithoutState writes only the block and its metadata to the database,
@@ -2042,6 +2101,86 @@ func (bc *BlockChain) update() {
}
}
+// maintainTxIndex is responsible for the construction and deletion of the
+// transaction index.
+//
+// User can use flag `txlookuplimit` to specify a "recentness" block, below
+// which ancient tx indices get deleted. If `txlookuplimit` is 0, it means
+// all tx indices will be reserved.
+//
+// The user can adjust the txlookuplimit value for each launch after fast
+// sync, Geth will automatically construct the missing indices and delete
+// the extra indices.
+func (bc *BlockChain) maintainTxIndex(ancients uint64) {
+ // Before starting the actual maintenance, we need to handle a special case,
+ // where user might init Geth with an external ancient database. If so, we
+ // need to reindex all necessary transactions before starting to process any
+ // pruning requests.
+ if ancients > 0 {
+ var from = uint64(0)
+ if bc.txLookupLimit != 0 && ancients > bc.txLookupLimit {
+ from = ancients - bc.txLookupLimit
+ }
+ rawdb.IndexTransactions(bc.db, from, ancients)
+ }
+ // indexBlocks reindexes or unindexes transactions depending on user configuration
+ indexBlocks := func(tail *uint64, head uint64, done chan struct{}) {
+ defer func() { done <- struct{}{} }()
+
+ // If the user just upgraded Geth to a new version which supports transaction
+ // index pruning, write the new tail and remove anything older.
+ if tail == nil {
+ if bc.txLookupLimit == 0 || head < bc.txLookupLimit {
+ // Nothing to delete, write the tail and return
+ rawdb.WriteTxIndexTail(bc.db, 0)
+ } else {
+ // Prune all stale tx indices and record the tx index tail
+ rawdb.UnindexTransactions(bc.db, 0, head-bc.txLookupLimit+1)
+ }
+ return
+ }
+ // If a previous indexing existed, make sure that we fill in any missing entries
+ if bc.txLookupLimit == 0 || head < bc.txLookupLimit {
+ if *tail > 0 {
+ rawdb.IndexTransactions(bc.db, 0, *tail)
+ }
+ return
+ }
+ // Update the transaction index to the new chain state
+ if head-bc.txLookupLimit+1 < *tail {
+ // Reindex a part of missing indices and rewind index tail to HEAD-limit
+ rawdb.IndexTransactions(bc.db, head-bc.txLookupLimit+1, *tail)
+ } else {
+ // Unindex a part of stale indices and forward index tail to HEAD-limit
+ rawdb.UnindexTransactions(bc.db, *tail, head-bc.txLookupLimit+1)
+ }
+ }
+ // Any reindexing done, start listening to chain events and moving the index window
+ var (
+ done chan struct{} // Non-nil if background unindexing or reindexing routine is active.
+ headCh = make(chan ChainHeadEvent, 1) // Buffered to avoid locking up the event feed
+ )
+ sub := bc.SubscribeChainHeadEvent(headCh)
+ if sub == nil {
+ return
+ }
+ defer sub.Unsubscribe()
+
+ for {
+ select {
+ case head := <-headCh:
+ if done == nil {
+ done = make(chan struct{})
+ go indexBlocks(rawdb.ReadTxIndexTail(bc.db), head.Block.NumberU64(), done)
+ }
+ case <-done:
+ done = nil
+ case <-bc.quit:
+ return
+ }
+ }
+}
+
// BadBlocks returns a list of the last 'bad blocks' that the client has seen on the network
func (bc *BlockChain) BadBlocks() []*types.Block {
blocks := make([]*types.Block, 0, bc.badBlocks.Len())
diff --git a/core/blockchain_test.go b/core/blockchain_test.go
index 34f68a6443..a6efc1d840 100644
--- a/core/blockchain_test.go
+++ b/core/blockchain_test.go
@@ -54,7 +54,7 @@ func newCanonical(engine consensus.Engine, n int, full bool) (ethdb.Database, *B
)
// Initialize a fresh chain with only a genesis block
- blockchain, _ := NewBlockChain(db, nil, params.AllEthashProtocolChanges, engine, vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, params.AllEthashProtocolChanges, engine, vm.Config{}, nil, nil)
// Create and inject the requested chain
if n == 0 {
return db, blockchain, nil
@@ -509,7 +509,7 @@ func testReorgBadHashes(t *testing.T, full bool) {
blockchain.Stop()
// Create a new BlockChain and check that it rolled back the state.
- ncm, err := NewBlockChain(blockchain.db, nil, blockchain.chainConfig, ethash.NewFaker(), vm.Config{}, nil)
+ ncm, err := NewBlockChain(blockchain.db, nil, blockchain.chainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create new chain manager: %v", err)
}
@@ -621,7 +621,7 @@ func TestFastVsFullChains(t *testing.T) {
// Import the chain as an archive node for the comparison baseline
archiveDb := rawdb.NewMemoryDatabase()
gspec.MustCommit(archiveDb)
- archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer archive.Stop()
if n, err := archive.InsertChain(blocks); err != nil {
@@ -630,7 +630,7 @@ func TestFastVsFullChains(t *testing.T) {
// Fast import the chain as a non-archive node to test
fastDb := rawdb.NewMemoryDatabase()
gspec.MustCommit(fastDb)
- fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer fast.Stop()
headers := make([]*types.Header, len(blocks))
@@ -654,7 +654,7 @@ func TestFastVsFullChains(t *testing.T) {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
- ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer ancient.Stop()
if n, err := ancient.InsertHeaderChain(headers, 1); err != nil {
@@ -750,7 +750,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as an archive node and ensure all pointers are updated
archiveDb, delfn := makeDb()
defer delfn()
- archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ archive, _ := NewBlockChain(archiveDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
if n, err := archive.InsertChain(blocks); err != nil {
t.Fatalf("failed to process block %d: %v", n, err)
}
@@ -763,7 +763,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as a non-archive node and ensure all pointers are updated
fastDb, delfn := makeDb()
defer delfn()
- fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ fast, _ := NewBlockChain(fastDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer fast.Stop()
headers := make([]*types.Header, len(blocks))
@@ -783,7 +783,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as a ancient-first node and ensure all pointers are updated
ancientDb, delfn := makeDb()
defer delfn()
- ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer ancient.Stop()
if n, err := ancient.InsertHeaderChain(headers, 1); err != nil {
@@ -802,7 +802,7 @@ func TestLightVsFastVsFullChainHeads(t *testing.T) {
// Import the chain as a light node and ensure all pointers are updated
lightDb, delfn := makeDb()
defer delfn()
- light, _ := NewBlockChain(lightDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ light, _ := NewBlockChain(lightDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
if n, err := light.InsertHeaderChain(headers, 1); err != nil {
t.Fatalf("failed to insert header %d: %v", n, err)
}
@@ -871,7 +871,7 @@ func TestChainTxReorgs(t *testing.T) {
}
})
// Import the chain. This runs all block validation rules.
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
if i, err := blockchain.InsertChain(chain); err != nil {
t.Fatalf("failed to insert original chain[%d]: %v", i, err)
}
@@ -941,7 +941,7 @@ func TestLogReorgs(t *testing.T) {
signer = types.NewEIP155Signer(gspec.Config.ChainID)
)
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
rmLogsCh := make(chan RemovedLogsEvent)
@@ -1018,7 +1018,7 @@ func TestLogRebirth(t *testing.T) {
}
}
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
logsCh := make(chan []*types.Log)
@@ -1140,7 +1140,7 @@ func TestSideLogRebirth(t *testing.T) {
}
}
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
logsCh := make(chan []*types.Log)
@@ -1195,7 +1195,7 @@ func TestReorgSideEvent(t *testing.T) {
signer = types.NewEIP155Signer(gspec.Config.ChainID)
)
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
chain, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, gen *BlockGen) {})
@@ -1324,7 +1324,7 @@ func TestEIP155Transition(t *testing.T) {
genesis = gspec.MustCommit(db)
)
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
blocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 4, func(i int, block *BlockGen) {
@@ -1432,7 +1432,7 @@ func TestEIP161AccountRemoval(t *testing.T) {
}
genesis = gspec.MustCommit(db)
)
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
blocks, _ := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, 3, func(i int, block *BlockGen) {
@@ -1507,7 +1507,7 @@ func TestBlockchainHeaderchainReorgConsistency(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@@ -1551,7 +1551,7 @@ func TestTrieForkGC(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@@ -1590,7 +1590,7 @@ func TestLargeReorgTrieGC(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@@ -1650,7 +1650,7 @@ func TestBlockchainRecovery(t *testing.T) {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
- ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
headers := make([]*types.Header, len(blocks))
for i, block := range blocks {
@@ -1669,7 +1669,7 @@ func TestBlockchainRecovery(t *testing.T) {
rawdb.WriteHeadFastBlockHash(ancientDb, midBlock.Hash())
// Reopen broken blockchain again
- ancient, _ = NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ ancient, _ = NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer ancient.Stop()
if num := ancient.CurrentBlock().NumberU64(); num != 0 {
t.Errorf("head block mismatch: have #%v, want #%v", num, 0)
@@ -1706,7 +1706,7 @@ func TestIncompleteAncientReceiptChainInsertion(t *testing.T) {
t.Fatalf("failed to create temp freezer db: %v", err)
}
gspec.MustCommit(ancientDb)
- ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ ancient, _ := NewBlockChain(ancientDb, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer ancient.Stop()
headers := make([]*types.Header, len(blocks))
@@ -1763,7 +1763,7 @@ func TestLowDiffLongChain(t *testing.T) {
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@@ -1810,7 +1810,7 @@ func testSideImport(t *testing.T, numCanonBlocksInSidechain, blocksBetweenCommon
blocks, _ := GenerateChain(params.TestChainConfig, genesis, engine, db, 2*TriesInMemory, nil)
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@@ -1907,7 +1907,7 @@ func testInsertKnownChainData(t *testing.T, typ string) {
new(Genesis).MustCommit(chaindb)
defer os.RemoveAll(dir)
- chain, err := NewBlockChain(chaindb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(chaindb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
@@ -2030,7 +2030,7 @@ func getLongAndShortChains() (*BlockChain, []*types.Block, []*types.Block, error
diskdb := rawdb.NewMemoryDatabase()
new(Genesis).MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
return nil, nil, nil, fmt.Errorf("failed to create tester chain: %v", err)
}
@@ -2134,6 +2134,219 @@ func TestReorgToShorterRemovesCanonMappingHeaderChain(t *testing.T) {
}
}
+func TestTransactionIndices(t *testing.T) {
+ // Configure and generate a sample block chain
+ var (
+ gendb = rawdb.NewMemoryDatabase()
+ key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+ address = crypto.PubkeyToAddress(key.PublicKey)
+ funds = big.NewInt(1000000000)
+ gspec = &Genesis{Config: params.TestChainConfig, Alloc: GenesisAlloc{address: {Balance: funds}}}
+ genesis = gspec.MustCommit(gendb)
+ signer = types.NewEIP155Signer(gspec.Config.ChainID)
+ )
+ height := uint64(128)
+ blocks, receipts := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), gendb, int(height), func(i int, block *BlockGen) {
+ tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, nil, nil), signer, key)
+ if err != nil {
+ panic(err)
+ }
+ block.AddTx(tx)
+ })
+ blocks2, _ := GenerateChain(gspec.Config, blocks[len(blocks)-1], ethash.NewFaker(), gendb, 10, nil)
+
+ check := func(tail *uint64, chain *BlockChain) {
+ stored := rawdb.ReadTxIndexTail(chain.db)
+ if tail == nil && stored != nil {
+ t.Fatalf("Oldest indexded block mismatch, want nil, have %d", *stored)
+ }
+ if tail != nil && *stored != *tail {
+ t.Fatalf("Oldest indexded block mismatch, want %d, have %d", *tail, *stored)
+ }
+ if tail != nil {
+ for i := *tail; i <= chain.CurrentBlock().NumberU64(); i++ {
+ block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i)
+ if block.Transactions().Len() == 0 {
+ continue
+ }
+ for _, tx := range block.Transactions() {
+ if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index == nil {
+ t.Fatalf("Miss transaction indice, number %d hash %s", i, tx.Hash().Hex())
+ }
+ }
+ }
+ for i := uint64(0); i < *tail; i++ {
+ block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i)
+ if block.Transactions().Len() == 0 {
+ continue
+ }
+ for _, tx := range block.Transactions() {
+ if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index != nil {
+ t.Fatalf("Transaction indice should be deleted, number %d hash %s", i, tx.Hash().Hex())
+ }
+ }
+ }
+ }
+ }
+ frdir, err := ioutil.TempDir("", "")
+ if err != nil {
+ t.Fatalf("failed to create temp freezer dir: %v", err)
+ }
+ defer os.Remove(frdir)
+ ancientDb, err := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "")
+ if err != nil {
+ t.Fatalf("failed to create temp freezer db: %v", err)
+ }
+ gspec.MustCommit(ancientDb)
+
+ // Import all blocks into ancient db
+ l := uint64(0)
+ chain, err := NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, &l)
+ if err != nil {
+ t.Fatalf("failed to create tester chain: %v", err)
+ }
+ headers := make([]*types.Header, len(blocks))
+ for i, block := range blocks {
+ headers[i] = block.Header()
+ }
+ if n, err := chain.InsertHeaderChain(headers, 0); err != nil {
+ t.Fatalf("failed to insert header %d: %v", n, err)
+ }
+ if n, err := chain.InsertReceiptChain(blocks, receipts, 128); err != nil {
+ t.Fatalf("block %d: failed to insert into chain: %v", n, err)
+ }
+ chain.Stop()
+ ancientDb.Close()
+
+ // Init block chain with external ancients, check all needed indices has been indexed.
+ limit := []uint64{0, 32, 64, 128}
+ for _, l := range limit {
+ ancientDb, err = rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "")
+ if err != nil {
+ t.Fatalf("failed to create temp freezer db: %v", err)
+ }
+ gspec.MustCommit(ancientDb)
+ chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, &l)
+ if err != nil {
+ t.Fatalf("failed to create tester chain: %v", err)
+ }
+ time.Sleep(50 * time.Millisecond) // Wait for indices initialisation
+ var tail uint64
+ if l != 0 {
+ tail = uint64(128) - l + 1
+ }
+ check(&tail, chain)
+ chain.Stop()
+ ancientDb.Close()
+ }
+
+ // Reconstruct a block chain which only reserves HEAD-64 tx indices
+ ancientDb, err = rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "")
+ if err != nil {
+ t.Fatalf("failed to create temp freezer db: %v", err)
+ }
+ gspec.MustCommit(ancientDb)
+
+ limit = []uint64{0, 64 /* drop stale */, 32 /* shorten history */, 64 /* extend history */, 0 /* restore all */}
+ tails := []uint64{0, 67 /* 130 - 64 + 1 */, 100 /* 131 - 32 + 1 */, 69 /* 132 - 64 + 1 */, 0}
+ for i, l := range limit {
+ chain, err = NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, &l)
+ if err != nil {
+ t.Fatalf("failed to create tester chain: %v", err)
+ }
+ chain.InsertChain(blocks2[i : i+1]) // Feed chain a higher block to trigger indices updater.
+ time.Sleep(50 * time.Millisecond) // Wait for indices initialisation
+ check(&tails[i], chain)
+ chain.Stop()
+ }
+}
+
+func TestSkipStaleTxIndicesInFastSync(t *testing.T) {
+ // Configure and generate a sample block chain
+ var (
+ gendb = rawdb.NewMemoryDatabase()
+ key, _ = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+ address = crypto.PubkeyToAddress(key.PublicKey)
+ funds = big.NewInt(1000000000)
+ gspec = &Genesis{Config: params.TestChainConfig, Alloc: GenesisAlloc{address: {Balance: funds}}}
+ genesis = gspec.MustCommit(gendb)
+ signer = types.NewEIP155Signer(gspec.Config.ChainID)
+ )
+ height := uint64(128)
+ blocks, receipts := GenerateChain(gspec.Config, genesis, ethash.NewFaker(), gendb, int(height), func(i int, block *BlockGen) {
+ tx, err := types.SignTx(types.NewTransaction(block.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, nil, nil), signer, key)
+ if err != nil {
+ panic(err)
+ }
+ block.AddTx(tx)
+ })
+
+ check := func(tail *uint64, chain *BlockChain) {
+ stored := rawdb.ReadTxIndexTail(chain.db)
+ if tail == nil && stored != nil {
+ t.Fatalf("Oldest indexded block mismatch, want nil, have %d", *stored)
+ }
+ if tail != nil && *stored != *tail {
+ t.Fatalf("Oldest indexded block mismatch, want %d, have %d", *tail, *stored)
+ }
+ if tail != nil {
+ for i := *tail; i <= chain.CurrentBlock().NumberU64(); i++ {
+ block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i)
+ if block.Transactions().Len() == 0 {
+ continue
+ }
+ for _, tx := range block.Transactions() {
+ if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index == nil {
+ t.Fatalf("Miss transaction indice, number %d hash %s", i, tx.Hash().Hex())
+ }
+ }
+ }
+ for i := uint64(0); i < *tail; i++ {
+ block := rawdb.ReadBlock(chain.db, rawdb.ReadCanonicalHash(chain.db, i), i)
+ if block.Transactions().Len() == 0 {
+ continue
+ }
+ for _, tx := range block.Transactions() {
+ if index := rawdb.ReadTxLookupEntry(chain.db, tx.Hash()); index != nil {
+ t.Fatalf("Transaction indice should be deleted, number %d hash %s", i, tx.Hash().Hex())
+ }
+ }
+ }
+ }
+ }
+
+ frdir, err := ioutil.TempDir("", "")
+ if err != nil {
+ t.Fatalf("failed to create temp freezer dir: %v", err)
+ }
+ defer os.Remove(frdir)
+ ancientDb, err := rawdb.NewDatabaseWithFreezer(rawdb.NewMemoryDatabase(), frdir, "")
+ if err != nil {
+ t.Fatalf("failed to create temp freezer db: %v", err)
+ }
+ gspec.MustCommit(ancientDb)
+
+ // Import all blocks into ancient db, only HEAD-32 indices are kept.
+ l := uint64(32)
+ chain, err := NewBlockChain(ancientDb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, &l)
+ if err != nil {
+ t.Fatalf("failed to create tester chain: %v", err)
+ }
+ headers := make([]*types.Header, len(blocks))
+ for i, block := range blocks {
+ headers[i] = block.Header()
+ }
+ if n, err := chain.InsertHeaderChain(headers, 0); err != nil {
+ t.Fatalf("failed to insert header %d: %v", n, err)
+ }
+ // The indices before ancient-N(32) should be ignored. After that all blocks should be indexed.
+ if n, err := chain.InsertReceiptChain(blocks, receipts, 64); err != nil {
+ t.Fatalf("block %d: failed to insert into chain: %v", n, err)
+ }
+ tail := uint64(32)
+ check(&tail, chain)
+}
+
// Benchmarks large blocks with value transfers to non-existing accounts
func benchmarkLargeNumberOfValueToNonexisting(b *testing.B, numTxs, numBlocks int, recipientFn func(uint64) common.Address, dataFn func(uint64) []byte) {
var (
@@ -2179,7 +2392,7 @@ func benchmarkLargeNumberOfValueToNonexisting(b *testing.B, numTxs, numBlocks in
diskdb := rawdb.NewMemoryDatabase()
gspec.MustCommit(diskdb)
- chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil)
+ chain, err := NewBlockChain(diskdb, nil, params.TestChainConfig, engine, vm.Config{}, nil, nil)
if err != nil {
b.Fatalf("failed to create tester chain: %v", err)
}
diff --git a/core/chain_makers_test.go b/core/chain_makers_test.go
index 32e3888d55..85a029f7c7 100644
--- a/core/chain_makers_test.go
+++ b/core/chain_makers_test.go
@@ -79,7 +79,7 @@ func ExampleGenerateChain() {
})
// Import the chain. This runs all block validation rules.
- blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, _ := NewBlockChain(db, nil, gspec.Config, ethash.NewFaker(), vm.Config{}, nil, nil)
defer blockchain.Stop()
if i, err := blockchain.InsertChain(chain); err != nil {
diff --git a/core/dao_test.go b/core/dao_test.go
index 4e8dba9e84..89e1d83d7a 100644
--- a/core/dao_test.go
+++ b/core/dao_test.go
@@ -45,7 +45,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
proConf.DAOForkBlock = forkBlock
proConf.DAOForkSupport = true
- proBc, _ := NewBlockChain(proDb, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil)
+ proBc, _ := NewBlockChain(proDb, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, nil)
defer proBc.Stop()
conDb := rawdb.NewMemoryDatabase()
@@ -55,7 +55,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
conConf.DAOForkBlock = forkBlock
conConf.DAOForkSupport = false
- conBc, _ := NewBlockChain(conDb, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
+ conBc, _ := NewBlockChain(conDb, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, nil)
defer conBc.Stop()
if _, err := proBc.InsertChain(prefix); err != nil {
@@ -69,7 +69,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Create a pro-fork block, and try to feed into the no-fork chain
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
- bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
+ bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, nil)
defer bc.Stop()
blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64()))
@@ -94,7 +94,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Create a no-fork block, and try to feed into the pro-fork chain
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
- bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil)
+ bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, nil)
defer bc.Stop()
blocks = proBc.GetBlocksFromHash(proBc.CurrentBlock().Hash(), int(proBc.CurrentBlock().NumberU64()))
@@ -120,7 +120,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Verify that contra-forkers accept pro-fork extra-datas after forking finishes
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
- bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil)
+ bc, _ := NewBlockChain(db, nil, &conConf, ethash.NewFaker(), vm.Config{}, nil, nil)
defer bc.Stop()
blocks := conBc.GetBlocksFromHash(conBc.CurrentBlock().Hash(), int(conBc.CurrentBlock().NumberU64()))
@@ -140,7 +140,7 @@ func TestDAOForkRangeExtradata(t *testing.T) {
// Verify that pro-forkers accept contra-fork extra-datas after forking finishes
db = rawdb.NewMemoryDatabase()
gspec.MustCommit(db)
- bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil)
+ bc, _ = NewBlockChain(db, nil, &proConf, ethash.NewFaker(), vm.Config{}, nil, nil)
defer bc.Stop()
blocks = proBc.GetBlocksFromHash(proBc.CurrentBlock().Hash(), int(proBc.CurrentBlock().NumberU64()))
diff --git a/core/genesis_test.go b/core/genesis_test.go
index c6bcd0aa54..fdae0b620a 100644
--- a/core/genesis_test.go
+++ b/core/genesis_test.go
@@ -120,7 +120,7 @@ func TestSetupGenesis(t *testing.T) {
// Advance to block #4, past the homestead transition block of customg.
genesis := oldcustomg.MustCommit(db)
- bc, _ := NewBlockChain(db, nil, oldcustomg.Config, ethash.NewFullFaker(), vm.Config{}, nil)
+ bc, _ := NewBlockChain(db, nil, oldcustomg.Config, ethash.NewFullFaker(), vm.Config{}, nil, nil)
defer bc.Stop()
blocks, _ := GenerateChain(oldcustomg.Config, genesis, ethash.NewFaker(), db, 4, nil)
diff --git a/core/rawdb/accessors_chain.go b/core/rawdb/accessors_chain.go
index fab7ca56c3..3564605ee5 100644
--- a/core/rawdb/accessors_chain.go
+++ b/core/rawdb/accessors_chain.go
@@ -171,6 +171,43 @@ func WriteFastTrieProgress(db ethdb.KeyValueWriter, count uint64) {
}
}
+// ReadTxIndexTail retrieves the number of oldest indexed block
+// whose transaction indices has been indexed. If the corresponding entry
+// is non-existent in database it means the indexing has been finished.
+func ReadTxIndexTail(db ethdb.KeyValueReader) *uint64 {
+ data, _ := db.Get(txIndexTailKey)
+ if len(data) != 8 {
+ return nil
+ }
+ number := binary.BigEndian.Uint64(data)
+ return &number
+}
+
+// WriteTxIndexTail stores the number of oldest indexed block
+// into database.
+func WriteTxIndexTail(db ethdb.KeyValueWriter, number uint64) {
+ if err := db.Put(txIndexTailKey, encodeBlockNumber(number)); err != nil {
+ log.Crit("Failed to store the transaction index tail", "err", err)
+ }
+}
+
+// ReadFastTxLookupLimit retrieves the tx lookup limit used in fast sync.
+func ReadFastTxLookupLimit(db ethdb.KeyValueReader) *uint64 {
+ data, _ := db.Get(fastTxLookupLimitKey)
+ if len(data) != 8 {
+ return nil
+ }
+ number := binary.BigEndian.Uint64(data)
+ return &number
+}
+
+// WriteFastTxLookupLimit stores the txlookup limit used in fast sync into database.
+func WriteFastTxLookupLimit(db ethdb.KeyValueWriter, number uint64) {
+ if err := db.Put(fastTxLookupLimitKey, encodeBlockNumber(number)); err != nil {
+ log.Crit("Failed to store transaction lookup limit for fast sync", "err", err)
+ }
+}
+
// ReadHeaderRLP retrieves a block header in its raw RLP database encoding.
func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
data, _ := db.Ancient(freezerHeaderTable, number)
diff --git a/core/rawdb/accessors_indexes.go b/core/rawdb/accessors_indexes.go
index 38f8fe10ea..6ae97a951b 100644
--- a/core/rawdb/accessors_indexes.go
+++ b/core/rawdb/accessors_indexes.go
@@ -65,7 +65,18 @@ func WriteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
// DeleteTxLookupEntry removes all transaction data associated with a hash.
func DeleteTxLookupEntry(db ethdb.KeyValueWriter, hash common.Hash) {
- db.Delete(txLookupKey(hash))
+ if err := db.Delete(txLookupKey(hash)); err != nil {
+ log.Crit("Failed to delete transaction lookup entry", "err", err)
+ }
+}
+
+// DeleteTxLookupEntries removes all transaction lookups for a given block.
+func DeleteTxLookupEntries(db ethdb.KeyValueWriter, block *types.Block) {
+ for _, tx := range block.Transactions() {
+ if err := db.Delete(txLookupKey(tx.Hash())); err != nil {
+ log.Crit("Failed to delete transaction lookup entry", "err", err)
+ }
+ }
}
// ReadTransaction retrieves a specific transaction from the database, along with
diff --git a/core/rawdb/chain_iterator.go b/core/rawdb/chain_iterator.go
new file mode 100644
index 0000000000..bb238c8d78
--- /dev/null
+++ b/core/rawdb/chain_iterator.go
@@ -0,0 +1,222 @@
+// Copyright 2019 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package rawdb
+
+import (
+ "errors"
+ "math"
+ "runtime"
+ "sync/atomic"
+ "time"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/common/prque"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/ethdb"
+ "github.com/ethereum/go-ethereum/log"
+)
+
+type (
+ prepareCallback func(*types.Block) // Callback for custom concurrent pre-computations on block data
+ actionCallback func(ethdb.Batch, *types.Block) // Callback for custom sequential operations on block data.
+)
+
+// iterateCanonicalChain iterates the specified range of canonical blocks and applies
+// the given action callback. Note, both for forward and backward iteration, the range
+// is [from, to).
+func iterateCanonicalChain(db ethdb.Database, from uint64, to uint64, prepare prepareCallback, action actionCallback, reverse bool, progMsg, doneMsg string) error {
+ // Short circuit if the action is nil or if the range is invalid
+ if action == nil {
+ return nil
+ }
+ if from >= to {
+ return nil
+ }
+ // Spawn multi-routines, iterate over the specified blocks and invoke prepare
+ // callback concurrently.
+ var (
+ number int64
+ results = make(chan *types.Block, 4*runtime.NumCPU())
+ )
+ if !reverse {
+ number = int64(from - 1)
+ } else {
+ number = int64(to)
+ }
+ abort := make(chan struct{})
+ defer close(abort)
+
+ threads := to - from
+ if cpus := runtime.NumCPU(); threads > uint64(cpus) {
+ threads = uint64(cpus)
+ }
+ for i := 0; i < int(threads); i++ {
+ go func() {
+ for {
+ // Fetch the next task number, terminating if everything's done
+ var n int64
+ if !reverse {
+ n = atomic.AddInt64(&number, 1)
+ if n >= int64(to) {
+ return
+ }
+ } else {
+ n = atomic.AddInt64(&number, -1)
+ if n < int64(from) {
+ return
+ }
+ }
+ block := ReadBlock(db, ReadCanonicalHash(db, uint64(n)), uint64(n))
+ if prepare != nil && block != nil {
+ prepare(block)
+ }
+ // Feed the block to the aggregator, or abort on interrupt
+ select {
+ case results <- block:
+ case <-abort:
+ return
+ }
+ }
+ }()
+ }
+ // Reassemble the blocks into a contiguous stream and apply the action callback
+ var (
+ next, first, last int64
+ queue = prque.New(nil)
+
+ batch = db.NewBatch()
+ start = time.Now()
+ logged time.Time
+ )
+ if !reverse {
+ next, first, last = int64(from), int64(from), int64(to)
+ } else {
+ next, first, last = int64(to-1), int64(to-1), int64(from-1)
+ }
+ for i := from; i < to; i++ {
+ // Retrieve the next result and bail if it's nil
+ block := <-results
+ if block == nil {
+ return errors.New("broken database")
+ }
+ // Push the block into the import queue and process contiguous ranges
+ priority := -int64(block.NumberU64())
+ if reverse {
+ priority = int64(block.NumberU64())
+ }
+ queue.Push(block, priority)
+ for !queue.Empty() {
+ // If the next available item is gapped, return
+ if _, priority := queue.Peek(); !reverse && -priority != next || reverse && priority != next {
+ break
+ }
+ // Next block available, pop it off and index it
+ block = queue.PopItem().(*types.Block)
+
+ if !reverse {
+ next++
+ } else {
+ next--
+ }
+ // Invoke action to inject specified data into key-value database.
+ action(batch, block)
+
+ // If enough data was accumulated in memory or we're at the last block, dump to disk
+ if batch.ValueSize() > ethdb.IdealBatchSize || next == last {
+ if err := batch.Write(); err != nil {
+ return err
+ }
+ batch.Reset()
+ }
+ // If we've spent too much time already, notify the user of what we're doing
+ if time.Since(logged) > 8*time.Second {
+ log.Info(progMsg, "blocks", int64(math.Abs(float64(next-first))), "total", to-from, "number", block.Number(), "hash", block.Hash(), "elapsed", common.PrettyDuration(time.Since(start)))
+ logged = time.Now()
+ }
+ }
+ }
+ log.Info(doneMsg, "blocks", to-from, "elapsed", common.PrettyDuration(time.Since(start)))
+ return nil
+}
+
+// InitDatabaseFromFreezer reinitializes an empty database from a previous batch
+// of frozen ancient blocks. The method iterates over all the frozen blocks and
+// injects into the database the block hash->number mappings.
+func InitDatabaseFromFreezer(db ethdb.Database) {
+ // If we can't access the freezer or it's empty, abort
+ frozen, err := db.Ancients()
+ if err != nil || frozen == 0 {
+ return
+ }
+ // hashBlock calculates block hash in advance using the multi-routine's concurrent
+ // computing power.
+ hashBlock := func(block *types.Block) { block.Hash() }
+
+ // writeIndex injects hash <-> number mapping into the database.
+ writeIndex := func(batch ethdb.Batch, block *types.Block) { WriteHeaderNumber(batch, block.Hash(), block.NumberU64()) }
+
+ if err := iterateCanonicalChain(db, 0, frozen, hashBlock, writeIndex, false, "Initializing database from freezer", "Initialized database from freezer"); err != nil {
+ log.Crit("Failed to init database from freezer", "err", err)
+ }
+ hash := ReadCanonicalHash(db, frozen-1)
+ WriteHeadHeaderHash(db, hash)
+ WriteHeadFastBlockHash(db, hash)
+}
+
+// IndexTransactions creates txlookup indices of the specified block range.
+//
+// This function iterates canonical chain in reverse order, it has one main advantage:
+// We can write tx index tail flag periodically even without the whole indexing
+// procedure is finished. So that we can resume indexing procedure next time quickly.
+func IndexTransactions(db ethdb.Database, from uint64, to uint64) {
+ // hashTxs calculates transaction hash in advance using the multi-routine's
+ // concurrent computing power.
+ hashTxs := func(block *types.Block) {
+ for _, tx := range block.Transactions() {
+ tx.Hash()
+ }
+ }
+ // writeIndices injects txlookup indices into the database.
+ writeIndices := func(batch ethdb.Batch, block *types.Block) {
+ WriteTxLookupEntries(batch, block)
+ if block.NumberU64() == to-1 || block.NumberU64()%10000 == 0 {
+ WriteTxIndexTail(batch, block.NumberU64())
+ }
+ }
+ if err := iterateCanonicalChain(db, from, to, hashTxs, writeIndices, true, "Indexing transactions", "Indexed transactions"); err != nil {
+ log.Crit("Failed to index transactions", "err", err)
+ }
+ WriteTxIndexTail(db, from)
+}
+
+// UnindexTransactions removes txlookup indices of the specified block range.
+func UnindexTransactions(db ethdb.Database, from uint64, to uint64) {
+ // Write flag first and then unindex the transaction indices. Some indices
+ // will be left in the database if crash happens but it's fine.
+ WriteTxIndexTail(db, to)
+
+ // If only one block is unindexed, do it direclty
+ if from+1 == to {
+ DeleteTxLookupEntries(db, ReadBlock(db, ReadCanonicalHash(db, from), from))
+ return
+ }
+ // Otherwise spin up the concurrent iterator and unindexer
+ deleteIndices := func(batch ethdb.Batch, block *types.Block) { DeleteTxLookupEntries(batch, block) }
+ if err := iterateCanonicalChain(db, from, to, nil, deleteIndices, false, "Unindexing transactions", "Unindexed transactions"); err != nil {
+ log.Crit("Failed to unindex transactions", "err", err)
+ }
+}
diff --git a/core/rawdb/chain_iterator_test.go b/core/rawdb/chain_iterator_test.go
new file mode 100644
index 0000000000..d9505c62da
--- /dev/null
+++ b/core/rawdb/chain_iterator_test.go
@@ -0,0 +1,67 @@
+// Copyright 2019 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see .
+
+package rawdb
+
+import (
+ "math/big"
+ "reflect"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/ethdb"
+)
+
+func TestChainIterator(t *testing.T) {
+ // Construct test chain db
+ chainDb := NewMemoryDatabase()
+
+ var block *types.Block
+ for i := uint64(0); i <= 10; i++ {
+ if i == 0 {
+ block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, nil, nil, nil) // Empty genesis block
+ } else {
+ tx := types.NewTransaction(i, common.BytesToAddress([]byte{0x11}), big.NewInt(111), 1111, big.NewInt(11111), []byte{0x11, 0x11, 0x11})
+ block = types.NewBlock(&types.Header{Number: big.NewInt(int64(i))}, []*types.Transaction{tx}, nil, nil)
+ }
+ WriteBlock(chainDb, block)
+ WriteCanonicalHash(chainDb, block.Hash(), block.NumberU64())
+ }
+
+ var cases = []struct {
+ from, to uint64
+ reverse bool
+ expect []uint64
+ }{
+ {0, 11, true, []uint64{10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0}},
+ {0, 0, true, nil},
+ {10, 11, true, []uint64{10}},
+ {0, 11, false, []uint64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10}},
+ {0, 0, false, nil},
+ {10, 11, false, []uint64{10}},
+ }
+ for i, c := range cases {
+ var visit []uint64
+ err := iterateCanonicalChain(chainDb, c.from, c.to, nil, func(db ethdb.Batch, b *types.Block) { visit = append(visit, b.NumberU64()) }, c.reverse, "", "")
+ if err != nil {
+ t.Fatalf("Case %d failed, err %v", i, err)
+ }
+ if !reflect.DeepEqual(visit, c.expect) {
+ t.Fatalf("Case %d failed, visit element mismatch, want %v, got %v", i, c.expect, visit)
+ }
+ }
+}
diff --git a/core/rawdb/freezer_reinit.go b/core/rawdb/freezer_reinit.go
deleted file mode 100644
index ea4dd33d1d..0000000000
--- a/core/rawdb/freezer_reinit.go
+++ /dev/null
@@ -1,127 +0,0 @@
-// Copyright 2019 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package rawdb
-
-import (
- "errors"
- "runtime"
- "sync/atomic"
- "time"
-
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/common/prque"
- "github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/ethdb"
- "github.com/ethereum/go-ethereum/log"
-)
-
-// InitDatabaseFromFreezer reinitializes an empty database from a previous batch
-// of frozen ancient blocks. The method iterates over all the frozen blocks and
-// injects into the database the block hash->number mappings and the transaction
-// lookup entries.
-func InitDatabaseFromFreezer(db ethdb.Database) error {
- // If we can't access the freezer or it's empty, abort
- frozen, err := db.Ancients()
- if err != nil || frozen == 0 {
- return err
- }
- // Blocks previously frozen, iterate over- and hash them concurrently
- var (
- number = ^uint64(0) // -1
- results = make(chan *types.Block, 4*runtime.NumCPU())
- )
- abort := make(chan struct{})
- defer close(abort)
-
- for i := 0; i < runtime.NumCPU(); i++ {
- go func() {
- for {
- // Fetch the next task number, terminating if everything's done
- n := atomic.AddUint64(&number, 1)
- if n >= frozen {
- return
- }
- // Retrieve the block from the freezer (no need for the hash, we pull by
- // number from the freezer). If successful, pre-cache the block hash and
- // the individual transaction hashes for storing into the database.
- block := ReadBlock(db, common.Hash{}, n)
- if block != nil {
- block.Hash()
- for _, tx := range block.Transactions() {
- tx.Hash()
- }
- }
- // Feed the block to the aggregator, or abort on interrupt
- select {
- case results <- block:
- case <-abort:
- return
- }
- }
- }()
- }
- // Reassemble the blocks into a contiguous stream and push them out to disk
- var (
- queue = prque.New(nil)
- next = int64(0)
-
- batch = db.NewBatch()
- start = time.Now()
- logged time.Time
- )
- for i := uint64(0); i < frozen; i++ {
- // Retrieve the next result and bail if it's nil
- block := <-results
- if block == nil {
- return errors.New("broken ancient database")
- }
- // Push the block into the import queue and process contiguous ranges
- queue.Push(block, -int64(block.NumberU64()))
- for !queue.Empty() {
- // If the next available item is gapped, return
- if _, priority := queue.Peek(); -priority != next {
- break
- }
- // Next block available, pop it off and index it
- block = queue.PopItem().(*types.Block)
- next++
-
- // Inject hash<->number mapping and txlookup indexes
- WriteHeaderNumber(batch, block.Hash(), block.NumberU64())
- WriteTxLookupEntries(batch, block)
-
- // If enough data was accumulated in memory or we're at the last block, dump to disk
- if batch.ValueSize() > ethdb.IdealBatchSize || uint64(next) == frozen {
- if err := batch.Write(); err != nil {
- return err
- }
- batch.Reset()
- }
- // If we've spent too much time already, notify the user of what we're doing
- if time.Since(logged) > 8*time.Second {
- log.Info("Initializing chain from ancient data", "number", block.Number(), "hash", block.Hash(), "total", frozen-1, "elapsed", common.PrettyDuration(time.Since(start)))
- logged = time.Now()
- }
- }
- }
- hash := ReadCanonicalHash(db, frozen-1)
- WriteHeadHeaderHash(db, hash)
- WriteHeadFastBlockHash(db, hash)
-
- log.Info("Initialized chain from ancient data", "number", frozen-1, "hash", hash, "elapsed", common.PrettyDuration(time.Since(start)))
- return nil
-}
diff --git a/core/rawdb/schema.go b/core/rawdb/schema.go
index a44a2c99f9..745a59922b 100644
--- a/core/rawdb/schema.go
+++ b/core/rawdb/schema.go
@@ -41,6 +41,12 @@ var (
// fastTrieProgressKey tracks the number of trie entries imported during fast sync.
fastTrieProgressKey = []byte("TrieSync")
+ // txIndexTailKey tracks the oldest block whose transactions have been indexed.
+ txIndexTailKey = []byte("TransactionIndexTail")
+
+ // fastTxLookupLimitKey tracks the transaction lookup limit during fast sync.
+ fastTxLookupLimitKey = []byte("FastTransactionLookupLimit")
+
// Data item prefixes (use single byte to avoid mixing data types, avoid `i`, used for indexes).
headerPrefix = []byte("h") // headerPrefix + num (uint64 big endian) + hash -> header
headerTDSuffix = []byte("t") // headerPrefix + num (uint64 big endian) + hash + headerTDSuffix -> td
diff --git a/eth/backend.go b/eth/backend.go
index 83e05e96a8..6ec34d0539 100644
--- a/eth/backend.go
+++ b/eth/backend.go
@@ -184,7 +184,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
TrieTimeLimit: config.TrieTimeout,
}
)
- eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve)
+ eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve, &config.TxLookupLimit)
if err != nil {
return nil, err
}
diff --git a/eth/config.go b/eth/config.go
index 5094a533bf..fe846c0546 100644
--- a/eth/config.go
+++ b/eth/config.go
@@ -98,6 +98,8 @@ type Config struct {
NoPruning bool // Whether to disable pruning and flush everything to disk
NoPrefetch bool // Whether to disable prefetching and only load state on demand
+ TxLookupLimit uint64 `toml:",omitempty"` // The maximum number of blocks from head whose tx indices are reserved.
+
// Whitelist of required block number -> hash values to accept
Whitelist map[uint64]common.Hash `toml:"-"`
diff --git a/eth/gen_config.go b/eth/gen_config.go
index bc4b55b120..1a19314dc2 100644
--- a/eth/gen_config.go
+++ b/eth/gen_config.go
@@ -23,6 +23,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
SyncMode downloader.SyncMode
NoPruning bool
NoPrefetch bool
+ TxLookupLimit uint64 `toml:",omitempty"`
Whitelist map[uint64]common.Hash `toml:"-"`
LightServ int `toml:",omitempty"`
LightIngress int `toml:",omitempty"`
@@ -56,6 +57,7 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.SyncMode = c.SyncMode
enc.NoPruning = c.NoPruning
enc.NoPrefetch = c.NoPrefetch
+ enc.TxLookupLimit = c.TxLookupLimit
enc.Whitelist = c.Whitelist
enc.LightServ = c.LightServ
enc.LightIngress = c.LightIngress
@@ -93,6 +95,7 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
SyncMode *downloader.SyncMode
NoPruning *bool
NoPrefetch *bool
+ TxLookupLimit *uint64 `toml:",omitempty"`
Whitelist map[uint64]common.Hash `toml:"-"`
LightServ *int `toml:",omitempty"`
LightIngress *int `toml:",omitempty"`
@@ -139,6 +142,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.NoPrefetch != nil {
c.NoPrefetch = *dec.NoPrefetch
}
+ if dec.TxLookupLimit != nil {
+ c.TxLookupLimit = *dec.TxLookupLimit
+ }
if dec.Whitelist != nil {
c.Whitelist = dec.Whitelist
}
diff --git a/eth/handler.go b/eth/handler.go
index d2355a8768..d6dc9c5009 100644
--- a/eth/handler.go
+++ b/eth/handler.go
@@ -51,7 +51,7 @@ const (
// The number is referenced from the size of tx pool.
txChanSize = 4096
- // minimim number of peers to broadcast new blocks to
+ // minBroadcastPeers is the minimal number of peers to broadcast new blocks to.
minBroadcastPeers = 4
)
@@ -75,6 +75,7 @@ type ProtocolManager struct {
txpool txPool
blockchain *core.BlockChain
+ chaindb ethdb.Database
maxPeers int
downloader *downloader.Downloader
@@ -109,6 +110,7 @@ func NewProtocolManager(config *params.ChainConfig, checkpoint *params.TrustedCh
eventMux: mux,
txpool: txpool,
blockchain: blockchain,
+ chaindb: chaindb,
peers: newPeerSet(),
whitelist: whitelist,
newPeerCh: make(chan *peer),
diff --git a/eth/handler_test.go b/eth/handler_test.go
index 256883d1f0..c4083854eb 100644
--- a/eth/handler_test.go
+++ b/eth/handler_test.go
@@ -491,7 +491,7 @@ func testCheckpointChallenge(t *testing.T, syncmode downloader.SyncMode, checkpo
}
}
// Create a checkpoint aware protocol manager
- blockchain, err := core.NewBlockChain(db, nil, config, ethash.NewFaker(), vm.Config{}, nil)
+ blockchain, err := core.NewBlockChain(db, nil, config, ethash.NewFaker(), vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create new blockchain: %v", err)
}
@@ -578,7 +578,7 @@ func testBroadcastBlock(t *testing.T, totalPeers, broadcastExpected int) {
gspec = &core.Genesis{Config: config}
genesis = gspec.MustCommit(db)
)
- blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil)
+ blockchain, err := core.NewBlockChain(db, nil, config, pow, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create new blockchain: %v", err)
}
diff --git a/eth/helper_test.go b/eth/helper_test.go
index e66910334f..ec907ed1cc 100644
--- a/eth/helper_test.go
+++ b/eth/helper_test.go
@@ -62,7 +62,7 @@ func newTestProtocolManager(mode downloader.SyncMode, blocks int, generator func
Alloc: core.GenesisAlloc{testBank: {Balance: big.NewInt(1000000)}},
}
genesis = gspec.MustCommit(db)
- blockchain, _ = core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil)
+ blockchain, _ = core.NewBlockChain(db, nil, gspec.Config, engine, vm.Config{}, nil, nil)
)
chain, _ := core.GenerateChain(gspec.Config, genesis, ethash.NewFaker(), db, blocks, generator)
if _, err := blockchain.InsertChain(chain); err != nil {
diff --git a/eth/sync.go b/eth/sync.go
index 9e180ee200..57fa914d38 100644
--- a/eth/sync.go
+++ b/eth/sync.go
@@ -22,6 +22,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/log"
@@ -186,6 +187,24 @@ func (pm *ProtocolManager) synchronise(peer *peer) {
return
}
}
+ if mode == downloader.FastSync {
+ // Before launch the fast sync, we have to ensure user uses the same
+ // txlookup limit.
+ // The main concern here is: during the fast sync Geth won't index the
+ // block(generate tx indices) before the HEAD-limit. But if user changes
+ // the limit in the next fast sync(e.g. user kill Geth manually and
+ // restart) then it will be hard for Geth to figure out the oldest block
+ // has been indexed. So here for the user-experience wise, it's non-optimal
+ // that user can't change limit during the fast sync. If changed, Geth
+ // will just blindly use the original one.
+ limit := pm.blockchain.TxLookupLimit()
+ if stored := rawdb.ReadFastTxLookupLimit(pm.chaindb); stored == nil {
+ rawdb.WriteFastTxLookupLimit(pm.chaindb, limit)
+ } else if *stored != limit {
+ pm.blockchain.SetTxLookupLimit(*stored)
+ log.Warn("Update txLookup limit", "provided", limit, "updated", *stored)
+ }
+ }
// Run the sync cycle, and disable fast sync if we've went past the pivot block
if err := pm.downloader.Synchronise(peer.id, pHead, pTd, mode); err != nil {
return
diff --git a/light/odr_test.go b/light/odr_test.go
index debd5544c3..de09a3b0ee 100644
--- a/light/odr_test.go
+++ b/light/odr_test.go
@@ -257,7 +257,7 @@ func testChainOdr(t *testing.T, protocol int, fn odrTestFn) {
)
gspec.MustCommit(ldb)
// Assemble the test environment
- blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
+ blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, nil)
gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, 4, testChainGen)
if _, err := blockchain.InsertChain(gchain); err != nil {
t.Fatal(err)
diff --git a/light/trie_test.go b/light/trie_test.go
index 4919f89641..052194b4d8 100644
--- a/light/trie_test.go
+++ b/light/trie_test.go
@@ -40,7 +40,7 @@ func TestNodeIterator(t *testing.T) {
genesis = gspec.MustCommit(fulldb)
)
gspec.MustCommit(lightdb)
- blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
+ blockchain, _ := core.NewBlockChain(fulldb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, nil)
gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), fulldb, 4, testChainGen)
if _, err := blockchain.InsertChain(gchain); err != nil {
panic(err)
diff --git a/light/txpool_test.go b/light/txpool_test.go
index 0996bd7c9c..39d5afe52f 100644
--- a/light/txpool_test.go
+++ b/light/txpool_test.go
@@ -88,7 +88,7 @@ func TestTxPool(t *testing.T) {
)
gspec.MustCommit(ldb)
// Assemble the test environment
- blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil)
+ blockchain, _ := core.NewBlockChain(sdb, nil, params.TestChainConfig, ethash.NewFullFaker(), vm.Config{}, nil, nil)
gchain, _ := core.GenerateChain(params.TestChainConfig, genesis, ethash.NewFaker(), sdb, poolTestBlocks, txPoolTestChainGen)
if _, err := blockchain.InsertChain(gchain); err != nil {
panic(err)
diff --git a/miner/worker_test.go b/miner/worker_test.go
index faebee1a72..59646c55ef 100644
--- a/miner/worker_test.go
+++ b/miner/worker_test.go
@@ -116,7 +116,7 @@ func newTestWorkerBackend(t *testing.T, chainConfig *params.ChainConfig, engine
}
genesis := gspec.MustCommit(db)
- chain, _ := core.NewBlockChain(db, &core.CacheConfig{TrieDirtyDisabled: true}, gspec.Config, engine, vm.Config{}, nil)
+ chain, _ := core.NewBlockChain(db, &core.CacheConfig{TrieDirtyDisabled: true}, gspec.Config, engine, vm.Config{}, nil, nil)
txpool := core.NewTxPool(testTxPoolConfig, chainConfig, chain)
// Generate a small n-block chain and an uncle block for it
@@ -213,7 +213,7 @@ func testGenerateBlockAndImport(t *testing.T, isClique bool) {
db2 := rawdb.NewMemoryDatabase()
b.genesis.MustCommit(db2)
- chain, _ := core.NewBlockChain(db2, nil, b.chain.Config(), engine, vm.Config{}, nil)
+ chain, _ := core.NewBlockChain(db2, nil, b.chain.Config(), engine, vm.Config{}, nil, nil)
defer chain.Stop()
newBlock := make(chan struct{})
diff --git a/tests/block_test_util.go b/tests/block_test_util.go
index 81dd7b1d04..ac5957fba6 100644
--- a/tests/block_test_util.go
+++ b/tests/block_test_util.go
@@ -118,7 +118,7 @@ func (t *BlockTest) Run() error {
} else {
engine = ethash.NewShared()
}
- chain, err := core.NewBlockChain(db, &core.CacheConfig{TrieCleanLimit: 0}, config, engine, vm.Config{}, nil)
+ chain, err := core.NewBlockChain(db, &core.CacheConfig{TrieCleanLimit: 0}, config, engine, vm.Config{}, nil, nil)
if err != nil {
return err
}