channeldb: Add dcrlnd migration 01

This migration fixes the prior migration 20, introduced when porting the
upstream code.

That migration erroneously encoded outpoints without the Tree field that
exists in decred code, thus rendering the index incorrect.

The new migration corrects the issue by assuming the tree of every entry
is zero, which is true because the channels can only reside in regular
(as opposed to stake) transactions.

channeldb/db.go

@@ -1262,6 +1262,12 @@ func (d *DB) syncVersions(versions []version) error {
 		// In dry-run mode, return an error to prevent the transaction
 		// from committing.
 		if d.dryRun {
+			// In dry run mode, also attempt dcrlnd migrations
+			// before stopping.
+			if err := d.syncDcrlndDBVersions(tx); err != nil {
+				return err
+			}
+
 			return ErrDryRunMigrationOK
 		}
 

channeldb/db_test.go

@@ -34,7 +34,7 @@ func TestOpenWithCreate(t *testing.T) {
 
 	// Next, open thereby creating channeldb for the first time.
 	dbPath := filepath.Join(tempDirName, "cdb")
-	backend, cleanup, err := kvdb.GetTestBackend(dbPath, "cdb")
+	backend, cleanup, err := kvdb.GetTestBackend(dbPath, dbName)
 	if err != nil {
 		t.Fatalf("unable to get test db backend: %v", err)
 	}

channeldb/dcrdb.go

@@ -1,10 +1,118 @@
 package channeldb
 
-import "github.com/decred/dcrlnd/kvdb"
+import (
+	dcrmigration01 "github.com/decred/dcrlnd/channeldb/dcrmigrations/migration01"
+	"github.com/decred/dcrlnd/kvdb"
+)
+
+var (
+	dcrMetaBucket = []byte("dcrlnd_metadata")
+)
+
+var (
+	// dcrDbVersions are the decred-only migrations.
+	dcrDbVersions = []version{
+		{number: 1, migration: dcrmigration01.FixMigration20},
+	}
+)
+
+// latestDcrDbVersion returns the latest version of the decred-specific db
+// migrations.
+func latestDcrDbVersion() uint32 {
+	return dcrDbVersions[len(dcrDbVersions)-1].number
+}
+
+// applyDecredMigations applies the decred-only migrations.
+func (d *DB) applyDecredMigations(tx kvdb.RwTx, dbVersion uint32) error {
+	latestVersion := dcrDbVersions[len(dcrDbVersions)-1].number
+	log.Infof("Checking for decred-specicic migrations latest_version=%d, "+
+		"db_version=%d", latestVersion, dbVersion)
+
+	if latestVersion < dbVersion {
+		log.Errorf("Refusing to revert from decred db_version=%d to "+
+			"lower version=%d", dbVersion, latestVersion)
+		return ErrDBReversion
+	}
+
+	if latestVersion == dbVersion {
+		// Nothing to do.
+		return nil
+	}
+
+	log.Infof("Performing decred-specific database schema migration")
+
+	metaBucket := tx.ReadWriteBucket(dcrMetaBucket)
+	for _, mig := range dcrDbVersions {
+		if mig.migration == nil {
+			continue
+		}
+		if mig.number <= dbVersion {
+			continue
+		}
+
+		log.Infof("Applying migration #%d", mig.number)
+
+		if err := mig.migration(tx); err != nil {
+			log.Infof("Unable to apply migration #%d",
+				mig.number)
+			return err
+		}
+
+		// Save the new db version.
+		dbVersion = mig.number
+		err := metaBucket.Put(dbVersionKey, byteOrder.AppendUint32(nil, dbVersion))
+		if err != nil {
+			return err
+		}
+	}
+
+	// Stop if running in dry-run mode.
+	if d.dryRun {
+		return ErrDryRunMigrationOK
+	}
+
+	return nil
+}
+
+// syncDcrlndDBVersions performs the dcrlnd-specific db upgrades.
+func (d *DB) syncDcrlndDBVersions(tx kvdb.RwTx) error {
+	// Read dcr-specific version.
+	var dbVersion uint32
+	bucket := tx.ReadWriteBucket(dcrMetaBucket)
+	if bucket == nil {
+		// Filled meta bucket but empty dcr-specific meta bucket.
+		// Create dcr one.
+		var err error
+		bucket, err = tx.CreateTopLevelBucket(dcrMetaBucket)
+		if err != nil {
+			return err
+		}
+
+		// If the global meta bucket is empty, it's a new db.
+		if tx.ReadBucket(metaBucket) == nil {
+			dbVersion = latestDcrDbVersion()
+		}
+
+		// dbVersion == 0.
+		bucket.Put(dbVersionKey, byteOrder.AppendUint32(nil, dbVersion))
+	} else {
+		v := bucket.Get(dbVersionKey)
+		if v != nil {
+			dbVersion = byteOrder.Uint32(v)
+		}
+	}
+
+	// Apply dcr-specific migrations.
+	return d.applyDecredMigations(tx, dbVersion)
+}
 
 // initDcrlndFeatures initializes features that are specific to dcrlnd.
-func (db *DB) initDcrlndFeatures() error {
-	return kvdb.Update(db, func(tx kvdb.RwTx) error {
+func (d *DB) initDcrlndFeatures() error {
+	return kvdb.Update(d, func(tx kvdb.RwTx) error {
+		if err := d.syncDcrlndDBVersions(tx); err != nil {
+			return err
+		}
+
 		// If the inflight payments index bucket doesn't exist,
 		// initialize it.
 		indexBucket := tx.ReadWriteBucket(paymentsInflightIndexBucket)

channeldb/dcrmigrations/migration01/codec.go

@@ -0,0 +1,48 @@
+package dcrmigration01
+
+import (
+	"encoding/binary"
+	"io"
+
+	"github.com/decred/dcrd/wire"
+)
+
+// readMig20Outpoint reads an outpoint that was stored by the migration20.
+func readMig20Outpoint(r io.Reader, o *wire.OutPoint) error {
+	if _, err := io.ReadFull(r, o.Hash[:]); err != nil {
+		return err
+	}
+	if err := binary.Read(r, byteOrder, &o.Index); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// writeMig20Outpoint writes an outpoint from the passed writer.
+func writeMig20Outpoint(w io.Writer, o *wire.OutPoint) error {
+	if _, err := w.Write(o.Hash[:]); err != nil {
+		return err
+	}
+	if err := binary.Write(w, byteOrder, o.Index); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// writeOkOutpoint writes an outpoint with the correct format to the passed
+// writer.
+func writeOkOutpoint(w io.Writer, o *wire.OutPoint) error {
+	if _, err := w.Write(o.Hash[:]); err != nil {
+		return err
+	}
+	if err := binary.Write(w, byteOrder, o.Index); err != nil {
+		return err
+	}
+	if err := binary.Write(w, byteOrder, o.Tree); err != nil {
+		return err
+	}
+
+	return nil
+}

channeldb/dcrmigrations/migration01/dcrmigration01.go

@@ -0,0 +1,77 @@
+package dcrmigration01
+
+import (
+	"bytes"
+	"encoding/binary"
+
+	"github.com/decred/dcrd/wire"
+	"github.com/decred/dcrlnd/kvdb"
+)
+
+var (
+	byteOrder = binary.BigEndian
+
+	// outpointBucket is an index mapping outpoints to a tlv
+	// stream of channel data.
+	outpointBucket = []byte("outpoint-bucket")
+)
+
+// FixMigration20 fixes a version of the version 20 that had a wrong
+// implementation for the writeOutpoint codec function. This assumes that
+// migration20 was executed and now needs to be fixed.
+func FixMigration20(tx kvdb.RwTx) error {
+	// Get the target bucket.
+	bucket := tx.ReadWriteBucket(outpointBucket)
+
+	// Collect the data that needs migration.
+	var keys []*wire.OutPoint
+	values := map[*wire.OutPoint][]byte{}
+	err := bucket.ForEach(func(k, v []byte) error {
+		op := new(wire.OutPoint)
+		r := bytes.NewReader(k)
+		if err := readMig20Outpoint(r, op); err != nil {
+			return err
+		}
+
+		keys = append(keys, op)
+		switch {
+		case v == nil:
+			values[op] = nil
+		case len(v) == 0:
+			values[op] = []byte{}
+		default:
+			values[op] = append([]byte(nil), v...)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	log.Infof("Migrating %d entries", len(keys))
+
+	for _, op := range keys {
+		log.Debugf("Migrating outpoint %s", op)
+
+		var oldOpBuf bytes.Buffer
+		if err := writeMig20Outpoint(&oldOpBuf, op); err != nil {
+			return err
+		}
+
+		if err := bucket.Delete(oldOpBuf.Bytes()); err != nil {
+			return err
+		}
+
+		var newOpBuf bytes.Buffer
+		if err := writeOkOutpoint(&newOpBuf, op); err != nil {
+			return err
+		}
+		value := values[op]
+		if err := bucket.Put(newOpBuf.Bytes(), value); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

channeldb/dcrmigrations/migration01/dcrmigration01_test.go

@@ -0,0 +1,45 @@
+package dcrmigration01
+
+import (
+	"testing"
+
+	"github.com/decred/dcrlnd/channeldb/migtest"
+	"github.com/decred/dcrlnd/kvdb"
+)
+
+var (
+	hexStr = migtest.Hex
+
+	tlvOutpointOpen   = hexStr("000100")
+	tlvOutpointClosed = hexStr("000101")
+
+	outpointMig20     = hexStr("81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce952d6c6c7")
+	outpointMig20_2   = hexStr("abb637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce952d6c6c7")
+	outpointDataMig20 = map[string]interface{}{
+		outpointMig20:   tlvOutpointOpen,
+		outpointMig20_2: tlvOutpointClosed,
+	}
+
+	outpointCorrect     = hexStr("81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce952d6c6c700")
+	outpointCorrect_2   = hexStr("abb637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce952d6c6c700")
+	outpointDataCorrect = map[string]interface{}{
+		outpointCorrect:   tlvOutpointOpen,
+		outpointCorrect_2: tlvOutpointClosed,
+	}
+)
+
+func TestFixMigration20(t *testing.T) {
+	// Prime the database with the results of migration20 (wrong outpoint
+	// key).
+	before := func(tx kvdb.RwTx) error {
+		return migtest.RestoreDB(tx, outpointBucket, outpointDataMig20)
+	}
+
+	// Double check the keys were migrated to use the correct serialization
+	// of outpoint.
+	after := func(tx kvdb.RwTx) error {
+		return migtest.VerifyDB(tx, outpointBucket, outpointDataCorrect)
+	}
+
+	migtest.ApplyMigration(t, before, after, FixMigration20, false)
+}

channeldb/dcrmigrations/migration01/log.go

@@ -0,0 +1,14 @@
+package dcrmigration01
+
+import (
+	"github.com/decred/slog"
+)
+
+// 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 = slog.Disabled
+
+// UseLogger uses a specified Logger to output package logging info.
+func UseLogger(logger slog.Logger) {
+	log = logger
+}

channeldb/log.go

@@ -2,6 +2,7 @@ package channeldb
 
 import (
 	"github.com/decred/dcrlnd/build"
+	dcrmigration01 "github.com/decred/dcrlnd/channeldb/dcrmigrations/migration01"
 	mig "github.com/decred/dcrlnd/channeldb/migration"
 	"github.com/decred/dcrlnd/channeldb/migration12"
 	"github.com/decred/dcrlnd/channeldb/migration13"
@@ -39,4 +40,6 @@ func UseLogger(logger slog.Logger) {
 	migration16.UseLogger(logger)
 	migration20.UseLogger(logger)
 	kvdb.UseLogger(logger)
+
+	dcrmigration01.UseLogger(logger)
 }