lnd.xprv/channeldb/migration21/migration.go

388 lines
10 KiB
Go
Raw Normal View History

package migration21
import (
"bytes"
"encoding/binary"
"fmt"
lnwire "github.com/lightningnetwork/lnd/channeldb/migration/lnwire21"
"github.com/lightningnetwork/lnd/channeldb/migration21/common"
"github.com/lightningnetwork/lnd/channeldb/migration21/current"
"github.com/lightningnetwork/lnd/channeldb/migration21/legacy"
2021-04-26 20:08:11 +03:00
"github.com/lightningnetwork/lnd/kvdb"
)
var (
byteOrder = binary.BigEndian
// 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.
//
// openChan -> nodeID -> chanPoint
//
// TODO(roasbeef): flesh out comment
openChannelBucket = []byte("open-chan-bucket")
// commitDiffKey stores the current pending commitment state we've
// extended to the remote party (if any). Each time we propose a new
// state, we store the information necessary to reconstruct this state
// from the prior commitment. This allows us to resync the remote party
// to their expected state in the case of message loss.
//
// TODO(roasbeef): rename to commit chain?
commitDiffKey = []byte("commit-diff-key")
// unsignedAckedUpdatesKey is an entry in the channel bucket that
// contains the remote updates that we have acked, but not yet signed
// for in one of our remote commits.
unsignedAckedUpdatesKey = []byte("unsigned-acked-updates-key")
// remoteUnsignedLocalUpdatesKey is an entry in the channel bucket that
// contains the local updates that the remote party has acked, but
// has not yet signed for in one of their local commits.
remoteUnsignedLocalUpdatesKey = []byte("remote-unsigned-local-updates-key")
// networkResultStoreBucketKey is used for the root level bucket that
// stores the network result for each payment ID.
networkResultStoreBucketKey = []byte("network-result-store-bucket")
// closedChannelBucket stores summarization information concerning
// previously open, but now closed channels.
closedChannelBucket = []byte("closed-chan-bucket")
// fwdPackagesKey is the root-level bucket that all forwarding packages
// are written. This bucket is further subdivided based on the short
// channel ID of each channel.
fwdPackagesKey = []byte("fwd-packages")
)
// MigrateDatabaseWireMessages performs a migration in all areas that we
// currently store wire messages without length prefixes. This includes the
// CommitDiff struct, ChannelCloseSummary, LogUpdates, and also the
// networkResult struct as well.
func MigrateDatabaseWireMessages(tx kvdb.RwTx) error {
// The migration will proceed in three phases: we'll need to update any
// pending commit diffs, then any unsigned acked updates for all open
// channels, then finally we'll need to update all the current
// stored network results for payments in the switch.
//
// In this phase, we'll migrate the open channel data.
if err := migrateOpenChanBucket(tx); err != nil {
return err
}
// Next, we'll update all the present close channel summaries as well.
if err := migrateCloseChanSummaries(tx); err != nil {
return err
}
// We'll migrate forwarding packages, which have log updates as part of
// their serialized data.
if err := migrateForwardingPackages(tx); err != nil {
return err
}
// Finally, we'll update the pending network results as well.
return migrateNetworkResults(tx)
}
func migrateOpenChanBucket(tx kvdb.RwTx) error {
openChanBucket := tx.ReadWriteBucket(openChannelBucket)
// If no bucket is found, we can exit early.
if openChanBucket == nil {
return nil
}
type channelPath struct {
nodePub []byte
chainHash []byte
chanPoint []byte
}
var channelPaths []channelPath
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
}
nodeChanBucket := openChanBucket.NestedReadBucket(nodePub)
if nodeChanBucket == nil {
return fmt.Errorf("no bucket for node %x", nodePub)
}
// The next layer down is all the chains that this node
// has channels on with us.
return nodeChanBucket.ForEach(func(chainHash, v []byte) error {
// If there's a value, it's not a bucket so
// ignore it.
if v != nil {
return nil
}
chainBucket := nodeChanBucket.NestedReadBucket(
chainHash,
)
if chainBucket == nil {
return fmt.Errorf("unable to read "+
"bucket for chain=%x", chainHash)
}
return chainBucket.ForEach(func(chanPoint, v []byte) error {
// If there's a value, it's not a bucket so
// ignore it.
if v != nil {
return nil
}
channelPaths = append(channelPaths, channelPath{
nodePub: nodePub,
chainHash: chainHash,
chanPoint: chanPoint,
})
return nil
})
})
})
if err != nil {
return err
}
// Now that we have all the paths of the channel we need to migrate,
// we'll update all the state in a distinct step to avoid weird
// behavior from modifying buckets in a ForEach statement.
for _, channelPath := range channelPaths {
// First, we'll extract it from the node's chain bucket.
nodeChanBucket := openChanBucket.NestedReadWriteBucket(
channelPath.nodePub,
)
chainBucket := nodeChanBucket.NestedReadWriteBucket(
channelPath.chainHash,
)
chanBucket := chainBucket.NestedReadWriteBucket(
channelPath.chanPoint,
)
// At this point, we have the channel bucket now, so we'll
// check to see if this channel has a pending commitment or
// not.
commitDiffBytes := chanBucket.Get(commitDiffKey)
if commitDiffBytes != nil {
// Now that we have the commit diff in the _old_
// encoding, we'll write it back to disk using the new
// encoding which has a length prefix in front of the
// CommitSig.
commitDiff, err := legacy.DeserializeCommitDiff(
bytes.NewReader(commitDiffBytes),
)
if err != nil {
return err
}
var b bytes.Buffer
err = current.SerializeCommitDiff(&b, commitDiff)
if err != nil {
return err
}
err = chanBucket.Put(commitDiffKey, b.Bytes())
if err != nil {
return err
}
}
// With the commit diff migrated, we'll now check to see if
// there're any un-acked updates we need to migrate as well.
updateBytes := chanBucket.Get(unsignedAckedUpdatesKey)
if updateBytes != nil {
// We have un-acked updates we need to migrate so we'll
// decode then re-encode them here using the new
// format.
legacyUnackedUpdates, err := legacy.DeserializeLogUpdates(
bytes.NewReader(updateBytes),
)
if err != nil {
return err
}
var b bytes.Buffer
err = current.SerializeLogUpdates(&b, legacyUnackedUpdates)
if err != nil {
return err
}
err = chanBucket.Put(unsignedAckedUpdatesKey, b.Bytes())
if err != nil {
return err
}
}
// Remote unsiged updates as well.
updateBytes = chanBucket.Get(remoteUnsignedLocalUpdatesKey)
if updateBytes != nil {
legacyUnsignedUpdates, err := legacy.DeserializeLogUpdates(
bytes.NewReader(updateBytes),
)
if err != nil {
return err
}
var b bytes.Buffer
err = current.SerializeLogUpdates(&b, legacyUnsignedUpdates)
if err != nil {
return err
}
err = chanBucket.Put(remoteUnsignedLocalUpdatesKey, b.Bytes())
if err != nil {
return err
}
}
}
return nil
}
func migrateCloseChanSummaries(tx kvdb.RwTx) error {
closedChanBucket := tx.ReadWriteBucket(closedChannelBucket)
// Exit early if bucket is not found.
if closedChannelBucket == nil {
return nil
}
type closedChan struct {
chanKey []byte
summaryBytes []byte
}
var closedChans []closedChan
err := closedChanBucket.ForEach(func(k, v []byte) error {
closedChans = append(closedChans, closedChan{
chanKey: k,
summaryBytes: v,
})
return nil
})
if err != nil {
return err
}
for _, closedChan := range closedChans {
oldSummary, err := legacy.DeserializeCloseChannelSummary(
bytes.NewReader(closedChan.summaryBytes),
)
if err != nil {
return err
}
var newSummaryBytes bytes.Buffer
err = current.SerializeChannelCloseSummary(
&newSummaryBytes, oldSummary,
)
if err != nil {
return err
}
err = closedChanBucket.Put(
closedChan.chanKey, newSummaryBytes.Bytes(),
)
if err != nil {
return err
}
}
return nil
}
func migrateForwardingPackages(tx kvdb.RwTx) error {
fwdPkgBkt := tx.ReadWriteBucket(fwdPackagesKey)
// Exit early if bucket is not found.
if fwdPkgBkt == nil {
return nil
}
// We Go through the bucket and fetches all short channel IDs.
var sources []lnwire.ShortChannelID
err := fwdPkgBkt.ForEach(func(k, v []byte) error {
source := lnwire.NewShortChanIDFromInt(byteOrder.Uint64(k))
sources = append(sources, source)
return nil
})
if err != nil {
return err
}
// Now load all forwading packages using the legacy encoding.
var pkgsToMigrate []*common.FwdPkg
for _, source := range sources {
packager := legacy.NewChannelPackager(source)
fwdPkgs, err := packager.LoadFwdPkgs(tx)
if err != nil {
return err
}
pkgsToMigrate = append(pkgsToMigrate, fwdPkgs...)
}
// Add back the packages using the current encoding.
for _, pkg := range pkgsToMigrate {
packager := current.NewChannelPackager(pkg.Source)
err := packager.AddFwdPkg(tx, pkg)
if err != nil {
return err
}
}
return nil
}
func migrateNetworkResults(tx kvdb.RwTx) error {
networkResults := tx.ReadWriteBucket(networkResultStoreBucketKey)
// Exit early if bucket is not found.
if networkResults == nil {
return nil
}
// Similar to the prior migrations, we'll do this one in two phases:
// we'll first grab all the keys we need to migrate in one loop, then
// update them all in another loop.
var netResultsToMigrate [][2][]byte
err := networkResults.ForEach(func(k, v []byte) error {
netResultsToMigrate = append(netResultsToMigrate, [2][]byte{
k, v,
})
return nil
})
if err != nil {
return err
}
for _, netResult := range netResultsToMigrate {
resKey := netResult[0]
resBytes := netResult[1]
oldResult, err := legacy.DeserializeNetworkResult(
bytes.NewReader(resBytes),
)
if err != nil {
return err
}
var newResultBuf bytes.Buffer
err = current.SerializeNetworkResult(&newResultBuf, oldResult)
if err != nil {
return err
}
err = networkResults.Put(resKey, newResultBuf.Bytes())
if err != nil {
return err
}
}
return nil
}