Merge pull request #10050 from ellemouton/graphMig2-channels

[graph mig 2]: graph/db: migrate graph channels and policies from kvdb to SQL

Author: guggero

docs/release-notes/release-notes-0.20.0.md

@@ -97,6 +97,7 @@ circuit. The indices are only available for forwarding events saved after v0.20.
     * [11](https://github.com/lightningnetwork/lnd/pull/9972)
   * Add graph SQL migration logic:
     * [1](https://github.com/lightningnetwork/lnd/pull/10036)
+    * [2](https://github.com/lightningnetwork/lnd/pull/10050)
 
 ## RPC Updates
 * Previously the `RoutingPolicy` would return the inbound fee record in its

graph/db/graph_test.go

@@ -3734,12 +3734,12 @@ func TestDisabledChannelIDs(t *testing.T) {
 	}
 }
 
-// TestEdgePolicyMissingMaxHtcl tests that if we find a ChannelEdgePolicy in
+// TestEdgePolicyMissingMaxHTLC tests that if we find a ChannelEdgePolicy in
 // the DB that indicates that it should support the htlc_maximum_value_msat
 // field, but it is not part of the opaque data, then we'll handle it as it is
 // unknown. It also checks that we are correctly able to overwrite it when we
 // receive the proper update.
-func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
+func TestEdgePolicyMissingMaxHTLC(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 
@@ -3797,45 +3797,10 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
 	require.ErrorIs(t, err, ErrEdgePolicyOptionalFieldNotFound)
 
 	// Put the stripped bytes in the DB.
-	err = kvdb.Update(boltStore.db, func(tx kvdb.RwTx) error {
-		edges := tx.ReadWriteBucket(edgeBucket)
-		if edges == nil {
-			return ErrEdgeNotFound
-		}
-
-		edgeIndex := edges.NestedReadWriteBucket(edgeIndexBucket)
-		if edgeIndex == nil {
-			return ErrEdgeNotFound
-		}
-
-		var edgeKey [33 + 8]byte
-		copy(edgeKey[:], from)
-		byteOrder.PutUint64(edgeKey[33:], edge1.ChannelID)
-
-		var scratch [8]byte
-		var indexKey [8 + 8]byte
-		copy(indexKey[:], scratch[:])
-		byteOrder.PutUint64(indexKey[8:], edge1.ChannelID)
-
-		updateIndex, err := edges.CreateBucketIfNotExists(
-			edgeUpdateIndexBucket,
-		)
-		if err != nil {
-			return err
-		}
-
-		if err := updateIndex.Put(indexKey[:], nil); err != nil {
-			return err
-		}
-
-		return edges.Put(edgeKey[:], stripped)
-	}, func() {})
-	require.NoError(t, err, "error writing db")
+	putSerializedPolicy(t, boltStore.db, from, chanID, stripped)
 
 	// And add the second, unmodified edge.
-	if err := graph.UpdateEdgePolicy(ctx, edge2); err != nil {
-		t.Fatalf("unable to update edge: %v", err)
-	}
+	require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
 
 	// Attempt to fetch the edge and policies from the DB. Since the policy
 	// we added is invalid according to the new format, it should be as we
@@ -3870,6 +3835,38 @@ func TestEdgePolicyMissingMaxHtcl(t *testing.T) {
 	assertEdgeInfoEqual(t, dbEdgeInfo, edgeInfo)
 }
 
+// putSerializedPolicy is a helper function that writes a serialized
+// ChannelEdgePolicy to the edge bucket in the database.
+func putSerializedPolicy(t *testing.T, db kvdb.Backend, from []byte,
+	chanID uint64, b []byte) {
+
+	err := kvdb.Update(db, func(tx kvdb.RwTx) error {
+		edges := tx.ReadWriteBucket(edgeBucket)
+		require.NotNil(t, edges)
+
+		edgeIndex := edges.NestedReadWriteBucket(edgeIndexBucket)
+		require.NotNil(t, edgeIndex)
+
+		var edgeKey [33 + 8]byte
+		copy(edgeKey[:], from)
+		byteOrder.PutUint64(edgeKey[33:], chanID)
+
+		var scratch [8]byte
+		var indexKey [8 + 8]byte
+		copy(indexKey[:], scratch[:])
+		byteOrder.PutUint64(indexKey[8:], chanID)
+
+		updateIndex, err := edges.CreateBucketIfNotExists(
+			edgeUpdateIndexBucket,
+		)
+		require.NoError(t, err)
+		require.NoError(t, updateIndex.Put(indexKey[:], nil))
+
+		return edges.Put(edgeKey[:], b)
+	}, func() {})
+	require.NoError(t, err, "error writing db")
+}
+
 // assertNumZombies queries the provided ChannelGraph for NumZombies, and
 // asserts that the returned number is equal to expZombies.
 func assertNumZombies(t *testing.T, graph *ChannelGraph, expZombies uint64) {

graph/db/kv_store.go

@@ -2844,7 +2844,17 @@ func (c *KVStore) UpdateEdgePolicy(ctx context.Context,
 			edgeNotFound = false
 		},
 		Do: func(tx kvdb.RwTx) error {
-			var err error
+			// Validate that the ExtraOpaqueData is in fact a valid
+			// TLV stream. This is done here instead of within
+			// updateEdgePolicy so that updateEdgePolicy can be used
+			// by unit tests to recreate the case where we already
+			// have nodes persisted with invalid TLV data.
+			err := edge.ExtraOpaqueData.ValidateTLV()
+			if err != nil {
+				return fmt.Errorf("%w: %w",
+					ErrParsingExtraTLVBytes, err)
+			}
+
 			from, to, isUpdate1, err = updateEdgePolicy(tx, edge)
 			if err != nil {
 				log.Errorf("UpdateEdgePolicy faild: %v", err)
@@ -4487,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),
 		)
@@ -4703,12 +4714,6 @@ func serializeChanEdgePolicy(w io.Writer, edge *models.ChannelEdgePolicy,
 		}
 	}
 
-	// Validate that the ExtraOpaqueData is in fact a valid TLV stream.
-	err = edge.ExtraOpaqueData.ValidateTLV()
-	if err != nil {
-		return fmt.Errorf("%w: %w", ErrParsingExtraTLVBytes, err)
-	}
-
 	if len(edge.ExtraOpaqueData) > MaxAllowedExtraOpaqueBytes {
 		return ErrTooManyExtraOpaqueBytes(len(edge.ExtraOpaqueData))
 	}