graph/db+sqldb: make channel SQL mig retry-safe

In this commit, we make the channel part of the graph SQL migration
idempotent (retry-safe!). We do this by adding a migration-only channel
insert query that will not error out if a the query is called and a
chanenl with the given scid&version already exists. We also ensure that
errors are not thrown if existing channel features & extra types are
re-added.

Author: ellemouton

graph/db/sql_migration.go

@@ -524,7 +524,7 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
 		chunk++
 
 		// Migrate the channel info along with its policies.
-		dbChanInfo, err := insertChannel(ctx, sqlDB, channel)
+		dbChanInfo, err := insertChannelMig(ctx, sqlDB, channel)
 		if err != nil {
 			return fmt.Errorf("could not insert record for "+
 				"channel %d in SQL store: %w", scid, err)
@@ -577,8 +577,13 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
 
 		return nil
 	}, func() {
-		// No reset is needed since if a retry occurs, the entire
-		// migration will be retried from the start.
+		channelCount = 0
+		policyCount = 0
+		chunk = 0
+		skippedChanCount = 0
+		skippedPolicyCount = 0
+		t0 = time.Now()
+		batch = make(map[int64]*migChanInfo, cfg.QueryCfg.MaxBatchSize)
 	})
 	if err != nil {
 		return fmt.Errorf("could not migrate channels and policies: %w",
@@ -1452,3 +1457,106 @@ func insertNodeSQLMig(ctx context.Context, db SQLQueries,
 
 	return nodeID, nil
 }
+
+// dbChanInfo holds the DB level IDs of a channel and the nodes involved in the
+// channel.
+type dbChanInfo struct {
+	channelID int64
+	node1ID   int64
+	node2ID   int64
+}
+
+// insertChannelMig inserts a new channel record into the database during the
+// graph SQL migration.
+func insertChannelMig(ctx context.Context, db SQLQueries,
+	edge *models.ChannelEdgeInfo) (*dbChanInfo, error) {
+
+	// Make sure that at least a "shell" entry for each node is present in
+	// the nodes table.
+	//
+	// NOTE: we need this even during the SQL migration where nodes are
+	// migrated first because there are cases were some nodes may have
+	// been skipped due to invalid TLV data.
+	node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create shell node: %w", err)
+	}
+
+	node2DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey2Bytes)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create shell node: %w", err)
+	}
+
+	var capacity sql.NullInt64
+	if edge.Capacity != 0 {
+		capacity = sqldb.SQLInt64(int64(edge.Capacity))
+	}
+
+	createParams := sqlc.InsertChannelMigParams{
+		Version:     int16(ProtocolV1),
+		Scid:        channelIDToBytes(edge.ChannelID),
+		NodeID1:     node1DBID,
+		NodeID2:     node2DBID,
+		Outpoint:    edge.ChannelPoint.String(),
+		Capacity:    capacity,
+		BitcoinKey1: edge.BitcoinKey1Bytes[:],
+		BitcoinKey2: edge.BitcoinKey2Bytes[:],
+	}
+
+	if edge.AuthProof != nil {
+		proof := edge.AuthProof
+
+		createParams.Node1Signature = proof.NodeSig1Bytes
+		createParams.Node2Signature = proof.NodeSig2Bytes
+		createParams.Bitcoin1Signature = proof.BitcoinSig1Bytes
+		createParams.Bitcoin2Signature = proof.BitcoinSig2Bytes
+	}
+
+	// Insert the new channel record.
+	dbChanID, err := db.InsertChannelMig(ctx, createParams)
+	if err != nil {
+		return nil, err
+	}
+
+	// Insert any channel features.
+	for feature := range edge.Features.Features() {
+		err = db.InsertChannelFeature(
+			ctx, sqlc.InsertChannelFeatureParams{
+				ChannelID:  dbChanID,
+				FeatureBit: int32(feature),
+			},
+		)
+		if err != nil {
+			return nil, fmt.Errorf("unable to insert channel(%d) "+
+				"feature(%v): %w", dbChanID, feature, err)
+		}
+	}
+
+	// Finally, insert any extra TLV fields in the channel announcement.
+	extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
+	if err != nil {
+		return nil, fmt.Errorf("unable to marshal extra opaque "+
+			"data: %w", err)
+	}
+
+	for tlvType, value := range extra {
+		err := db.UpsertChannelExtraType(
+			ctx, sqlc.UpsertChannelExtraTypeParams{
+				ChannelID: dbChanID,
+				Type:      int64(tlvType),
+				Value:     value,
+			},
+		)
+		if err != nil {
+			return nil, fmt.Errorf("unable to upsert "+
+				"channel(%d) extra signed field(%v): %w",
+				edge.ChannelID, tlvType, err)
+		}
+	}
+
+	return &dbChanInfo{
+		channelID: dbChanID,
+		node1ID:   node1DBID,
+		node2ID:   node2DBID,
+	}, nil
+}

graph/db/sql_migration_test.go

@@ -225,7 +225,6 @@ func TestMigrateGraphToSQL(t *testing.T) {
 				numNodes:    4,
 				numChannels: 3,
 			},
-			expNotRetrySafety: true,
 		},
 		{
 			name:  "channels and policies",

graph/db/sql_store.go

@@ -112,7 +112,7 @@ type SQLQueries interface {
 	GetSCIDByOutpoint(ctx context.Context, arg sqlc.GetSCIDByOutpointParams) ([]byte, error)
 	DeleteChannels(ctx context.Context, ids []int64) error
 
-	CreateChannelExtraType(ctx context.Context, arg sqlc.CreateChannelExtraTypeParams) error
+	UpsertChannelExtraType(ctx context.Context, arg sqlc.UpsertChannelExtraTypeParams) error
 	GetChannelExtrasBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelExtraType, error)
 	InsertChannelFeature(ctx context.Context, arg sqlc.InsertChannelFeatureParams) error
 	GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]sqlc.GraphChannelFeature, error)
@@ -163,6 +163,7 @@ type SQLQueries interface {
 		structs.
 	*/
 	InsertNodeMig(ctx context.Context, arg sqlc.InsertNodeMigParams) (int64, error)
+	InsertChannelMig(ctx context.Context, arg sqlc.InsertChannelMigParams) (int64, error)
 }
 
 // BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -627,9 +628,7 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context,
 					err)
 			}
 
-			_, err = insertChannel(ctx, tx, edge)
-
-			return err
+			return insertChannel(ctx, tx, edge)
 		},
 		OnCommit: func(err error) error {
 			switch {
@@ -3799,28 +3798,20 @@ func marshalExtraOpaqueData(data []byte) (map[uint64][]byte, error) {
 	return records, nil
 }
 
-// dbChanInfo holds the DB level IDs of a channel and the nodes involved in the
-// channel.
-type dbChanInfo struct {
-	channelID int64
-	node1ID   int64
-	node2ID   int64
-}
-
 // insertChannel inserts a new channel record into the database.
 func insertChannel(ctx context.Context, db SQLQueries,
-	edge *models.ChannelEdgeInfo) (*dbChanInfo, error) {
+	edge *models.ChannelEdgeInfo) error {
 
 	// Make sure that at least a "shell" entry for each node is present in
 	// the nodes table.
 	node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes)
 	if err != nil {
-		return nil, fmt.Errorf("unable to create shell node: %w", err)
+		return fmt.Errorf("unable to create shell node: %w", err)
 	}
 
 	node2DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey2Bytes)
 	if err != nil {
-		return nil, fmt.Errorf("unable to create shell node: %w", err)
+		return fmt.Errorf("unable to create shell node: %w", err)
 	}
 
 	var capacity sql.NullInt64
@@ -3851,7 +3842,7 @@ func insertChannel(ctx context.Context, db SQLQueries,
 	// Insert the new channel record.
 	dbChanID, err := db.CreateChannel(ctx, createParams)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	// Insert any channel features.
@@ -3863,38 +3854,34 @@ func insertChannel(ctx context.Context, db SQLQueries,
 			},
 		)
 		if err != nil {
-			return nil, fmt.Errorf("unable to insert channel(%d) "+
+			return fmt.Errorf("unable to insert channel(%d) "+
 				"feature(%v): %w", dbChanID, feature, err)
 		}
 	}
 
 	// Finally, insert any extra TLV fields in the channel announcement.
 	extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
 	if err != nil {
-		return nil, fmt.Errorf("unable to marshal extra opaque "+
-			"data: %w", err)
+		return fmt.Errorf("unable to marshal extra opaque data: %w",
+			err)
 	}
 
 	for tlvType, value := range extra {
-		err := db.CreateChannelExtraType(
-			ctx, sqlc.CreateChannelExtraTypeParams{
+		err := db.UpsertChannelExtraType(
+			ctx, sqlc.UpsertChannelExtraTypeParams{
 				ChannelID: dbChanID,
 				Type:      int64(tlvType),
 				Value:     value,
 			},
 		)
 		if err != nil {
-			return nil, fmt.Errorf("unable to upsert "+
-				"channel(%d) extra signed field(%v): %w",
-				edge.ChannelID, tlvType, err)
+			return fmt.Errorf("unable to upsert channel(%d) "+
+				"extra signed field(%v): %w", edge.ChannelID,
+				tlvType, err)
 		}
 	}
 
-	return &dbChanInfo{
-		channelID: dbChanID,
-		node1ID:   node1DBID,
-		node2ID:   node2DBID,
-	}, nil
+	return nil
 }
 
 // maybeCreateShellNode checks if a shell node entry exists for the