core/rawdb: untie freezer and ancient chain data

cmd/geth/dbcmd.go

@@ -363,7 +363,7 @@ func inspect(ctx *cli.Context) error {
 	return rawdb.InspectDatabase(db, prefix, start)
 }
 
-func showLeveldbStats(db ethdb.Stater) {
+func showLeveldbStats(db ethdb.KeyValueStater) {
 	if stats, err := db.Stat("leveldb.stats"); err != nil {
 		log.Warn("Failed to read database stats", "error", err)
 	} else {

core/rawdb/accessors_chain.go

@@ -36,7 +36,7 @@ import (
 // ReadCanonicalHash retrieves the hash assigned to a canonical block number.
 func ReadCanonicalHash(db ethdb.Reader, number uint64) common.Hash {
 	var data []byte
-	db.ReadAncients(func(reader ethdb.AncientReader) error {
+	db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
 		data, _ = reader.Ancient(freezerHashTable, number)
 		if len(data) == 0 {
 			// Get it by hash from leveldb
@@ -332,7 +332,7 @@ func ReadHeaderRange(db ethdb.Reader, number uint64, count uint64) []rlp.RawValu
 // ReadHeaderRLP retrieves a block header in its raw RLP database encoding.
 func ReadHeaderRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
 	var data []byte
-	db.ReadAncients(func(reader ethdb.AncientReader) error {
+	db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
 		// First try to look up the data in ancient database. Extra hash
 		// comparison is necessary since ancient database only maintains
 		// the canonical data.
@@ -411,7 +411,7 @@ func deleteHeaderWithoutNumber(db ethdb.KeyValueWriter, hash common.Hash, number
 
 // isCanon is an internal utility method, to check whether the given number/hash
 // is part of the ancient (canon) set.
-func isCanon(reader ethdb.AncientReader, number uint64, hash common.Hash) bool {
+func isCanon(reader ethdb.AncientReaderOp, number uint64, hash common.Hash) bool {
 	h, err := reader.Ancient(freezerHashTable, number)
 	if err != nil {
 		return false
@@ -425,7 +425,7 @@ func ReadBodyRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue
 	// comparison is necessary since ancient database only maintains
 	// the canonical data.
 	var data []byte
-	db.ReadAncients(func(reader ethdb.AncientReader) error {
+	db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
 		// Check if the data is in ancients
 		if isCanon(reader, number, hash) {
 			data, _ = reader.Ancient(freezerBodiesTable, number)
@@ -442,7 +442,7 @@ func ReadBodyRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue
 // block at number, in RLP encoding.
 func ReadCanonicalBodyRLP(db ethdb.Reader, number uint64) rlp.RawValue {
 	var data []byte
-	db.ReadAncients(func(reader ethdb.AncientReader) error {
+	db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
 		data, _ = reader.Ancient(freezerBodiesTable, number)
 		if len(data) > 0 {
 			return nil
@@ -508,7 +508,7 @@ func DeleteBody(db ethdb.KeyValueWriter, hash common.Hash, number uint64) {
 // ReadTdRLP retrieves a block's total difficulty corresponding to the hash in RLP encoding.
 func ReadTdRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
 	var data []byte
-	db.ReadAncients(func(reader ethdb.AncientReader) error {
+	db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
 		// Check if the data is in ancients
 		if isCanon(reader, number, hash) {
 			data, _ = reader.Ancient(freezerDifficultyTable, number)
@@ -568,7 +568,7 @@ func HasReceipts(db ethdb.Reader, hash common.Hash, number uint64) bool {
 // ReadReceiptsRLP retrieves all the transaction receipts belonging to a block in RLP encoding.
 func ReadReceiptsRLP(db ethdb.Reader, hash common.Hash, number uint64) rlp.RawValue {
 	var data []byte
-	db.ReadAncients(func(reader ethdb.AncientReader) error {
+	db.ReadAncients(func(reader ethdb.AncientReaderOp) error {
 		// Check if the data is in ancients
 		if isCanon(reader, number, hash) {
 			data, _ = reader.Ancient(freezerReceiptTable, number)

core/rawdb/chain_freezer.go

@@ -0,0 +1,299 @@
+// Copyright 2022 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 <http://www.gnu.org/licenses/>.
+
+package rawdb
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/params"
+)
+
+const (
+	// freezerRecheckInterval is the frequency to check the key-value database for
+	// chain progression that might permit new blocks to be frozen into immutable
+	// storage.
+	freezerRecheckInterval = time.Minute
+
+	// freezerBatchLimit is the maximum number of blocks to freeze in one batch
+	// before doing an fsync and deleting it from the key-value store.
+	freezerBatchLimit = 30000
+)
+
+// chainFreezer is a wrapper of freezer with additional chain freezing feature.
+// The background thread will keep moving ancient chain segments from key-value
+// database to flat files for saving space on live database.
+type chainFreezer struct {
+	*Freezer
+
+	threshold uint64 // Number of recent blocks not to freeze (params.FullImmutabilityThreshold apart from tests)
+	quit      chan struct{}
+	wg        sync.WaitGroup
+	trigger   chan chan struct{} // Manual blocking freeze trigger, test determinism
+}
+
+// newChainFreezer initializes the freezer for ancient chain data.
+func newChainFreezer(datadir string, namespace string, readonly bool, maxTableSize uint32, tables map[string]bool) (*chainFreezer, error) {
+	freezer, err := NewFreezer(datadir, namespace, readonly, maxTableSize, tables)
+	if err != nil {
+		return nil, err
+	}
+	return &chainFreezer{
+		Freezer:   freezer,
+		threshold: params.FullImmutabilityThreshold,
+		quit:      make(chan struct{}),
+		trigger:   make(chan chan struct{}),
+	}, nil
+}
+
+// Close closes the chain freezer instance and terminates the background thread.
+func (f *chainFreezer) Close() error {
+	err := f.Freezer.Close()
+	select {
+	case <-f.quit:
+	default:
+		close(f.quit)
+	}
+	f.wg.Wait()
+	return err
+}
+
+// freeze is a background thread that periodically checks the blockchain for any
+// import progress and moves ancient data from the fast database into the freezer.
+//
+// This functionality is deliberately broken off from block importing to avoid
+// incurring additional data shuffling delays on block propagation.
+func (f *chainFreezer) freeze(db ethdb.KeyValueStore) {
+	nfdb := &nofreezedb{KeyValueStore: db}
+
+	var (
+		backoff   bool
+		triggered chan struct{} // Used in tests
+	)
+	for {
+		select {
+		case <-f.quit:
+			log.Info("Freezer shutting down")
+			return
+		default:
+		}
+		if backoff {
+			// If we were doing a manual trigger, notify it
+			if triggered != nil {
+				triggered <- struct{}{}
+				triggered = nil
+			}
+			select {
+			case <-time.NewTimer(freezerRecheckInterval).C:
+				backoff = false
+			case triggered = <-f.trigger:
+				backoff = false
+			case <-f.quit:
+				return
+			}
+		}
+		// Retrieve the freezing threshold.
+		hash := ReadHeadBlockHash(nfdb)
+		if hash == (common.Hash{}) {
+			log.Debug("Current full block hash unavailable") // new chain, empty database
+			backoff = true
+			continue
+		}
+		number := ReadHeaderNumber(nfdb, hash)
+		threshold := atomic.LoadUint64(&f.threshold)
+
+		switch {
+		case number == nil:
+			log.Error("Current full block number unavailable", "hash", hash)
+			backoff = true
+			continue
+
+		case *number < threshold:
+			log.Debug("Current full block not old enough", "number", *number, "hash", hash, "delay", threshold)
+			backoff = true
+			continue
+
+		case *number-threshold <= f.frozen:
+			log.Debug("Ancient blocks frozen already", "number", *number, "hash", hash, "frozen", f.frozen)
+			backoff = true
+			continue
+		}
+		head := ReadHeader(nfdb, hash, *number)
+		if head == nil {
+			log.Error("Current full block unavailable", "number", *number, "hash", hash)
+			backoff = true
+			continue
+		}
+
+		// Seems we have data ready to be frozen, process in usable batches
+		var (
+			start    = time.Now()
+			first, _ = f.Ancients()
+			limit    = *number - threshold
+		)
+		if limit-first > freezerBatchLimit {
+			limit = first + freezerBatchLimit
+		}
+		ancients, err := f.freezeRange(nfdb, first, limit)
+		if err != nil {
+			log.Error("Error in block freeze operation", "err", err)
+			backoff = true
+			continue
+		}
+
+		// Batch of blocks have been frozen, flush them before wiping from leveldb
+		if err := f.Sync(); err != nil {
+			log.Crit("Failed to flush frozen tables", "err", err)
+		}
+
+		// Wipe out all data from the active database
+		batch := db.NewBatch()
+		for i := 0; i < len(ancients); i++ {
+			// Always keep the genesis block in active database
+			if first+uint64(i) != 0 {
+				DeleteBlockWithoutNumber(batch, ancients[i], first+uint64(i))
+				DeleteCanonicalHash(batch, first+uint64(i))
+			}
+		}
+		if err := batch.Write(); err != nil {
+			log.Crit("Failed to delete frozen canonical blocks", "err", err)
+		}
+		batch.Reset()
+
+		// Wipe out side chains also and track dangling side chains
+		var dangling []common.Hash
+		for number := first; number < f.frozen; number++ {
+			// Always keep the genesis block in active database
+			if number != 0 {
+				dangling = ReadAllHashes(db, number)
+				for _, hash := range dangling {
+					log.Trace("Deleting side chain", "number", number, "hash", hash)
+					DeleteBlock(batch, hash, number)
+				}
+			}
+		}
+		if err := batch.Write(); err != nil {
+			log.Crit("Failed to delete frozen side blocks", "err", err)
+		}
+		batch.Reset()
+
+		// Step into the future and delete and dangling side chains
+		if f.frozen > 0 {
+			tip := f.frozen
+			for len(dangling) > 0 {
+				drop := make(map[common.Hash]struct{})
+				for _, hash := range dangling {
+					log.Debug("Dangling parent from Freezer", "number", tip-1, "hash", hash)
+					drop[hash] = struct{}{}
+				}
+				children := ReadAllHashes(db, tip)
+				for i := 0; i < len(children); i++ {
+					// Dig up the child and ensure it's dangling
+					child := ReadHeader(nfdb, children[i], tip)
+					if child == nil {
+						log.Error("Missing dangling header", "number", tip, "hash", children[i])
+						continue
+					}
+					if _, ok := drop[child.ParentHash]; !ok {
+						children = append(children[:i], children[i+1:]...)
+						i--
+						continue
+					}
+					// Delete all block data associated with the child
+					log.Debug("Deleting dangling block", "number", tip, "hash", children[i], "parent", child.ParentHash)
+					DeleteBlock(batch, children[i], tip)
+				}
+				dangling = children
+				tip++
+			}
+			if err := batch.Write(); err != nil {
+				log.Crit("Failed to delete dangling side blocks", "err", err)
+			}
+		}
+
+		// Log something friendly for the user
+		context := []interface{}{
+			"blocks", f.frozen - first, "elapsed", common.PrettyDuration(time.Since(start)), "number", f.frozen - 1,
+		}
+		if n := len(ancients); n > 0 {
+			context = append(context, []interface{}{"hash", ancients[n-1]}...)
+		}
+		log.Info("Deep froze chain segment", context...)
+
+		// Avoid database thrashing with tiny writes
+		if f.frozen-first < freezerBatchLimit {
+			backoff = true
+		}
+	}
+}
+
+func (f *chainFreezer) freezeRange(nfdb *nofreezedb, number, limit uint64) (hashes []common.Hash, err error) {
+	hashes = make([]common.Hash, 0, limit-number)
+
+	_, err = f.ModifyAncients(func(op ethdb.AncientWriteOp) error {
+		for ; number <= limit; number++ {
+			// Retrieve all the components of the canonical block.
+			hash := ReadCanonicalHash(nfdb, number)
+			if hash == (common.Hash{}) {
+				return fmt.Errorf("canonical hash missing, can't freeze block %d", number)
+			}
+			header := ReadHeaderRLP(nfdb, hash, number)
+			if len(header) == 0 {
+				return fmt.Errorf("block header missing, can't freeze block %d", number)
+			}
+			body := ReadBodyRLP(nfdb, hash, number)
+			if len(body) == 0 {
+				return fmt.Errorf("block body missing, can't freeze block %d", number)
+			}
+			receipts := ReadReceiptsRLP(nfdb, hash, number)
+			if len(receipts) == 0 {
+				return fmt.Errorf("block receipts missing, can't freeze block %d", number)
+			}
+			td := ReadTdRLP(nfdb, hash, number)
+			if len(td) == 0 {
+				return fmt.Errorf("total difficulty missing, can't freeze block %d", number)
+			}
+
+			// Write to the batch.
+			if err := op.AppendRaw(freezerHashTable, number, hash[:]); err != nil {
+				return fmt.Errorf("can't write hash to Freezer: %v", err)
+			}
+			if err := op.AppendRaw(freezerHeaderTable, number, header); err != nil {
+				return fmt.Errorf("can't write header to Freezer: %v", err)
+			}
+			if err := op.AppendRaw(freezerBodiesTable, number, body); err != nil {
+				return fmt.Errorf("can't write body to Freezer: %v", err)
+			}
+			if err := op.AppendRaw(freezerReceiptTable, number, receipts); err != nil {
+				return fmt.Errorf("can't write receipts to Freezer: %v", err)
+			}
+			if err := op.AppendRaw(freezerDifficultyTable, number, td); err != nil {
+				return fmt.Errorf("can't write td to Freezer: %v", err)
+			}
+
+			hashes = append(hashes, hash)
+		}
+		return nil
+	})
+
+	return hashes, err
+}