Prune spendjournal depth

blockchain/chain.go

@@ -119,6 +119,10 @@ type BlockChain struct {
 	// is pruned.
 	pruneTarget uint64
 
+	// pruneSpendJDepth is the block depth of spend journal entries to
+	// keep. Zero means disabled spend journal pruning.
+	pruneSpendJDepth int32
+
 	// These fields are related to the memory block index.  They both have
 	// their own locks, however they are often also protected by the chain
 	// lock to help prevent logic races when blocks are being processed.
@@ -676,6 +680,17 @@ func (b *BlockChain) connectBlock(node *blockNode, block *btcutil.Block,
 			return err
 		}
 
+		// Prune the spend journal depth if spend journal pruning is
+		// enabled.
+		if b.pruneSpendJDepth != 0 {
+			err = dbPruneSpendJournalDepth(
+				dbTx, b.pruneSpendJDepth, block.Height(),
+			)
+			if err != nil {
+				return err
+			}
+		}
+
 		// Allow the index manager to call each of the currently active
 		// optional indexes with the block being connected so they can
 		// update themselves accordingly.
@@ -2117,6 +2132,43 @@ func (b *BlockChain) ReconsiderBlock(hash *chainhash.Hash) error {
 	return b.reorganizeChain(detachNodes, attachNodes)
 }
 
+// setupSpendJournalPruning initializes internal journal pruning control
+// structures and prunes the spend journal if needed.
+func (b *BlockChain) setupSpendJournalPruning(interrupt <-chan struct{}) error {
+	// If spend journal pruning is disabled, we have nothing to do here.
+	if b.pruneSpendJDepth == 0 {
+		return nil
+	}
+
+	log.Info("Executing the initial spend journal pruning, the first " +
+		"initialization after enabling or changing the journal prune " +
+		"depth may take a long time.")
+
+	var stats spendJournalStats
+	err := b.db.Update(func(tx database.Tx) error {
+		err := dbInitSpendJournalStats(
+			tx, interrupt, b.pruneSpendJDepth,
+		)
+		if err != nil {
+			return err
+		}
+		stats, err = dbFetchSpendJournalStats(tx)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	bestHeight := b.bestChain.height()
+	return dbPruneSpendJournalDepthSlow(
+		b.db, b.pruneSpendJDepth, stats.nextHeight, bestHeight,
+		interrupt,
+	)
+}
+
 // IndexManager provides a generic interface that the is called when blocks are
 // connected and disconnected to and from the tip of the main chain for the
 // purpose of supporting optional indexes.
@@ -2213,6 +2265,11 @@ type Config struct {
 	// will target for with block files.  Prune at 0 specifies that no
 	// blocks will be deleted.
 	Prune uint64
+
+	// PruneSpendJ specifies the target block depth of spend journal
+	// entries to keep. A value of zero means that spend journal pruning is
+	// disabled.
+	PruneSpendJ int32
 }
 
 // New returns a BlockChain instance using the provided configuration details.
@@ -2271,6 +2328,7 @@ func New(config *Config) (*BlockChain, error) {
 		warningCaches:       newThresholdCaches(vbNumBits),
 		deploymentCaches:    newThresholdCaches(chaincfg.DefinedDeployments),
 		pruneTarget:         config.Prune,
+		pruneSpendJDepth:    config.PruneSpendJ,
 	}
 
 	// Ensure all the deployments are synchronized with our clock if
@@ -2323,6 +2381,11 @@ func New(config *Config) (*BlockChain, error) {
 		bestNode.height, bestNode.hash, b.stateSnapshot.TotalTxns,
 		bestNode.workSum)
 
+	// Setup spend journal pruning.
+	if err := b.setupSpendJournalPruning(config.Interrupt); err != nil {
+		return nil, err
+	}
+
 	return &b, nil
 }
 

blockchain/chainio.go

@@ -7,6 +7,7 @@ package blockchain
 import (
 	"bytes"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"math/big"
 	"sync"
@@ -59,6 +60,8 @@ var (
 	// the version of the spend journal currently in the database.
 	spendJournalVersionKeyName = []byte("spendjournalversion")
 
+	spendJournalStatsKeyName = []byte("spendjournalstats")
+
 	// spendJournalBucketName is the name of the db bucket used to house
 	// transactions outputs that are spent in each block.
 	spendJournalBucketName = []byte("spendjournal")
@@ -76,6 +79,10 @@ var (
 	byteOrder = binary.LittleEndian
 )
 
+var (
+	errMissingSpendJournalStats = errors.New("missing spend journal stats")
+)
+
 // errNotInMainChain signifies that a block hash or height that is not in the
 // main chain was requested.
 type errNotInMainChain string
@@ -513,6 +520,256 @@ func dbPruneSpendJournalEntry(dbTx database.Tx, blockHashes []chainhash.Hash) er
 	return nil
 }
 
+// spendJournalStats is the internal control structure used for depth pruning of
+// the spend journal.
+type spendJournalStats struct {
+	pruneDepth int32
+	nextHeight int32
+}
+
+// deserialize reads data into s.
+func (s *spendJournalStats) deserialize(data []byte) error {
+	if len(data) != 8 {
+		return fmt.Errorf("bad spend journal stats len")
+	}
+
+	s.pruneDepth = int32(byteOrder.Uint32(data[:4]))
+	s.nextHeight = int32(byteOrder.Uint32(data[4:8]))
+
+	return nil
+}
+
+// serialize appends the serialization of s into buf and returns the serialized
+// spendJournalStats.
+func (s *spendJournalStats) serialize(buf []byte) []byte {
+	buf = byteOrder.AppendUint32(buf, uint32(s.pruneDepth))
+	return byteOrder.AppendUint32(buf, uint32(s.nextHeight))
+}
+
+// dbFetchSpendJournalStats returns the spendJournalStats from DB, or error in
+// case the entry does not exists.
+func dbFetchSpendJournalStats(dbTx database.Tx) (spendJournalStats, error) {
+	var zr spendJournalStats
+	serialized := dbTx.Metadata().Get(spendJournalStatsKeyName)
+	if serialized == nil {
+		return zr, errMissingSpendJournalStats
+	}
+
+	var stats spendJournalStats
+	err := stats.deserialize(serialized)
+	if err != nil {
+		return zr, err
+	}
+
+	return stats, nil
+}
+
+// DBSpendJournalPruneDepth returns the previously configured spend journal
+// prune depth, or 0 if spend journal pruning was never enabled.
+func DBSpendJournalPruneDepth(dbTx database.Tx) (int32, error) {
+	stats, err := dbFetchSpendJournalStats(dbTx)
+	if err != nil && errors.Is(err, errMissingSpendJournalStats) {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+
+	return stats.pruneDepth, nil
+}
+
+// dbPutSpendJournalStats stores the spend journal stats into the database.
+func dbPutSpendJournalStats(dbTx database.Tx, stats spendJournalStats) error {
+	return dbTx.Metadata().Put(
+		spendJournalStatsKeyName, stats.serialize(nil),
+	)
+}
+
+// dbInitSpendJournalStats updates the spend journal stats pruneDepth if it
+// exists in the DB, otherwise, the spend journal will be scanned to initialize
+// spend journal stats nextHeight with the height of the lowest spend journal
+// block height and the passed pruneDepth.
+func dbInitSpendJournalStats(dbTx database.Tx,
+	interrupt <-chan struct{}, pruneDepth int32) error {
+
+	initialized := true
+	stats, err := dbFetchSpendJournalStats(dbTx)
+	if err != nil && errors.Is(err, errMissingSpendJournalStats) {
+		initialized = false
+	} else if err != nil {
+		return err
+	}
+
+	// Update the stored prune depth if the node was started with a
+	// different --prunespendjournal flag. By storing this config in the
+	// database, --prunespendjournal can be passed only once, which improves
+	// the UX, as next initializations may omit the flag.
+	if initialized && pruneDepth != stats.pruneDepth {
+		stats.pruneDepth = pruneDepth
+
+		return dbPutSpendJournalStats(dbTx, stats)
+	}
+
+	// Find the lowest block height in the spend journal.
+	spendCursor := dbTx.Metadata().Bucket(spendJournalBucketName).Cursor()
+	var height int32
+	{
+		ok := spendCursor.First()
+		if !ok {
+			return dbPutSpendJournalStats(
+				dbTx, spendJournalStats{
+					pruneDepth: pruneDepth,
+					nextHeight: height,
+				},
+			)
+		}
+		hash := spendCursor.Key()
+		if len(hash) != 32 {
+			return fmt.Errorf("bad hash length: %d", len(hash))
+		}
+		bHeight, err := dbFetchHeightByHash(
+			dbTx, (*chainhash.Hash)(hash),
+		)
+		if err != nil {
+			return err
+		}
+
+		height = bHeight
+	}
+	ok := spendCursor.Next()
+	for ; ok; ok = spendCursor.Next() {
+		select {
+		case <-interrupt:
+			return errInterruptRequested
+		default:
+		}
+
+		hash := spendCursor.Key()
+		if len(hash) != 32 {
+			return fmt.Errorf("bad hash length: %d", len(hash))
+		}
+		bHeight, err := dbFetchHeightByHash(
+			dbTx, (*chainhash.Hash)(hash),
+		)
+		if err != nil {
+			return err
+		}
+		height = min(height, bHeight)
+	}
+
+	return dbPutSpendJournalStats(
+		dbTx, spendJournalStats{
+			pruneDepth: pruneDepth, nextHeight: height,
+		},
+	)
+}
+
+// dbPruneSpendJournalDepthSlow prunes the spend journal depth to the passed
+// spend journal prune depth.
+//
+// NOTE: this procedure uses several internal transactions to avoid reaching
+// database transaction size limit.
+func dbPruneSpendJournalDepthSlow(db database.DB, pruneDepth,
+	spendJournalHeight, bestHeight int32, interrupt <-chan struct{}) error {
+
+	const maxBatch = 100000
+
+	totalEntries := bestHeight - spendJournalHeight
+	if totalEntries <= pruneDepth {
+		return nil
+	}
+
+	toDelete := totalEntries - pruneDepth
+	hSize := min(toDelete, maxBatch)
+	hashes := make([]chainhash.Hash, 0, hSize)
+
+	// loop while deleted entries is less than entries to delete.
+	var deleted int32
+	for deleted < toDelete {
+		err := db.Update(func(tx database.Tx) error {
+			hashes = hashes[:0]
+			var i int32
+			for i < maxBatch && deleted < toDelete {
+				select {
+				case <-interrupt:
+					return errInterruptRequested
+				default:
+				}
+
+				hash, err := dbFetchHashByHeight(
+					tx, spendJournalHeight+i,
+				)
+				if err != nil {
+					return err
+				}
+				hashes = append(hashes, *hash)
+
+				i++
+				deleted++
+			}
+
+			err := dbPruneSpendJournalEntry(tx, hashes)
+			if err != nil {
+				return err
+			}
+
+			spendJournalHeight += int32(len(hashes))
+			err = dbPutSpendJournalStats(
+				tx, spendJournalStats{
+					nextHeight: spendJournalHeight,
+				},
+			)
+			if err != nil {
+				return err
+			}
+
+			return nil
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// dbPruneSpendJournalDepth prunes the spend journal depth to the passed spend
+// journal prune depth.
+func dbPruneSpendJournalDepth(dbTx database.Tx, pruneDepth,
+	bestHeight int32) error {
+
+	stats, err := dbFetchSpendJournalStats(dbTx)
+	if err != nil {
+		return err
+	}
+	totalEntries := bestHeight - stats.nextHeight
+	if totalEntries <= pruneDepth {
+		return nil
+	}
+
+	toDelete := totalEntries - pruneDepth
+	hashes := make([]chainhash.Hash, 0, toDelete)
+	for i := range toDelete {
+		hash, err := dbFetchHashByHeight(dbTx, stats.nextHeight+i)
+		if err != nil {
+			return err
+		}
+		hashes = append(hashes, *hash)
+	}
+	err = dbPruneSpendJournalEntry(dbTx, hashes)
+	if err != nil {
+		return err
+	}
+
+	stats.nextHeight += toDelete
+	err = dbPutSpendJournalStats(dbTx, stats)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // -----------------------------------------------------------------------------
 // The unspent transaction output (utxo) set consists of an entry for each
 // unspent output using a format that is optimized to reduce space using domain