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.

Author: ellemouton

graph/db/kv_store.go

@@ -4497,8 +4497,9 @@ func putChanEdgePolicy(edges kvdb.RwBucket, edge *models.ChannelEdgePolicy,
 		//
 		// TODO(halseth): get rid of these invalid policies in a
 		// migration.
-		// TODO(elle): complete the above TODO in migration from kvdb
-		// to SQL.
+		//
+		// NOTE: the above TODO was completed in the SQL migration and
+		// so such edge cases no longer need to be handled there.
 		oldEdgePolicy, err := deserializeChanEdgePolicy(
 			bytes.NewReader(edgeBytes),
 		)

graph/db/sql_migration.go

@@ -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
+}