graph/db: migrate channels and polices to SQL
In this commit, the `MigrateGraphToSQL` function is expanded to migrate the channel and channe policy data. Both of these have the special case where the kvdb store records may contain invalid TLV. If we encounter a channel with invalid TLV, we skip it and its policies. If we encounter a policy with invalid TLV, we skip it. The `TestMigrateGraphToSQL` and `TestMigrationWithChannelDB` tests are updated accordingly.
This commit is contained in:
3 files changed
+714
-5
No files matched your search
+282
-1
@@ -23,7 +23,7 @@ import (
|
||||
// only for now and will be called from the main lnd binary once the
|
||||
// migration is fully implemented and tested.
|
||||
func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
|
||||
sqlDB SQLQueries, _ chainhash.Hash) error {
|
||||
sqlDB SQLQueries, chain chainhash.Hash) error {
|
||||
|
||||
log.Infof("Starting migration of the graph store from KV to SQL")
|
||||
t0 := time.Now()
|
||||
@@ -48,6 +48,13 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
|
||||
return fmt.Errorf("could not migrate source node: %w", err)
|
||||
}
|
||||
|
||||
// 3) Migrate all the channels and channel policies.
|
||||
err = migrateChannelsAndPolicies(ctx, kvBackend, sqlDB, chain)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not migrate channels and policies: %w",
|
||||
err)
|
||||
}
|
||||
|
||||
log.Infof("Finished migration of the graph store from KV to SQL in %v",
|
||||
time.Since(t0))
|
||||
|
||||
@@ -261,3 +268,277 @@ func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend,
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// migrateChannelsAndPolicies migrates all channels and their policies
|
||||
// from the KV backend to the SQL database.
|
||||
func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
|
||||
sqlDB SQLQueries, chain chainhash.Hash) error {
|
||||
|
||||
var (
|
||||
channelCount uint64
|
||||
skippedChanCount uint64
|
||||
policyCount uint64
|
||||
skippedPolicyCount uint64
|
||||
)
|
||||
migChanPolicy := func(policy *models.ChannelEdgePolicy) error {
|
||||
// If the policy is nil, we can skip it.
|
||||
if policy == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Unlike the special case of invalid TLV bytes for node and
|
||||
// channel announcements, we don't need to handle the case for
|
||||
// channel policies here because it is already handled in the
|
||||
// `forEachChannel` function. If the policy has invalid TLV
|
||||
// bytes, then `nil` will be passed to this function.
|
||||
|
||||
policyCount++
|
||||
|
||||
_, _, _, err := updateChanEdgePolicy(ctx, sqlDB, policy)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not migrate channel "+
|
||||
"policy %d: %w", policy.ChannelID, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Iterate over each channel in the KV store and migrate it and its
|
||||
// policies to the SQL database.
|
||||
err := forEachChannel(kvBackend, func(channel *models.ChannelEdgeInfo,
|
||||
policy1 *models.ChannelEdgePolicy,
|
||||
policy2 *models.ChannelEdgePolicy) error {
|
||||
|
||||
scid := channel.ChannelID
|
||||
|
||||
// Here, we do a sanity check to ensure that the chain hash of
|
||||
// the channel returned by the KV store matches the expected
|
||||
// chain hash. This is important since in the SQL store, we will
|
||||
// no longer explicitly store the chain hash in the channel
|
||||
// info, but rather rely on the chain hash LND is running with.
|
||||
// So this is our way of ensuring that LND is running on the
|
||||
// correct network at migration time.
|
||||
if channel.ChainHash != chain {
|
||||
return fmt.Errorf("channel %d has chain hash %s, "+
|
||||
"expected %s", scid, channel.ChainHash, chain)
|
||||
}
|
||||
|
||||
// Sanity check to ensure that the channel has valid extra
|
||||
// opaque data. If it does not, we'll skip it. We need to do
|
||||
// this because previously we would just persist any TLV bytes
|
||||
// that we received without validating them. Now, however, we
|
||||
// normalise the storage of extra opaque data, so we need to
|
||||
// ensure that the data is valid. We don't want to abort the
|
||||
// migration if we encounter a channel with invalid extra opaque
|
||||
// data, so we'll just skip it and log a warning.
|
||||
_, err := marshalExtraOpaqueData(channel.ExtraOpaqueData)
|
||||
if errors.Is(err, ErrParsingExtraTLVBytes) {
|
||||
log.Warnf("Skipping channel %d with invalid "+
|
||||
"extra opaque data: %v", scid,
|
||||
channel.ExtraOpaqueData)
|
||||
|
||||
skippedChanCount++
|
||||
|
||||
// If we skip a channel, we also skip its policies.
|
||||
if policy1 != nil {
|
||||
skippedPolicyCount++
|
||||
}
|
||||
if policy2 != nil {
|
||||
skippedPolicyCount++
|
||||
}
|
||||
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return fmt.Errorf("unable to marshal extra opaque "+
|
||||
"data for channel %d (%v): %w", scid,
|
||||
channel.ExtraOpaqueData, err)
|
||||
}
|
||||
|
||||
channelCount++
|
||||
err = migrateSingleChannel(
|
||||
ctx, sqlDB, channel, policy1, policy2, migChanPolicy,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not migrate channel %d: %w",
|
||||
scid, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not migrate channels and policies: %w",
|
||||
err)
|
||||
}
|
||||
|
||||
log.Infof("Migrated %d channels and %d policies from KV to SQL "+
|
||||
"(skipped %d channels and %d policies due to invalid TLV "+
|
||||
"streams)", channelCount, policyCount, skippedChanCount,
|
||||
skippedPolicyCount)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries,
|
||||
channel *models.ChannelEdgeInfo,
|
||||
policy1, policy2 *models.ChannelEdgePolicy,
|
||||
migChanPolicy func(*models.ChannelEdgePolicy) error) error {
|
||||
|
||||
scid := channel.ChannelID
|
||||
|
||||
// First, migrate the channel info along with its policies.
|
||||
dbChanInfo, err := insertChannel(ctx, sqlDB, channel)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not insert record for channel %d "+
|
||||
"in SQL store: %w", scid, err)
|
||||
}
|
||||
|
||||
// Now, migrate the two channel policies.
|
||||
err = migChanPolicy(policy1)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not migrate policy1(%d): %w", scid,
|
||||
err)
|
||||
}
|
||||
err = migChanPolicy(policy2)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not migrate policy2(%d): %w", scid,
|
||||
err)
|
||||
}
|
||||
|
||||
// Now, fetch the channel and its policies from the SQL DB.
|
||||
row, err := sqlDB.GetChannelBySCIDWithPolicies(
|
||||
ctx, sqlc.GetChannelBySCIDWithPoliciesParams{
|
||||
Scid: channelIDToBytes(scid),
|
||||
Version: int16(ProtocolV1),
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not get channel by SCID(%d): %w", scid,
|
||||
err)
|
||||
}
|
||||
|
||||
// Assert that the DB IDs for the channel and nodes are as expected
|
||||
// given the inserted channel info.
|
||||
err = sqldb.CompareRecords(
|
||||
dbChanInfo.channelID, row.Channel.ID, "channel DB ID",
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = sqldb.CompareRecords(
|
||||
dbChanInfo.node1ID, row.Node.ID, "node1 DB ID",
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = sqldb.CompareRecords(
|
||||
dbChanInfo.node2ID, row.Node_2.ID, "node2 DB ID",
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
migChan, migPol1, migPol2, err := getAndBuildChanAndPolicies(
|
||||
ctx, sqlDB, row, channel.ChainHash,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not build migrated channel and "+
|
||||
"policies: %w", err)
|
||||
}
|
||||
|
||||
// Finally, compare the original channel info and
|
||||
// policies with the migrated ones to ensure they match.
|
||||
if len(channel.ExtraOpaqueData) == 0 {
|
||||
channel.ExtraOpaqueData = nil
|
||||
}
|
||||
if len(migChan.ExtraOpaqueData) == 0 {
|
||||
migChan.ExtraOpaqueData = nil
|
||||
}
|
||||
|
||||
err = sqldb.CompareRecords(
|
||||
channel, migChan, fmt.Sprintf("channel %d", scid),
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
checkPolicy := func(expPolicy,
|
||||
migPolicy *models.ChannelEdgePolicy) error {
|
||||
|
||||
switch {
|
||||
// Both policies are nil, nothing to compare.
|
||||
case expPolicy == nil && migPolicy == nil:
|
||||
return nil
|
||||
|
||||
// One of the policies is nil, but the other is not.
|
||||
case expPolicy == nil || migPolicy == nil:
|
||||
return fmt.Errorf("expected both policies to be "+
|
||||
"non-nil. Got expPolicy: %v, "+
|
||||
"migPolicy: %v", expPolicy, migPolicy)
|
||||
|
||||
// Both policies are non-nil, we can compare them.
|
||||
default:
|
||||
}
|
||||
|
||||
if len(expPolicy.ExtraOpaqueData) == 0 {
|
||||
expPolicy.ExtraOpaqueData = nil
|
||||
}
|
||||
if len(migPolicy.ExtraOpaqueData) == 0 {
|
||||
migPolicy.ExtraOpaqueData = nil
|
||||
}
|
||||
|
||||
return sqldb.CompareRecords(
|
||||
*expPolicy, *migPolicy, "channel policy",
|
||||
)
|
||||
}
|
||||
|
||||
err = checkPolicy(policy1, migPol1)
|
||||
if err != nil {
|
||||
return fmt.Errorf("policy1 mismatch for channel %d: %w", scid,
|
||||
err)
|
||||
}
|
||||
|
||||
err = checkPolicy(policy2, migPol2)
|
||||
if err != nil {
|
||||
return fmt.Errorf("policy2 mismatch for channel %d: %w", scid,
|
||||
err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries,
|
||||
row sqlc.GetChannelBySCIDWithPoliciesRow,
|
||||
chain chainhash.Hash) (*models.ChannelEdgeInfo,
|
||||
*models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) {
|
||||
|
||||
node1, node2, err := buildNodeVertices(
|
||||
row.Node.PubKey, row.Node_2.PubKey,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
|
||||
edge, err := getAndBuildEdgeInfo(
|
||||
ctx, db, chain, row.Channel.ID, row.Channel, node1, node2,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("unable to build channel "+
|
||||
"info: %w", err)
|
||||
}
|
||||
|
||||
dbPol1, dbPol2, err := extractChannelPolicies(row)
|
||||
if err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("unable to extract channel "+
|
||||
"policies: %w", err)
|
||||
}
|
||||
|
||||
policy1, policy2, err := getAndBuildChanPolicies(
|
||||
ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, nil, nil, fmt.Errorf("unable to build channel "+
|
||||
"policies: %w", err)
|
||||
}
|
||||
|
||||
return edge, policy1, policy2, nil
|
||||
}
|
||||
Reference in new issue
Block a user