Merge pull request #9945 from ziggie1984/optional-migration

Decayed log optional migration

channeldb/db.go

@@ -29,6 +29,7 @@ import (
 	"github.com/lightningnetwork/lnd/channeldb/migration31"
 	"github.com/lightningnetwork/lnd/channeldb/migration32"
 	"github.com/lightningnetwork/lnd/channeldb/migration33"
+	"github.com/lightningnetwork/lnd/channeldb/migration34"
 	"github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
 	"github.com/lightningnetwork/lnd/clock"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
@@ -74,12 +75,14 @@ type mandatoryVersion struct {
 // optional migrations.
 type MigrationConfig interface {
 	migration30.MigrateRevLogConfig
+	migration34.MigrationConfig
 }
 
 // MigrationConfigImpl is a super set of all the various migration configs and
 // an implementation of MigrationConfig.
 type MigrationConfigImpl struct {
 	migration30.MigrateRevLogConfigImpl
+	migration34.MigrationConfigImpl
 }
 
 // optionalMigration defines an optional migration function. When a migration
@@ -308,13 +311,23 @@ var (
 	// to determine its state.
 	optionalVersions = []optionalVersion{
 		{
-			name: "prune revocation log",
+			name: "prune_revocation_log",
 			migration: func(db kvdb.Backend,
 				cfg MigrationConfig) error {
 
 				return migration30.MigrateRevocationLog(db, cfg)
 			},
 		},
+		{
+			name: "gc_decayed_log",
+			migration: func(db kvdb.Backend,
+				cfg MigrationConfig) error {
+
+				return migration34.MigrateDecayedLog(
+					db, cfg,
+				)
+			},
+		},
 	}
 
 	// Big endian is the preferred byte order, due to cursor scans over
@@ -1731,10 +1744,8 @@ func (d *DB) syncVersions(versions []mandatoryVersion) error {
 	}, func() {})
 }
 
-// applyOptionalVersions takes a config to determine whether the optional
-// migrations will be applied.
-//
-// NOTE: only support the prune_revocation_log optional migration atm.
+// applyOptionalVersions applies the optional migrations to the database if
+// specified in the config.
 func (d *DB) applyOptionalVersions(cfg OptionalMiragtionConfig) error {
 	// TODO(yy): need to design the db to support dry run for optional
 	// migrations.
@@ -1751,52 +1762,73 @@ func (d *DB) applyOptionalVersions(cfg OptionalMiragtionConfig) error {
 				Versions: make(map[uint64]string),
 			}
 		} else {
-			return err
+			return fmt.Errorf("unable to fetch optional "+
+				"meta: %w", err)
 		}
 	}
 
-	log.Infof("Checking for optional update: prune_revocation_log=%v, "+
-		"db_version=%s", cfg.PruneRevocationLog, om)
-
-	// Exit early if the optional migration is not specified.
-	if !cfg.PruneRevocationLog {
-		return nil
-	}
-
-	// Exit early if the optional migration has already been applied.
-	if _, ok := om.Versions[0]; ok {
-		return nil
-	}
-
-	// Get the optional version.
-	version := optionalVersions[0]
-	log.Infof("Performing database optional migration: %s", version.name)
-
+	// migrationCfg is the parent configuration which implements the config
+	// interfaces of all the single optional migrations.
 	migrationCfg := &MigrationConfigImpl{
 		migration30.MigrateRevLogConfigImpl{
 			NoAmountData: d.noRevLogAmtData,
 		},
+		migration34.MigrationConfigImpl{
+			DecayedLog: cfg.DecayedLog,
+		},
 	}
 
-	// Migrate the data.
+	log.Infof("Applying %d optional migrations", len(optionalVersions))
+
+	// Apply the optional migrations if requested.
+	for number, version := range optionalVersions {
+		log.Infof("Checking for optional update: name=%v", version.name)
+
+		// Exit early if the optional migration is not specified.
+		if !cfg.MigrationFlags[number] {
+			log.Debugf("Skipping optional migration: name=%s as "+
+				"it is not specified in the config",
+				version.name)
+
+			continue
+		}
+
+		// Exit early if the optional migration has already been
+		// applied.
+		if _, ok := om.Versions[uint64(number)]; ok {
+			log.Debugf("Skipping optional migration: name=%s as "+
+				"it has already been applied", version.name)
+
+			continue
+		}
+
+		log.Infof("Performing database optional migration: %s",
+			version.name)
+
+		// Call the migration function for the specific optional
+		// migration.
 		if err := version.migration(d, migrationCfg); err != nil {
-		log.Errorf("Unable to apply optional migration: %s, error: %v",
-			version.name, err)
+			log.Errorf("Unable to apply optional migration: %s, "+
+				"error: %v", version.name, err)
 			return err
 		}
 
 		// Update the optional meta. Notice that unlike the mandatory db
-	// migrations where we perform the migration and updating meta in a
-	// single db transaction, we use different transactions here. Even when
-	// the following update is failed, we should be fine here as we would
-	// re-run the optional migration again, which is a noop, during next
-	// startup.
-	om.Versions[0] = version.name
+		// migrations where we perform the migration and updating meta
+		// in a single db transaction, we use different transactions
+		// here. Even when the following update is failed, we should be
+		// fine here as we would re-run the optional migration again,
+		// which is a noop, during next startup.
+		om.Versions[uint64(number)] = version.name
 		if err := d.putOptionalMeta(om); err != nil {
 			log.Errorf("Unable to update optional meta: %v", err)
 			return err
 		}
 
+		log.Infof("Successfully applied optional migration: %s",
+			version.name)
+	}
+
 	return nil
 }
 

channeldb/log.go

@@ -12,6 +12,7 @@ import (
 	"github.com/lightningnetwork/lnd/channeldb/migration31"
 	"github.com/lightningnetwork/lnd/channeldb/migration32"
 	"github.com/lightningnetwork/lnd/channeldb/migration33"
+	"github.com/lightningnetwork/lnd/channeldb/migration34"
 	"github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
 	"github.com/lightningnetwork/lnd/kvdb"
 )
@@ -46,5 +47,6 @@ func UseLogger(logger btclog.Logger) {
 	migration31.UseLogger(logger)
 	migration32.UseLogger(logger)
 	migration33.UseLogger(logger)
+	migration34.UseLogger(logger)
 	kvdb.UseLogger(logger)
 }

channeldb/meta.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"slices"
+	"strings"
 
 	"github.com/lightningnetwork/lnd/kvdb"
 	"github.com/lightningnetwork/lnd/tlv"
@@ -30,6 +32,10 @@ var (
 	// ErrMarkerNotPresent is the error that is returned if the queried
 	// marker is not present in the given database.
 	ErrMarkerNotPresent = errors.New("marker not present")
+
+	// ErrInvalidOptionalVersion is the error that is returned if the
+	// optional version persisted in the database is invalid.
+	ErrInvalidOptionalVersion = errors.New("invalid optional version")
 )
 
 // Meta structure holds the database meta information.
@@ -104,15 +110,28 @@ type OptionalMeta struct {
 	Versions map[uint64]string
 }
 
+// String returns a string representation of the optional meta.
 func (om *OptionalMeta) String() string {
-	s := ""
-	for index, name := range om.Versions {
-		s += fmt.Sprintf("%d: %s", index, name)
+	if len(om.Versions) == 0 {
+		return "empty"
 	}
-	if s == "" {
-		s = "empty"
+
+	// Create a slice of indices to sort
+	indices := make([]uint64, 0, len(om.Versions))
+	for index := range om.Versions {
+		indices = append(indices, index)
 	}
-	return s
+
+	// Sort the indices in ascending order.
+	slices.Sort(indices)
+
+	// Create the string parts in sorted order.
+	parts := make([]string, len(indices))
+	for i, index := range indices {
+		parts[i] = fmt.Sprintf("%d: %s", index, om.Versions[index])
+	}
+
+	return strings.Join(parts, ", ")
 }
 
 // fetchOptionalMeta reads the optional meta from the database.
@@ -146,7 +165,20 @@ func (d *DB) fetchOptionalMeta() (*OptionalMeta, error) {
 			if err != nil {
 				return err
 			}
-			om.Versions[version] = optionalVersions[i].name
+
+			// This check would not allow to downgrade LND software
+			// to a version with an optional migration when an
+			// optional migration not known to the current version
+			// has already been applied.
+			if version >= uint64(len(optionalVersions)) {
+				return fmt.Errorf("optional version read "+
+					"from db is %d, but only optional "+
+					"migrations up to %d are known: %w",
+					version, len(optionalVersions)-1,
+					ErrInvalidOptionalVersion)
+			}
+
+			om.Versions[version] = optionalVersions[version].name
 		}
 
 		return nil
@@ -174,8 +206,12 @@ func (d *DB) putOptionalMeta(om *OptionalMeta) error {
 			return err
 		}
 
-		// Write the version indexes.
+		// Write the version indexes of the single migrations.
 		for v := range om.Versions {
+			if v >= uint64(len(optionalVersions)) {
+				return ErrInvalidOptionalVersion
+			}
+
 			err := tlv.WriteVarInt(&b, v, &[8]byte{})
 			if err != nil {
 				return err

channeldb/meta_test.go

@@ -498,6 +498,7 @@ func TestOptionalMeta(t *testing.T) {
 	om = &OptionalMeta{
 		Versions: map[uint64]string{
 			0: optionalVersions[0].name,
+			1: optionalVersions[1].name,
 		},
 	}
 	err = db.putOptionalMeta(om)
@@ -506,29 +507,40 @@ func TestOptionalMeta(t *testing.T) {
 	om1, err := db.fetchOptionalMeta()
 	require.NoError(t, err, "error getting optional meta")
 	require.Equal(t, om, om1, "unexpected empty versions")
-	require.Equal(t, "0: prune revocation log", om.String())
+	require.Equal(
+		t, "0: prune_revocation_log, 1: gc_decayed_log",
+		om1.String(),
+	)
 }
 
 // TestApplyOptionalVersions checks that the optional migration is applied as
 // expected based on the config.
+//
+// NOTE: Cannot be run in parallel because we alter the optionalVersions
+// global variable which could be used by other tests.
 func TestApplyOptionalVersions(t *testing.T) {
-	t.Parallel()
-
 	db, err := MakeTestDB(t)
 	require.NoError(t, err)
 
-	// Overwrite the migration function so we can count how many times the
-	// migration has happened.
-	migrateCount := 0
-	optionalVersions[0].migration = func(_ kvdb.Backend,
+	// migrateCount is the number of migrations that have been run. It
+	// counts the number of times a migration function is called.
+	var migrateCount int
+
+	// Modify all migrations to track their execution.
+	for i := range optionalVersions {
+		optionalVersions[i].migration = func(_ kvdb.Backend,
 			_ MigrationConfig) error {
 
 			migrateCount++
+
 			return nil
 		}
+	}
 
-	// Test that when the flag is false, no migration happens.
-	cfg := OptionalMiragtionConfig{}
+	// All migrations are disabled by default.
+	cfg := NewOptionalMiragtionConfig()
+
+	// Run the optional migrations.
 	err = db.applyOptionalVersions(cfg)
 	require.NoError(t, err, "failed to apply optional migration")
 	require.Equal(t, 0, migrateCount, "expected no migration")
@@ -536,13 +548,18 @@ func TestApplyOptionalVersions(t *testing.T) {
 	// Check the optional meta is not updated.
 	om, err := db.fetchOptionalMeta()
 	require.NoError(t, err, "error getting optional meta")
-	require.Empty(t, om.Versions, "expected empty versions")
 
-	// Test that when specified, the optional migration is applied.
-	cfg.PruneRevocationLog = true
+	// Enable all optional migrations.
+	for i := range cfg.MigrationFlags {
+		cfg.MigrationFlags[i] = true
+	}
+
 	err = db.applyOptionalVersions(cfg)
 	require.NoError(t, err, "failed to apply optional migration")
-	require.Equal(t, 1, migrateCount, "expected migration")
+	require.Equal(
+		t, len(optionalVersions), migrateCount,
+		"expected all migrations to be run",
+	)
 
 	// Fetch the updated optional meta.
 	om, err = db.fetchOptionalMeta()
@@ -552,16 +569,20 @@ func TestApplyOptionalVersions(t *testing.T) {
 	omExpected := &OptionalMeta{
 		Versions: map[uint64]string{
 			0: optionalVersions[0].name,
+			1: optionalVersions[1].name,
 		},
 	}
 	require.Equal(t, omExpected, om, "unexpected empty versions")
 
-	// Test that though specified, the optional migration is not run since
-	// it's already been applied.
-	cfg.PruneRevocationLog = true
+	// We make sure running the migrations again does not call the
+	// migrations again because the meta data should signal that they have
+	// already been run.
 	err = db.applyOptionalVersions(cfg)
 	require.NoError(t, err, "failed to apply optional migration")
-	require.Equal(t, 1, migrateCount, "expected no migration")
+	require.Equal(
+		t, len(optionalVersions), migrateCount,
+		"expected all migrations to be run",
+	)
 }
 
 // TestFetchMeta tests that the FetchMeta returns the latest DB version for a

channeldb/migration34/log.go

@@ -0,0 +1,14 @@
+package migration34
+
+import (
+	"github.com/btcsuite/btclog/v2"
+)
+
+// log is a logger that is initialized as disabled.  This means the package will
+// not perform any logging by default until a logger is set.
+var log = btclog.Disabled
+
+// UseLogger uses a specified Logger to output package logging info.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

channeldb/migration34/migration.go

@@ -0,0 +1,76 @@
+package migration34
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/lightningnetwork/lnd/kvdb"
+)
+
+// Migration34 is an optional migration that garbage collects the decayed log
+// in particular the `batch-replay` bucket. However we did choose to use an
+// optional migration which defaults to true because the decayed log db is
+// separate from the channeldb and if we would have implemented it as a
+// required migration, then it would have required a bigger change to the
+// codebase.
+//
+// Most of the decayed log db will shrink significantly after this migration
+// because the other bucket called `shared-secrets` is garbage collected
+// continuously and the `batch-replay` bucket will be deleted.
+
+var (
+	// batchReplayBucket is a bucket that maps batch identifiers to
+	// serialized ReplaySets. This is used to give idempotency in the event
+	// that a batch is processed more than once.
+	batchReplayBucket = []byte("batch-replay")
+)
+
+// MigrationConfig is the interface for the migration configuration.
+type MigrationConfig interface {
+	GetDecayedLog() kvdb.Backend
+}
+
+// MigrationConfigImpl is the implementation of the migration configuration.
+type MigrationConfigImpl struct {
+	DecayedLog kvdb.Backend
+}
+
+// GetDecayedLog returns the decayed log backend.
+func (c *MigrationConfigImpl) GetDecayedLog() kvdb.Backend {
+	return c.DecayedLog
+}
+
+// MigrateDecayedLog migrates the decayed log. The migration deletes the
+// `batch-replay` bucket, which is no longer used.
+//
+// NOTE: This migration is idempotent. If the bucket does not exist, then this
+// migration is a no-op.
+func MigrateDecayedLog(db kvdb.Backend, cfg MigrationConfig) error {
+	decayedLog := cfg.GetDecayedLog()
+
+	// Make sure we have a reference to the decayed log.
+	if decayedLog == nil {
+		return fmt.Errorf("decayed log backend is not available")
+	}
+
+	log.Info("Migrating decayed log...")
+	err := decayedLog.Update(func(tx kvdb.RwTx) error {
+		err := tx.DeleteTopLevelBucket(batchReplayBucket)
+		if err != nil && !errors.Is(err, kvdb.ErrBucketNotFound) {
+			return fmt.Errorf("deleting top level bucket %s: %w",
+				batchReplayBucket, err)
+		}
+
+		log.Debugf("top level bucket %s deleted", batchReplayBucket)
+
+		return nil
+	}, func() {})
+
+	if err != nil {
+		return fmt.Errorf("failed to migrate decayed log: %w", err)
+	}
+
+	log.Info("Decayed log migrated successfully")
+
+	return nil
+}

channeldb/options.go

@@ -2,6 +2,7 @@ package channeldb
 
 import (
 	"github.com/lightningnetwork/lnd/clock"
+	"github.com/lightningnetwork/lnd/kvdb"
 )
 
 const (
@@ -25,9 +26,26 @@ const (
 // OptionalMiragtionConfig defines the flags used to signal whether a
 // particular migration needs to be applied.
 type OptionalMiragtionConfig struct {
-	// PruneRevocationLog specifies that the revocation log migration needs
-	// to be applied.
-	PruneRevocationLog bool
+	// MigrationFlags is an array of booleans indicating which optional
+	// migrations should be run. The index in the array corresponds to the
+	// migration number in optionalVersions.
+	MigrationFlags []bool
+
+	// DecayedLog is a reference to the decayed log database. The channeldb
+	// is inherently part of the optional migration flow so there is no need
+	// to specify it here. The DecayedLog is a separate database in case the
+	// kvdb backend is set to `bbolt`. And also for the kvdb SQL backend
+	// case it is a separate table therefore we need to reference it here
+	// as well to use the right query to access the decayed log.
+	DecayedLog kvdb.Backend
+}
+
+// NewOptionalMiragtionConfig creates a new OptionalMiragtionConfig with the
+// default migration flags.
+func NewOptionalMiragtionConfig() OptionalMiragtionConfig {
+	return OptionalMiragtionConfig{
+		MigrationFlags: make([]bool, len(optionalVersions)),
+	}
 }
 
 // Options holds parameters for tuning and customizing a channeldb.DB.
@@ -62,7 +80,7 @@ type Options struct {
 // DefaultOptions returns an Options populated with default values.
 func DefaultOptions() Options {
 	return Options{
-		OptionalMiragtionConfig: OptionalMiragtionConfig{},
+		OptionalMiragtionConfig: NewOptionalMiragtionConfig(),
 		NoMigration:             false,
 		clock:                   clock.NewDefaultClock(),
 	}
@@ -124,6 +142,24 @@ func OptionStoreFinalHtlcResolutions(
 // revocation logs needs to be applied or not.
 func OptionPruneRevocationLog(prune bool) OptionModifier {
 	return func(o *Options) {
-		o.OptionalMiragtionConfig.PruneRevocationLog = prune
+		o.OptionalMiragtionConfig.MigrationFlags[0] = prune
+	}
+}
+
+// OptionWithDecayedLogDB sets the decayed log database reference which might
+// be used for some migrations because generally we only touch the channeldb
+// databases in the migrations, this is a way to allow also access to the
+// decayed log database.
+func OptionWithDecayedLogDB(decayedLog kvdb.Backend) OptionModifier {
+	return func(o *Options) {
+		o.OptionalMiragtionConfig.DecayedLog = decayedLog
+	}
+}
+
+// OptionGcDecayedLog specifies whether the decayed log migration has to
+// take place.
+func OptionGcDecayedLog(noGc bool) OptionModifier {
+	return func(o *Options) {
+		o.OptionalMiragtionConfig.MigrationFlags[1] = !noGc
 	}
 }

config_builder.go

@@ -1073,6 +1073,8 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 		),
 		channeldb.OptionPruneRevocationLog(cfg.DB.PruneRevocation),
 		channeldb.OptionNoRevLogAmtData(cfg.DB.NoRevLogAmtData),
+		channeldb.OptionGcDecayedLog(cfg.DB.NoGcDecayedLog),
+		channeldb.OptionWithDecayedLogDB(dbs.DecayedLogDB),
 	}
 
 	// Otherwise, we'll open two instances, one for the state we only need

docs/release-notes/release-notes-0.19.2.md

@@ -51,6 +51,9 @@
 
 ## Code Health
 
+- [Add Optional Migration](https://github.com/lightningnetwork/lnd/pull/9945)
+  which garbage collects the `decayed log` also known as `sphinxreplay.db`.
+
 ## Breaking Changes
 
 ## Performance Improvements

lncfg/db.go

@@ -95,6 +95,8 @@ type DB struct {
 	PruneRevocation bool `long:"prune-revocation" description:"Run the optional migration that prunes the revocation logs to save disk space."`
 
 	NoRevLogAmtData bool `long:"no-rev-log-amt-data" description:"If set, the to-local and to-remote output amounts of revoked commitment transactions will not be stored in the revocation log. Note that once this data is lost, a watchtower client will not be able to back up the revoked state."`
+
+	NoGcDecayedLog bool `long:"no-gc-decayed-log" description:"Do not run the optional migration that garbage collects the decayed log to save disk space."`
 }
 
 // DefaultDB creates and returns a new default DB config.

sample-lnd.conf

@@ -1476,6 +1476,11 @@
 ; channels prior to lnd@v0.15.0.
 ; db.prune-revocation=false
 
+; Specify whether the optional migration for garbage collecting the decayed
+; sphinx logs should be applied. By default, the decayed log will be garbage
+; collected.
+; db.no-gc-decayed-log=false
+
 ; If set to true, then the to-local and to-remote output amount data of revoked
 ; commitment transactions will not be stored in the revocation log. Note that
 ; this flag can only be set if --wtclient.active is not set. It is not