From 55746e427eba214b49558ae2c248cc62f57777c7 Mon Sep 17 00:00:00 2001 From: yyforyongyu Date: Wed, 18 May 2022 19:44:17 +0800 Subject: [PATCH] channeldb+migration26: migrate balance fields into tlv records --- channeldb/db.go | 7 + channeldb/migration26/channel.go | 299 ++++++++++++++++++++++++ channeldb/migration26/log.go | 14 ++ channeldb/migration26/migration.go | 147 ++++++++++++ channeldb/migration26/migration_test.go | 164 +++++++++++++ 5 files changed, 631 insertions(+) create mode 100644 channeldb/migration26/channel.go create mode 100644 channeldb/migration26/log.go create mode 100644 channeldb/migration26/migration.go create mode 100644 channeldb/migration26/migration_test.go diff --git a/channeldb/db.go b/channeldb/db.go index a1979b7d0..e97b96dda 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -20,6 +20,7 @@ import ( "github.com/lightningnetwork/lnd/channeldb/migration23" "github.com/lightningnetwork/lnd/channeldb/migration24" "github.com/lightningnetwork/lnd/channeldb/migration25" + "github.com/lightningnetwork/lnd/channeldb/migration26" "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11" "github.com/lightningnetwork/lnd/clock" "github.com/lightningnetwork/lnd/kvdb" @@ -212,6 +213,12 @@ var ( number: 25, migration: migration25.MigrateInitialBalances, }, + { + // Migrate the initial local/remote balance fields into + // tlv records. + number: 26, + migration: migration26.MigrateBalancesToTlvRecords, + }, } // Big endian is the preferred byte order, due to cursor scans over diff --git a/channeldb/migration26/channel.go b/channeldb/migration26/channel.go new file mode 100644 index 000000000..f569ec3a1 --- /dev/null +++ b/channeldb/migration26/channel.go @@ -0,0 +1,299 @@ +package migration26 + +import ( + "bytes" + "fmt" + + lnwire "github.com/lightningnetwork/lnd/channeldb/migration/lnwire21" + mig25 "github.com/lightningnetwork/lnd/channeldb/migration25" + mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11" + "github.com/lightningnetwork/lnd/kvdb" + "github.com/lightningnetwork/lnd/tlv" +) + +const ( + // A tlv type definition used to serialize and deserialize a KeyLocator + // from the database. + keyLocType tlv.Type = 1 + + // A tlv type used to serialize and deserialize the + // `InitialLocalBalance` field. + initialLocalBalanceType tlv.Type = 2 + + // A tlv type used to serialize and deserialize the + // `InitialRemoteBalance` field. + initialRemoteBalanceType tlv.Type = 3 +) + +var ( + // chanInfoKey can be accessed within the bucket for a channel + // (identified by its chanPoint). This key stores all the static + // information for a channel which is decided at the end of the + // funding flow. + chanInfoKey = []byte("chan-info-key") + + // localUpfrontShutdownKey can be accessed within the bucket for a + // channel (identified by its chanPoint). This key stores an optional + // upfront shutdown script for the local peer. + localUpfrontShutdownKey = []byte("local-upfront-shutdown-key") + + // remoteUpfrontShutdownKey can be accessed within the bucket for a + // channel (identified by its chanPoint). This key stores an optional + // upfront shutdown script for the remote peer. + remoteUpfrontShutdownKey = []byte("remote-upfront-shutdown-key") + + // lastWasRevokeKey is a key that stores true when the last update we + // sent was a revocation and false when it was a commitment signature. + // This is nil in the case of new channels with no updates exchanged. + lastWasRevokeKey = []byte("last-was-revoke") + + // ErrNoChanInfoFound is returned when a particular channel does not + // have any channels state. + ErrNoChanInfoFound = fmt.Errorf("no chan info found") + + // ErrNoPastDeltas is returned when the channel delta bucket hasn't been + // created. + ErrNoPastDeltas = fmt.Errorf("channel has no recorded deltas") + + // ErrLogEntryNotFound is returned when we cannot find a log entry at + // the height requested in the revocation log. + ErrLogEntryNotFound = fmt.Errorf("log entry not found") + + // ErrNoCommitmentsFound is returned when a channel has not set + // commitment states. + ErrNoCommitmentsFound = fmt.Errorf("no commitments found") +) + +// OpenChannel embeds a mig25.OpenChannel with the extra update-to-date +// serialization and deserialization methods. +// +// NOTE: doesn't have the Packager field as it's not used in current migration. +type OpenChannel struct { + mig25.OpenChannel + + // chanStatus is the current status of this channel. If it is not in + // the state Default, it should not be used for forwarding payments. + chanStatus mig25.ChannelStatus +} + +// FetchChanInfo deserializes the channel info based on the legacy boolean. +// After migration25, the legacy format would have the fields +// `InitialLocalBalance` and `InitialRemoteBalance` directly encoded as bytes. +// For the new format, they will be put inside a tlv stream. +func FetchChanInfo(chanBucket kvdb.RBucket, c *OpenChannel, legacy bool) error { + infoBytes := chanBucket.Get(chanInfoKey) + if infoBytes == nil { + return ErrNoChanInfoFound + } + r := bytes.NewReader(infoBytes) + + var ( + chanType mig.ChannelType + chanStatus mig.ChannelStatus + ) + + if err := mig.ReadElements(r, + &chanType, &c.ChainHash, &c.FundingOutpoint, + &c.ShortChannelID, &c.IsPending, &c.IsInitiator, + &chanStatus, &c.FundingBroadcastHeight, + &c.NumConfsRequired, &c.ChannelFlags, + &c.IdentityPub, &c.Capacity, &c.TotalMSatSent, + &c.TotalMSatReceived, + ); err != nil { + return err + } + + c.ChanType = mig25.ChannelType(chanType) + c.chanStatus = mig25.ChannelStatus(chanStatus) + + // If this is the legacy format, we need to read the extra two new + // fields. + if legacy { + if err := mig.ReadElements(r, + &c.InitialLocalBalance, &c.InitialRemoteBalance, + ); err != nil { + return err + } + } + + // For single funder channels that we initiated and have the funding + // transaction to, read the funding txn. + if c.FundingTxPresent() { + if err := mig.ReadElement(r, &c.FundingTxn); err != nil { + return err + } + } + + if err := mig.ReadChanConfig(r, &c.LocalChanCfg); err != nil { + return err + } + if err := mig.ReadChanConfig(r, &c.RemoteChanCfg); err != nil { + return err + } + + // Retrieve the boolean stored under lastWasRevokeKey. + lastWasRevokeBytes := chanBucket.Get(lastWasRevokeKey) + if lastWasRevokeBytes == nil { + // If nothing has been stored under this key, we store false in + // the OpenChannel struct. + c.LastWasRevoke = false + } else { + // Otherwise, read the value into the LastWasRevoke field. + revokeReader := bytes.NewReader(lastWasRevokeBytes) + err := mig.ReadElements(revokeReader, &c.LastWasRevoke) + if err != nil { + return err + } + } + + // Make the tlv stream based on the legacy param. + var ( + ts *tlv.Stream + err error + localBalance uint64 + remoteBalance uint64 + ) + + keyLocRecord := mig25.MakeKeyLocRecord( + keyLocType, &c.RevocationKeyLocator, + ) + + // If it's legacy, create the stream with a single tlv record. + if legacy { + ts, err = tlv.NewStream(keyLocRecord) + } else { + // Otherwise, for the new format, we will encode the balance + // fields in the tlv stream too. + ts, err = tlv.NewStream( + keyLocRecord, + tlv.MakePrimitiveRecord( + initialLocalBalanceType, &localBalance, + ), + tlv.MakePrimitiveRecord( + initialRemoteBalanceType, &remoteBalance, + ), + ) + } + if err != nil { + return err + } + + if err := ts.Decode(r); err != nil { + return err + } + + // For the new format, attach the balance fields. + if !legacy { + c.InitialLocalBalance = lnwire.MilliSatoshi(localBalance) + c.InitialRemoteBalance = lnwire.MilliSatoshi(remoteBalance) + } + + // Finally, read the optional shutdown scripts. + if err := mig25.GetOptionalUpfrontShutdownScript( + chanBucket, localUpfrontShutdownKey, &c.LocalShutdownScript, + ); err != nil { + return err + } + + return mig25.GetOptionalUpfrontShutdownScript( + chanBucket, remoteUpfrontShutdownKey, &c.RemoteShutdownScript, + ) +} + +// MakeTlvStream creates a tlv stream based on whether we are deadling with +// legacy format or not. For the legacy format, we have a single record in the +// stream. For the new format, we have the extra balance records. +func MakeTlvStream(c *OpenChannel, legacy bool) (*tlv.Stream, error) { + keyLocRecord := mig25.MakeKeyLocRecord( + keyLocType, &c.RevocationKeyLocator, + ) + + // If it's legacy, return the stream with a single tlv record. + if legacy { + return tlv.NewStream(keyLocRecord) + } + + // Otherwise, for the new format, we will encode the balance fields in + // the tlv stream too. + localBalance := uint64(c.InitialLocalBalance) + remoteBalance := uint64(c.InitialRemoteBalance) + + // Create the tlv stream. + return tlv.NewStream( + keyLocRecord, + tlv.MakePrimitiveRecord( + initialLocalBalanceType, &localBalance, + ), + tlv.MakePrimitiveRecord( + initialRemoteBalanceType, &remoteBalance, + ), + ) +} + +// PutChanInfo serializes the channel info based on the legacy boolean. After +// migration25, the legacy format would have the fields `InitialLocalBalance` +// and `InitialRemoteBalance` directly encoded as bytes. For the new format, +// they will be put inside a tlv stream. +func PutChanInfo(chanBucket kvdb.RwBucket, c *OpenChannel, legacy bool) error { + var w bytes.Buffer + if err := mig.WriteElements(&w, + mig.ChannelType(c.ChanType), c.ChainHash, c.FundingOutpoint, + c.ShortChannelID, c.IsPending, c.IsInitiator, + mig.ChannelStatus(c.chanStatus), c.FundingBroadcastHeight, + c.NumConfsRequired, c.ChannelFlags, + c.IdentityPub, c.Capacity, c.TotalMSatSent, + c.TotalMSatReceived, + ); err != nil { + return err + } + + // If this is legacy format, we need to write the extra two fields. + if legacy { + if err := mig.WriteElements(&w, + c.InitialLocalBalance, c.InitialRemoteBalance, + ); err != nil { + return err + } + } + + // For single funder channels that we initiated, and we have the + // funding transaction, then write the funding txn. + if c.FundingTxPresent() { + if err := mig.WriteElement(&w, c.FundingTxn); err != nil { + return err + } + } + + if err := mig.WriteChanConfig(&w, &c.LocalChanCfg); err != nil { + return err + } + if err := mig.WriteChanConfig(&w, &c.RemoteChanCfg); err != nil { + return err + } + + // Make the tlv stream based on the legacy param. + tlvStream, err := MakeTlvStream(c, legacy) + if err != nil { + return err + } + + if err := tlvStream.Encode(&w); err != nil { + return err + } + + if err := chanBucket.Put(chanInfoKey, w.Bytes()); err != nil { + return err + } + + // Finally, add optional shutdown scripts for the local and remote peer + // if they are present. + if err := mig25.PutOptionalUpfrontShutdownScript( + chanBucket, localUpfrontShutdownKey, c.LocalShutdownScript, + ); err != nil { + return err + } + + return mig25.PutOptionalUpfrontShutdownScript( + chanBucket, remoteUpfrontShutdownKey, c.RemoteShutdownScript, + ) +} diff --git a/channeldb/migration26/log.go b/channeldb/migration26/log.go new file mode 100644 index 000000000..b7326a313 --- /dev/null +++ b/channeldb/migration26/log.go @@ -0,0 +1,14 @@ +package migration26 + +import ( + "github.com/btcsuite/btclog" +) + +// 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 +} diff --git a/channeldb/migration26/migration.go b/channeldb/migration26/migration.go new file mode 100644 index 000000000..b4f2e3c05 --- /dev/null +++ b/channeldb/migration26/migration.go @@ -0,0 +1,147 @@ +package migration26 + +import ( + "fmt" + + mig25 "github.com/lightningnetwork/lnd/channeldb/migration25" + + "github.com/lightningnetwork/lnd/kvdb" +) + +var ( + // openChanBucket stores all the currently open channels. This bucket + // has a second, nested bucket which is keyed by a node's ID. Within + // that node ID bucket, all attributes required to track, update, and + // close a channel are stored. + openChannelBucket = []byte("open-chan-bucket") + + // ErrNoChanDBExists is returned when a channel bucket hasn't been + // created. + ErrNoChanDBExists = fmt.Errorf("channel db has not yet been created") + + // ErrNoActiveChannels is returned when there is no active (open) + // channels within the database. + ErrNoActiveChannels = fmt.Errorf("no active channels exist") + + // ErrChannelNotFound is returned when we attempt to locate a channel + // for a specific chain, but it is not found. + ErrChannelNotFound = fmt.Errorf("channel not found") +) + +// MigrateBalancesToTlvRecords migrates the balance fields into tlv records. It +// does so by first reading a list of open channels, then rewriting the channel +// info with the updated tlv stream. +func MigrateBalancesToTlvRecords(tx kvdb.RwTx) error { + log.Infof("Migrating local and remote balances into tlv records...") + + openChanBucket := tx.ReadWriteBucket(openChannelBucket) + + // If no bucket is found, we can exit early. + if openChanBucket == nil { + return nil + } + + // Read a list of open channels. + channels, err := findOpenChannels(openChanBucket) + if err != nil { + return err + } + + // Migrate the balances. + for _, c := range channels { + if err := migrateBalances(tx, c); err != nil { + return err + } + } + + return err +} + +// findOpenChannels finds all open channels. +func findOpenChannels(openChanBucket kvdb.RBucket) ([]*OpenChannel, error) { + channels := []*OpenChannel{} + + // readChannel is a helper closure that reads the channel info from the + // channel bucket. + readChannel := func(chainBucket kvdb.RBucket, cp []byte) error { + c := &OpenChannel{} + + // Read the sub-bucket level 3. + chanBucket := chainBucket.NestedReadBucket( + cp, + ) + if chanBucket == nil { + log.Errorf("unable to read bucket for chanPoint=%x", cp) + return nil + } + + // Get the old channel info. + if err := FetchChanInfo(chanBucket, c, true); err != nil { + return fmt.Errorf("unable to fetch chan info: %v", err) + } + + channels = append(channels, c) + + return nil + } + + // Iterate the root bucket. + err := openChanBucket.ForEach(func(nodePub, v []byte) error { + // Ensure that this is a key the same size as a pubkey, and + // also that it leads directly to a bucket. + if len(nodePub) != 33 || v != nil { + return nil + } + + // Read the sub-bucket level 1. + nodeChanBucket := openChanBucket.NestedReadBucket(nodePub) + if nodeChanBucket == nil { + log.Errorf("no bucket for node %x", nodePub) + return nil + } + + // Iterate the bucket. + return nodeChanBucket.ForEach(func(chainHash, _ []byte) error { + // Read the sub-bucket level 2. + chainBucket := nodeChanBucket.NestedReadBucket( + chainHash, + ) + if chainBucket == nil { + log.Errorf("unable to read bucket for chain=%x", + chainHash) + return nil + } + + // Iterate the bucket. + return chainBucket.ForEach(func(cp, _ []byte) error { + return readChannel(chainBucket, cp) + }) + }) + }) + + if err != nil { + return nil, err + } + + return channels, nil +} + +// migrateBalances creates a new tlv stream which adds two more records to hold +// the balances info. +func migrateBalances(tx kvdb.RwTx, c *OpenChannel) error { + // Get the bucket. + chanBucket, err := mig25.FetchChanBucket(tx, &c.OpenChannel) + if err != nil { + return err + } + + // Update the channel info. There isn't much to do here as the + // `PutChanInfo` will read the values from `c.InitialLocalBalance` and + // `c.InitialRemoteBalance` then create the new tlv stream as + // requested. + if err := PutChanInfo(chanBucket, c, false); err != nil { + return fmt.Errorf("unable to put chan info: %v", err) + } + + return nil +} diff --git a/channeldb/migration26/migration_test.go b/channeldb/migration26/migration_test.go new file mode 100644 index 000000000..a775386f7 --- /dev/null +++ b/channeldb/migration26/migration_test.go @@ -0,0 +1,164 @@ +package migration26 + +import ( + "fmt" + "testing" + + "github.com/btcsuite/btcd/btcec/v2" + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + lnwire "github.com/lightningnetwork/lnd/channeldb/migration/lnwire21" + mig25 "github.com/lightningnetwork/lnd/channeldb/migration25" + mig "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11" + "github.com/lightningnetwork/lnd/channeldb/migtest" + "github.com/lightningnetwork/lnd/kvdb" +) + +var ( + // Create dummy values to be stored in db. + dummyPrivKey, _ = btcec.NewPrivateKey() + dummyPubKey = dummyPrivKey.PubKey() + dummyOp = wire.OutPoint{ + Hash: chainhash.Hash{}, + Index: 9, + } + + // ourAmt and theirAmt are the initial balances found in the local + // channel commitment at height 0. + testOurAmt = lnwire.MilliSatoshi(500_000) + testTheirAmt = lnwire.MilliSatoshi(1000_000) + + // testChannel is used to test the balance fields are correctly set. + testChannel = &OpenChannel{ + OpenChannel: mig25.OpenChannel{ + OpenChannel: mig.OpenChannel{ + IdentityPub: dummyPubKey, + FundingOutpoint: dummyOp, + }, + }, + } +) + +// TestMigrateBalancesToTlvRecords checks that the initial balances fields are +// saved using the tlv records. +func TestMigrateBalancesToTlvRecords(t *testing.T) { + testCases := []struct { + name string + ourAmt lnwire.MilliSatoshi + theirAmt lnwire.MilliSatoshi + beforeMigrationFunc func(kvdb.RwTx) error + afterMigrationFunc func(kvdb.RwTx) error + shouldFail bool + }{ + { + // Test when both balance fields are non-zero. + name: "non-zero local and remote", + ourAmt: testOurAmt, + theirAmt: testTheirAmt, + beforeMigrationFunc: genBeforeMigration(testChannel), + afterMigrationFunc: genAfterMigration(testChannel), + }, + { + // Test when local balance is non-zero. + name: "non-zero local balance", + ourAmt: testOurAmt, + theirAmt: 0, + beforeMigrationFunc: genBeforeMigration(testChannel), + afterMigrationFunc: genAfterMigration(testChannel), + }, + { + // Test when remote balance is non-zero. + name: "non-zero remote balance", + ourAmt: 0, + theirAmt: testTheirAmt, + beforeMigrationFunc: genBeforeMigration(testChannel), + afterMigrationFunc: genAfterMigration(testChannel), + }, + { + // Test when both balance fields are zero. + name: "zero local and remote", + ourAmt: 0, + theirAmt: 0, + beforeMigrationFunc: genBeforeMigration(testChannel), + afterMigrationFunc: genAfterMigration(testChannel), + }, + } + + for _, tc := range testCases { + tc := tc + + // Before running the test, set the balance fields based on the + // test params. + testChannel.InitialLocalBalance = tc.ourAmt + testChannel.InitialRemoteBalance = tc.theirAmt + + t.Run(tc.name, func(t *testing.T) { + migtest.ApplyMigration( + t, + tc.beforeMigrationFunc, + tc.afterMigrationFunc, + MigrateBalancesToTlvRecords, + tc.shouldFail, + ) + }) + } +} + +func genBeforeMigration(c *OpenChannel) func(kvdb.RwTx) error { + return func(tx kvdb.RwTx) error { + // Create the channel bucket. + chanBucket, err := mig25.CreateChanBucket(tx, &c.OpenChannel) + if err != nil { + return err + } + + // Save the channel info using legacy format. + if err := PutChanInfo(chanBucket, c, true); err != nil { + return err + } + + return nil + } +} + +func genAfterMigration(c *OpenChannel) func(kvdb.RwTx) error { + return func(tx kvdb.RwTx) error { + chanBucket, err := mig25.FetchChanBucket(tx, &c.OpenChannel) + if err != nil { + return err + } + + newChan := &OpenChannel{} + + // Fetch the channel info using the new format. + err = FetchChanInfo(chanBucket, newChan, false) + if err != nil { + return err + } + + // Check our initial amount is correct. + if newChan.InitialLocalBalance != c.InitialLocalBalance { + return fmt.Errorf("wrong local balance, got %d, "+ + "want %d", newChan.InitialLocalBalance, + c.InitialLocalBalance) + } + + // Check their initial amount is correct. + if newChan.InitialRemoteBalance != c.InitialRemoteBalance { + return fmt.Errorf("wrong remote balance, got %d, "+ + "want %d", newChan.InitialRemoteBalance, + c.InitialRemoteBalance) + } + + // We also check the relevant channel info fields stay the + // same. + if !newChan.IdentityPub.IsEqual(dummyPubKey) { + return fmt.Errorf("wrong IdentityPub") + } + if newChan.FundingOutpoint != dummyOp { + return fmt.Errorf("wrong FundingOutpoint") + } + + return nil + } +}