graph/db: migrate source node to SQL store

Author: ellemouton

graph/db/sql_migration.go

@@ -43,6 +43,11 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend,
 		return fmt.Errorf("could not migrate nodes: %w", err)
 	}
 
+	// 2) Migrate the source node.
+	if err := migrateSourceNode(ctx, kvBackend, sqlDB); err != nil {
+		return fmt.Errorf("could not migrate source node: %w", err)
+	}
+
 	log.Infof("Finished migration of the graph store from KV to SQL in %v",
 		time.Since(t0))
 
@@ -184,3 +189,75 @@ func migrateNodes(ctx context.Context, kvBackend kvdb.Backend,
 
 	return nil
 }
+
+// migrateSourceNode migrates the source node from the KV backend to the
+// SQL database.
+func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend,
+	sqlDB SQLQueries) error {
+
+	sourceNode, err := sourceNode(kvdb)
+	if errors.Is(err, ErrSourceNodeNotSet) {
+		// If the source node has not been set yet, we can skip this
+		// migration step.
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("could not get source node from kv "+
+			"store: %w", err)
+	}
+
+	pub := sourceNode.PubKeyBytes
+
+	// Get the DB ID of the source node by its public key. This node must
+	// already exist in the SQL database, as it should have been migrated
+	// in the previous node-migration step.
+	id, err := sqlDB.GetNodeIDByPubKey(
+		ctx, sqlc.GetNodeIDByPubKeyParams{
+			PubKey:  pub[:],
+			Version: int16(ProtocolV1),
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("could not get source node ID: %w", err)
+	}
+
+	// Now we can add the source node to the SQL database.
+	err = sqlDB.AddSourceNode(ctx, id)
+	if err != nil {
+		return fmt.Errorf("could not add source node to SQL store: %w",
+			err)
+	}
+
+	// Verify that the source node was added correctly by fetching it back
+	// from the SQL database and checking that the expected DB ID and
+	// pub key are returned. We don't need to do a whole node comparison
+	// here, as this was already done in the previous migration step.
+	srcNodes, err := sqlDB.GetSourceNodesByVersion(ctx, int16(ProtocolV1))
+	if err != nil {
+		return fmt.Errorf("could not get source nodes from SQL "+
+			"store: %w", err)
+	}
+
+	// The SQL store has support for multiple source nodes (for future
+	// protocol versions) but this migration is purely aimed at the V1
+	// store, and so we expect exactly one source node to be present.
+	if len(srcNodes) != 1 {
+		return fmt.Errorf("expected exactly one source node, "+
+			"got %d", len(srcNodes))
+	}
+
+	// Check that the source node ID and pub key match the original
+	// source node.
+	if srcNodes[0].NodeID != id {
+		return fmt.Errorf("source node ID mismatch after migration: "+
+			"expected %d, got %d", id, srcNodes[0].NodeID)
+	}
+	err = sqldb.CompareRecords(pub[:], srcNodes[0].PubKey, "source node")
+	if err != nil {
+		return fmt.Errorf("source node pubkey mismatch after "+
+			"migration: %w", err)
+	}
+
+	log.Infof("Migrated source node with pubkey %x to SQL", pub[:])
+
+	return nil
+}

graph/db/sql_migration_test.go

@@ -5,6 +5,7 @@ package graphdb
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"image/color"
 	"net"
@@ -106,6 +107,23 @@ func TestMigrateGraphToSQL(t *testing.T) {
 				numNodes: 6,
 			},
 		},
+		{
+			name: "source node",
+			write: func(t *testing.T, db *KVStore, object any) {
+				node, ok := object.(*models.LightningNode)
+				require.True(t, ok)
+
+				err := db.SetSourceNode(ctx, node)
+				require.NoError(t, err)
+			},
+			objects: []any{
+				makeTestNode(t),
+			},
+			expGraphStats: graphStats{
+				numNodes:   1,
+				srcNodeSet: true,
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -138,7 +156,8 @@ func TestMigrateGraphToSQL(t *testing.T) {
 
 // graphStats holds expected statistics about the graph after migration.
 type graphStats struct {
-	numNodes int
+	numNodes   int
+	srcNodeSet bool
 }
 
 // assertInSync checks that the KVStore and SQLStore both contain the same
@@ -150,6 +169,12 @@ func assertInSync(t *testing.T, kvDB *KVStore, sqlDB *SQLStore,
 	sqlNodes := fetchAllNodes(t, sqlDB)
 	require.Len(t, sqlNodes, stats.numNodes)
 	require.Equal(t, fetchAllNodes(t, kvDB), sqlNodes)
+
+	// 2) Check that the source nodes match (if indeed source nodes have
+	// been set).
+	sqlSourceNode := fetchSourceNode(t, sqlDB)
+	require.Equal(t, stats.srcNodeSet, sqlSourceNode != nil)
+	require.Equal(t, fetchSourceNode(t, kvDB), sqlSourceNode)
 }
 
 // fetchAllNodes retrieves all nodes from the given store and returns them
@@ -179,6 +204,18 @@ func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode {
 	return nodes
 }
 
+// fetchSourceNode retrieves the source node from the given store.
+func fetchSourceNode(t *testing.T, store V1Store) *models.LightningNode {
+	node, err := store.SourceNode(context.Background())
+	if errors.Is(err, ErrSourceNodeNotSet) {
+		return nil
+	} else {
+		require.NoError(t, err)
+	}
+
+	return node
+}
+
 // setUpKVStore initializes a new KVStore for testing.
 func setUpKVStore(t *testing.T) *KVStore {
 	kvDB, cleanup, err := kvdb.GetTestBackend(t.TempDir(), "graph")