Merge pull request #10161 from ellemouton/graphRetry

graph/db+sqldb: Make the SQL migration retry-safe/idempotent

graph/db/sql_migration.go

@@ -273,7 +273,7 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig,
 		// production.
 
 		// Write the node to the SQL database.
-		id, err := upsertNode(ctx, sqlDB, node)
+		id, err := insertNodeSQLMig(ctx, sqlDB, node)
 		if err != nil {
 			return fmt.Errorf("could not persist node(%x): %w", pub,
 				err)
@@ -303,8 +303,11 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig,
 
 		return nil
 	}, func() {
-		// No reset is needed since if a retry occurs, the entire
+		count = 0
-		// migration will be retried from the start.
+		chunk = 0
+		skipped = 0
+		t0 = time.Now()
+		batch = make(map[int64]*models.LightningNode, cfg.MaxBatchSize)
 	})
 	if err != nil {
 		return fmt.Errorf("could not migrate nodes: %w", err)
@@ -437,7 +440,9 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
 			Interval: 10 * time.Second,
 		}
 	)
-	migChanPolicy := func(policy *models.ChannelEdgePolicy) error {
+	migChanPolicy := func(dbChanInfo *dbChanInfo,
+		policy *models.ChannelEdgePolicy) error {
+
 		// If the policy is nil, we can skip it.
 		if policy == nil {
 			return nil
@@ -451,7 +456,7 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
 
 		policyCount++
 
-		_, _, _, err := updateChanEdgePolicy(ctx, sqlDB, policy)
+		err := insertChanEdgePolicyMig(ctx, sqlDB, dbChanInfo, policy)
 		if err != nil {
 			return fmt.Errorf("could not migrate channel "+
 				"policy %d: %w", policy.ChannelID, err)
@@ -521,19 +526,19 @@ 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)
 		}
 
 		// Now, migrate the two channel policies for the channel.
-		err = migChanPolicy(policy1)
+		err = migChanPolicy(dbChanInfo, policy1)
 		if err != nil {
 			return fmt.Errorf("could not migrate policy1(%d): %w",
 				scid, err)
 		}
-		err = migChanPolicy(policy2)
+		err = migChanPolicy(dbChanInfo, policy2)
 		if err != nil {
 			return fmt.Errorf("could not migrate policy2(%d): %w",
 				scid, err)
@@ -574,8 +579,13 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig,
 
 		return nil
 	}, func() {
-		// No reset is needed since if a retry occurs, the entire
+		channelCount = 0
-		// migration will be retried from the start.
+		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",
@@ -923,6 +933,14 @@ func migratePruneLog(ctx context.Context, cfg *sqldb.QueryConfig,
 
 			return nil
 		},
+		func() {
+			count = 0
+			chunk = 0
+			t0 = time.Now()
+			batch = make(
+				map[uint32]chainhash.Hash, cfg.MaxBatchSize,
+			)
+		},
 	)
 	if err != nil {
 		return fmt.Errorf("could not migrate prune log: %w", err)
@@ -975,7 +993,7 @@ func migratePruneLog(ctx context.Context, cfg *sqldb.QueryConfig,
 // forEachPruneLogEntry iterates over each prune log entry in the KV
 // backend and calls the provided callback function for each entry.
 func forEachPruneLogEntry(db kvdb.Backend, cb func(height uint32,
-	hash *chainhash.Hash) error) error {
+	hash *chainhash.Hash) error, reset func()) error {
 
 	return kvdb.View(db, func(tx kvdb.RTx) error {
 		metaBucket := tx.ReadBucket(graphMetaBucket)
@@ -997,7 +1015,7 @@ func forEachPruneLogEntry(db kvdb.Backend, cb func(height uint32,
 
 			return cb(blockHeight, &blockHash)
 		})
-	}, func() {})
+	}, reset)
 }
 
 // migrateClosedSCIDIndex migrates the closed SCID index from the KV backend to
@@ -1091,7 +1109,14 @@ func migrateClosedSCIDIndex(ctx context.Context, cfg *sqldb.QueryConfig,
 		return nil
 	}
 
-	err := forEachClosedSCID(kvBackend, migrateSingleClosedSCID)
+	err := forEachClosedSCID(
+		kvBackend, migrateSingleClosedSCID, func() {
+			count = 0
+			chunk = 0
+			t0 = time.Now()
+			batch = make([][]byte, 0, cfg.MaxBatchSize)
+		},
+	)
 	if err != nil {
 		return fmt.Errorf("could not migrate closed SCID index: %w",
 			err)
@@ -1270,6 +1295,11 @@ func migrateZombieIndex(ctx context.Context, cfg *sqldb.QueryConfig,
 		})
 
 		return nil
+	}, func() {
+		count = 0
+		chunk = 0
+		t0 = time.Now()
+		batch = make(map[uint64]*zombieEntry, cfg.MaxBatchSize)
 	})
 	if err != nil {
 		return fmt.Errorf("could not migrate zombie index: %w", err)
@@ -1293,7 +1323,7 @@ func migrateZombieIndex(ctx context.Context, cfg *sqldb.QueryConfig,
 // forEachZombieEntry iterates over each zombie channel entry in the
 // KV backend and calls the provided callback function for each entry.
 func forEachZombieEntry(db kvdb.Backend, cb func(chanID uint64, pubKey1,
-	pubKey2 [33]byte) error) error {
+	pubKey2 [33]byte) error, reset func()) error {
 
 	return kvdb.View(db, func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
@@ -1312,13 +1342,13 @@ func forEachZombieEntry(db kvdb.Backend, cb func(chanID uint64, pubKey1,
 
 			return cb(byteOrder.Uint64(k), pubKey1, pubKey2)
 		})
-	}, func() {})
+	}, reset)
 }
 
 // forEachClosedSCID iterates over each closed SCID in the KV backend and calls
 // the provided callback function for each SCID.
 func forEachClosedSCID(db kvdb.Backend,
-	cb func(lnwire.ShortChannelID) error) error {
+	cb func(lnwire.ShortChannelID) error, reset func()) error {
 
 	return kvdb.View(db, func(tx kvdb.RTx) error {
 		closedScids := tx.ReadBucket(closedScidBucket)
@@ -1331,5 +1361,280 @@ func forEachClosedSCID(db kvdb.Backend,
 				byteOrder.Uint64(k),
 			))
 		})
-	}, func() {})
+	}, reset)
+}
+
+// insertNodeSQLMig inserts the node record into the database during the graph
+// SQL migration. No error is expected if the node already exists. Unlike the
+// main upsertNode function, this function does not require that a new node
+// update have a newer timestamp than the existing one. This is because we want
+// the migration to be idempotent and dont want to error out if we re-insert the
+// exact same node.
+func insertNodeSQLMig(ctx context.Context, db SQLQueries,
+	node *models.LightningNode) (int64, error) {
+
+	params := sqlc.InsertNodeMigParams{
+		Version: int16(ProtocolV1),
+		PubKey:  node.PubKeyBytes[:],
+	}
+
+	if node.HaveNodeAnnouncement {
+		params.LastUpdate = sqldb.SQLInt64(node.LastUpdate.Unix())
+		params.Color = sqldb.SQLStr(EncodeHexColor(node.Color))
+		params.Alias = sqldb.SQLStr(node.Alias)
+		params.Signature = node.AuthSigBytes
+	}
+
+	nodeID, err := db.InsertNodeMig(ctx, params)
+	if err != nil {
+		return 0, fmt.Errorf("upserting node(%x): %w", node.PubKeyBytes,
+			err)
+	}
+
+	// We can exit here if we don't have the announcement yet.
+	if !node.HaveNodeAnnouncement {
+		return nodeID, nil
+	}
+
+	// Insert the node's features.
+	for feature := range node.Features.Features() {
+		err = db.InsertNodeFeature(ctx, sqlc.InsertNodeFeatureParams{
+			NodeID:     nodeID,
+			FeatureBit: int32(feature),
+		})
+		if err != nil {
+			return 0, fmt.Errorf("unable to insert node(%d) "+
+				"feature(%v): %w", nodeID, feature, err)
+		}
+	}
+
+	// Update the node's addresses.
+	newAddresses, err := collectAddressRecords(node.Addresses)
+	if err != nil {
+		return 0, err
+	}
+
+	// Any remaining entries in newAddresses are new addresses that need to
+	// be added to the database for the first time.
+	for addrType, addrList := range newAddresses {
+		for position, addr := range addrList {
+			err := db.UpsertNodeAddress(
+				ctx, sqlc.UpsertNodeAddressParams{
+					NodeID:   nodeID,
+					Type:     int16(addrType),
+					Address:  addr,
+					Position: int32(position),
+				},
+			)
+			if err != nil {
+				return 0, fmt.Errorf("unable to insert "+
+					"node(%d) address(%v): %w", nodeID,
+					addr, err)
+			}
+		}
+	}
+
+	// Convert the flat extra opaque data into a map of TLV types to
+	// values.
+	extra, err := marshalExtraOpaqueData(node.ExtraOpaqueData)
+	if err != nil {
+		return 0, fmt.Errorf("unable to marshal extra opaque data: %w",
+			err)
+	}
+
+	// Insert the node's extra signed fields.
+	for tlvType, value := range extra {
+		err = db.UpsertNodeExtraType(
+			ctx, sqlc.UpsertNodeExtraTypeParams{
+				NodeID: nodeID,
+				Type:   int64(tlvType),
+				Value:  value,
+			},
+		)
+		if err != nil {
+			return 0, fmt.Errorf("unable to upsert node(%d) extra "+
+				"signed field(%v): %w", nodeID, tlvType, err)
+		}
+	}
+
+	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
+}
+
+// insertChanEdgePolicyMig inserts the channel policy info we have stored for
+// a channel we already know of. This is used during the SQL migration
+// process to insert channel policies.
+func insertChanEdgePolicyMig(ctx context.Context, tx SQLQueries,
+	dbChan *dbChanInfo, edge *models.ChannelEdgePolicy) error {
+
+	// Figure out which node this edge is from.
+	isNode1 := edge.ChannelFlags&lnwire.ChanUpdateDirection == 0
+	nodeID := dbChan.node1ID
+	if !isNode1 {
+		nodeID = dbChan.node2ID
+	}
+
+	var (
+		inboundBase sql.NullInt64
+		inboundRate sql.NullInt64
+	)
+	edge.InboundFee.WhenSome(func(fee lnwire.Fee) {
+		inboundRate = sqldb.SQLInt64(fee.FeeRate)
+		inboundBase = sqldb.SQLInt64(fee.BaseFee)
+	})
+
+	id, err := tx.InsertEdgePolicyMig(ctx, sqlc.InsertEdgePolicyMigParams{
+		Version:     int16(ProtocolV1),
+		ChannelID:   dbChan.channelID,
+		NodeID:      nodeID,
+		Timelock:    int32(edge.TimeLockDelta),
+		FeePpm:      int64(edge.FeeProportionalMillionths),
+		BaseFeeMsat: int64(edge.FeeBaseMSat),
+		MinHtlcMsat: int64(edge.MinHTLC),
+		LastUpdate:  sqldb.SQLInt64(edge.LastUpdate.Unix()),
+		Disabled: sql.NullBool{
+			Valid: true,
+			Bool:  edge.IsDisabled(),
+		},
+		MaxHtlcMsat: sql.NullInt64{
+			Valid: edge.MessageFlags.HasMaxHtlc(),
+			Int64: int64(edge.MaxHTLC),
+		},
+		MessageFlags:            sqldb.SQLInt16(edge.MessageFlags),
+		ChannelFlags:            sqldb.SQLInt16(edge.ChannelFlags),
+		InboundBaseFeeMsat:      inboundBase,
+		InboundFeeRateMilliMsat: inboundRate,
+		Signature:               edge.SigBytes,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to upsert edge policy: %w", err)
+	}
+
+	// Convert the flat extra opaque data into a map of TLV types to
+	// values.
+	extra, err := marshalExtraOpaqueData(edge.ExtraOpaqueData)
+	if err != nil {
+		return fmt.Errorf("unable to marshal extra opaque data: %w",
+			err)
+	}
+
+	// Insert all new extra signed fields for the channel policy.
+	for tlvType, value := range extra {
+		err = tx.UpsertChanPolicyExtraType(
+			ctx, sqlc.UpsertChanPolicyExtraTypeParams{
+				ChannelPolicyID: id,
+				Type:            int64(tlvType),
+				Value:           value,
+			},
+		)
+		if err != nil {
+			return fmt.Errorf("unable to insert "+
+				"channel_policy(%d) extra signed field(%v): %w",
+				id, tlvType, err)
+		}
+	}
+
+	return nil
 }

graph/db/sql_migration_test.go

@@ -56,6 +56,13 @@ var (
 // for to ensure that our migration from a graph store backed by a KV DB to a
 // SQL database works as expected. At the end of each test, the DBs are compared
 // and expected to have the exact same data in them.
+// This test also ensures that the migration is "retry-safe". This is needed
+// because the migration is hooked up to 2 dbs: the source DB and the
+// destination. The source DB is a db behind the kvdb.Backend interface and
+// the migration makes use of methods on this interface that may be retried
+// under the hood. The migration often does logic inside call-back functions
+// passed to the source DB methods which may be retried, and so we need to
+// ensure that the migration can handle this.
 func TestMigrateGraphToSQL(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
@@ -157,6 +164,59 @@ func TestMigrateGraphToSQL(t *testing.T) {
 				srcNodeSet: true,
 			},
 		},
+		{
+			name:  "channel with no policies",
+			write: writeUpdate,
+			objects: []any{
+				// A channel with unknown nodes. This will
+				// result in two shell nodes being created.
+				// 	- channel count += 1
+				// 	- node count += 2
+				makeTestChannel(t),
+
+				// Insert some nodes.
+				// 	- node count += 1
+				makeTestNode(t, func(n *models.LightningNode) {
+					n.PubKeyBytes = node1
+				}),
+				// 	- node count += 1
+				makeTestNode(t, func(n *models.LightningNode) {
+					n.PubKeyBytes = node2
+				}),
+
+				// A channel with known nodes.
+				// - channel count += 1
+				makeTestChannel(
+					t, func(c *models.ChannelEdgeInfo) {
+						c.ChannelID = chanID1
+
+						c.NodeKey1Bytes = node1
+						c.NodeKey2Bytes = node2
+					},
+				),
+
+				// Insert a channel with no auth proof, no
+				// extra opaque data, and empty features.
+				// Use known nodes.
+				// 	- channel count += 1
+				makeTestChannel(
+					t, func(c *models.ChannelEdgeInfo) {
+						c.ChannelID = chanID2
+
+						c.NodeKey1Bytes = node1
+						c.NodeKey2Bytes = node2
+
+						c.AuthProof = nil
+						c.ExtraOpaqueData = nil
+						c.Features = testEmptyFeatures
+					},
+				),
+			},
+			expGraphStats: graphStats{
+				numNodes:    4,
+				numChannels: 3,
+			},
+		},
 		{
 			name:  "channels and policies",
 			write: writeUpdate,
@@ -338,6 +398,12 @@ func TestMigrateGraphToSQL(t *testing.T) {
 
 			// Validate that the two databases are now in sync.
 			assertInSync(t, kvDB, sql, test.expGraphStats)
+
+			// The migration should be retry-safe, so running it
+			// again should not change the state of the databases.
+			err = MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db)
+			require.NoError(t, err)
+			assertInSync(t, kvDB, sql, test.expGraphStats)
 		})
 	}
 }
@@ -513,6 +579,7 @@ func checkKVPruneLogEntries(t *testing.T, kv *KVStore, sql *SQLStore,
 
 			return nil
 		},
+		func() {},
 	)
 	require.NoError(t, err)
 
@@ -544,7 +611,7 @@ func checkClosedSCIDIndex(t *testing.T, kv kvdb.Backend, sql *SQLStore) {
 		require.True(t, closed)
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 }
 
@@ -576,7 +643,7 @@ func checkZombieIndex(t *testing.T, kv kvdb.Backend, sql *SQLStore) {
 		}
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 }
 
@@ -716,6 +783,7 @@ func makeTestPolicy(chanID uint64, toNode route.Vertex, isNode1 bool,
 		FeeBaseMSat:               math.MaxUint64,
 		FeeProportionalMillionths: math.MaxUint64,
 		ToNode:                    toNode,
+		ExtraOpaqueData:           testExtraData,
 	}
 
 	for _, opt := range opts {
@@ -1205,6 +1273,8 @@ func TestMigrateGraphToSQLRapid(t *testing.T) {
 // SQL store, generates random nodes and channels, populates the KV store,
 // runs the migration, and asserts that the SQL store contains the expected
 // state.
+//
+// The migration is run twice in order to test idempotency and retry-safety.
 func testMigrateGraphToSQLRapidOnce(t *testing.T, rt *rapid.T,
 	dbFixture *sqldb.TestPgFixture, maxNumNodes, maxNumChannels int) {
 
@@ -1355,6 +1425,15 @@ func testMigrateGraphToSQLRapidOnce(t *testing.T, rt *rapid.T,
 		nodes: nodesSlice,
 		chans: chanSetForState,
 	})
+
+	// The migration is expected to be idempotent and retry-safe. So running
+	// it again should yield the same result.
+	err = MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db)
+	require.NoError(t, err)
+	assertResultState(t, sql, dbState{
+		nodes: nodesSlice,
+		chans: chanSetForState,
+	})
 }
 
 // genRandomChannel is a rapid generator for creating random channel edge infos.

graph/db/sql_store.go

@@ -70,7 +70,7 @@ type SQLQueries interface {
 	UpsertNodeExtraType(ctx context.Context, arg sqlc.UpsertNodeExtraTypeParams) error
 	DeleteExtraNodeType(ctx context.Context, arg sqlc.DeleteExtraNodeTypeParams) error
 
-	InsertNodeAddress(ctx context.Context, arg sqlc.InsertNodeAddressParams) error
+	UpsertNodeAddress(ctx context.Context, arg sqlc.UpsertNodeAddressParams) error
 	GetNodeAddresses(ctx context.Context, nodeID int64) ([]sqlc.GetNodeAddressesRow, error)
 	GetNodeAddressesBatch(ctx context.Context, ids []int64) ([]sqlc.GraphNodeAddress, error)
 	DeleteNodeAddresses(ctx context.Context, nodeID int64) error
@@ -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)
@@ -124,7 +124,7 @@ type SQLQueries interface {
 	GetChannelPolicyByChannelAndNode(ctx context.Context, arg sqlc.GetChannelPolicyByChannelAndNodeParams) (sqlc.GraphChannelPolicy, error)
 	GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error)
 
-	InsertChanPolicyExtraType(ctx context.Context, arg sqlc.InsertChanPolicyExtraTypeParams) error
+	UpsertChanPolicyExtraType(ctx context.Context, arg sqlc.UpsertChanPolicyExtraTypeParams) error
 	GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]sqlc.GetChannelPolicyExtraTypesBatchRow, error)
 	DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error
 
@@ -153,6 +153,18 @@ type SQLQueries interface {
 	InsertClosedChannel(ctx context.Context, scid []byte) error
 	IsClosedChannel(ctx context.Context, scid []byte) (bool, error)
 	GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error)
+
+	/*
+		Migration specific queries.
+
+		NOTE: these should not be used in code other than migrations.
+		Once sqldbv2 is in place, these can be removed from this struct
+		as then migrations will have their own dedicated queries
+		structs.
+	*/
+	InsertNodeMig(ctx context.Context, arg sqlc.InsertNodeMigParams) (int64, error)
+	InsertChannelMig(ctx context.Context, arg sqlc.InsertChannelMigParams) (int64, error)
+	InsertEdgePolicyMig(ctx context.Context, arg sqlc.InsertEdgePolicyMigParams) (int64, error)
 }
 
 // BatchedSQLQueries is a version of SQLQueries that's capable of batched
@@ -617,9 +629,7 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context,
 					err)
 			}
 
-			_, err = insertChannel(ctx, tx, edge)
+			return insertChannel(ctx, tx, edge)
-
-			return err
 		},
 		OnCommit: func(err error) error {
 			switch {
@@ -3523,23 +3533,11 @@ const (
 	addressTypeOpaque dbAddressType = math.MaxInt8
 )
 
-// upsertNodeAddresses updates the node's addresses in the database. This
+// collectAddressRecords collects the addresses from the provided
-// includes deleting any existing addresses and inserting the new set of
+// net.Addr slice and returns a map of dbAddressType to a slice of address
-// addresses. The deletion is necessary since the ordering of the addresses may
+// strings.
-// change, and we need to ensure that the database reflects the latest set of
+func collectAddressRecords(addresses []net.Addr) (map[dbAddressType][]string,
-// addresses so that at the time of reconstructing the node announcement, the
+	error) {
-// order is preserved and the signature over the message remains valid.
-func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
-	addresses []net.Addr) error {
-
-	// Delete any existing addresses for the node. This is required since
-	// even if the new set of addresses is the same, the ordering may have
-	// changed for a given address type.
-	err := db.DeleteNodeAddresses(ctx, nodeID)
-	if err != nil {
-		return fmt.Errorf("unable to delete node(%d) addresses: %w",
-			nodeID, err)
-	}
 
 	// Copy the nodes latest set of addresses.
 	newAddresses := map[dbAddressType][]string{
@@ -3561,8 +3559,8 @@ func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
 			} else if ip6 := addr.IP.To16(); ip6 != nil {
 				addAddr(addressTypeIPv6, addr)
 			} else {
-				return fmt.Errorf("unhandled IP address: %v",
+				return nil, fmt.Errorf("unhandled IP "+
-					addr)
+					"address: %v", addr)
 			}
 
 		case *tor.OnionAddr:
@@ -3572,24 +3570,51 @@ func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
 			case tor.V3Len:
 				addAddr(addressTypeTorV3, addr)
 			default:
-				return fmt.Errorf("invalid length for a tor " +
+				return nil, fmt.Errorf("invalid length for " +
-					"address")
+					"a tor address")
 			}
 
 		case *lnwire.OpaqueAddrs:
 			addAddr(addressTypeOpaque, addr)
 
 		default:
-			return fmt.Errorf("unhandled address type: %T", addr)
+			return nil, fmt.Errorf("unhandled address type: %T",
+				addr)
 		}
 	}
 
+	return newAddresses, nil
+}
+
+// upsertNodeAddresses updates the node's addresses in the database. This
+// includes deleting any existing addresses and inserting the new set of
+// addresses. The deletion is necessary since the ordering of the addresses may
+// change, and we need to ensure that the database reflects the latest set of
+// addresses so that at the time of reconstructing the node announcement, the
+// order is preserved and the signature over the message remains valid.
+func upsertNodeAddresses(ctx context.Context, db SQLQueries, nodeID int64,
+	addresses []net.Addr) error {
+
+	// Delete any existing addresses for the node. This is required since
+	// even if the new set of addresses is the same, the ordering may have
+	// changed for a given address type.
+	err := db.DeleteNodeAddresses(ctx, nodeID)
+	if err != nil {
+		return fmt.Errorf("unable to delete node(%d) addresses: %w",
+			nodeID, err)
+	}
+
+	newAddresses, err := collectAddressRecords(addresses)
+	if err != nil {
+		return err
+	}
+
 	// Any remaining entries in newAddresses are new addresses that need to
 	// be added to the database for the first time.
 	for addrType, addrList := range newAddresses {
 		for position, addr := range addrList {
-			err := db.InsertNodeAddress(
+			err := db.UpsertNodeAddress(
-				ctx, sqlc.InsertNodeAddressParams{
+				ctx, sqlc.UpsertNodeAddressParams{
 					NodeID:   nodeID,
 					Type:     int16(addrType),
 					Address:  addr,
@@ -3774,28 +3799,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
@@ -3826,7 +3843,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.
@@ -3838,7 +3855,7 @@ 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)
 		}
 	}
@@ -3846,30 +3863,26 @@ func insertChannel(ctx context.Context, db SQLQueries,
 	// 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 "+
+		return fmt.Errorf("unable to marshal extra opaque data: %w",
-			"data: %w", err)
+			err)
 	}
 
 	for tlvType, value := range extra {
-		err := db.CreateChannelExtraType(
+		err := db.UpsertChannelExtraType(
-			ctx, sqlc.CreateChannelExtraTypeParams{
+			ctx, sqlc.UpsertChannelExtraTypeParams{
 				ChannelID: dbChanID,
 				Type:      int64(tlvType),
 				Value:     value,
 			},
 		)
 		if err != nil {
-			return nil, fmt.Errorf("unable to upsert "+
+			return fmt.Errorf("unable to upsert channel(%d) "+
-				"channel(%d) extra signed field(%v): %w",
+				"extra signed field(%v): %w", edge.ChannelID,
-				edge.ChannelID, tlvType, err)
+				tlvType, err)
 		}
 	}
 
-	return &dbChanInfo{
+	return nil
-		channelID: dbChanID,
-		node1ID:   node1DBID,
-		node2ID:   node2DBID,
-	}, nil
 }
 
 // maybeCreateShellNode checks if a shell node entry exists for the
@@ -3921,8 +3934,8 @@ func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries,
 
 	// Insert all new extra signed fields for the channel policy.
 	for tlvType, value := range extraFields {
-		err = db.InsertChanPolicyExtraType(
+		err = db.UpsertChanPolicyExtraType(
-			ctx, sqlc.InsertChanPolicyExtraTypeParams{
+			ctx, sqlc.UpsertChanPolicyExtraTypeParams{
 				ChannelPolicyID: chanPolicyID,
 				Type:            int64(tlvType),
 				Value:           value,

sqldb/sqlc/graph.sql.go

@@ -120,29 +120,6 @@ func (q *Queries) CreateChannel(ctx context.Context, arg CreateChannelParams) (i
 	return id, err
 }
 
-const createChannelExtraType = `-- name: CreateChannelExtraType :exec
-/* ─────────────────────────────────────────────
-   graph_channel_extra_types table queries
-   ─────────────────────────────────────────────
-*/
-
-INSERT INTO graph_channel_extra_types (
-    channel_id, type, value
-)
-VALUES ($1, $2, $3)
-`
-
-type CreateChannelExtraTypeParams struct {
-	ChannelID int64
-	Type      int64
-	Value     []byte
-}
-
-func (q *Queries) CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error {
-	_, err := q.db.ExecContext(ctx, createChannelExtraType, arg.ChannelID, arg.Type, arg.Value)
-	return err
-}
-
 const deleteChannelPolicyExtraTypes = `-- name: DeleteChannelPolicyExtraTypes :exec
 DELETE FROM graph_channel_policy_extra_types
 WHERE channel_policy_id = $1
@@ -2327,29 +2304,6 @@ func (q *Queries) HighestSCID(ctx context.Context, version int16) ([]byte, error
 	return scid, err
 }
 
-const insertChanPolicyExtraType = `-- name: InsertChanPolicyExtraType :exec
-/* ─────────────────────────────────────────────
-   graph_channel_policy_extra_types table queries
-   ─────────────────────────────────────────────
-*/
-
-INSERT INTO graph_channel_policy_extra_types (
-    channel_policy_id, type, value
-)
-VALUES ($1, $2, $3)
-`
-
-type InsertChanPolicyExtraTypeParams struct {
-	ChannelPolicyID int64
-	Type            int64
-	Value           []byte
-}
-
-func (q *Queries) InsertChanPolicyExtraType(ctx context.Context, arg InsertChanPolicyExtraTypeParams) error {
-	_, err := q.db.ExecContext(ctx, insertChanPolicyExtraType, arg.ChannelPolicyID, arg.Type, arg.Value)
-	return err
-}
-
 const insertChannelFeature = `-- name: InsertChannelFeature :exec
 /* ─────────────────────────────────────────────
    graph_channel_features table queries
@@ -2360,7 +2314,9 @@ INSERT INTO graph_channel_features (
     channel_id, feature_bit
 ) VALUES (
     $1, $2
-)
+) ON CONFLICT (channel_id, feature_bit)
+    -- Do nothing if the channel_id and feature_bit already exist.
+    DO NOTHING
 `
 
 type InsertChannelFeatureParams struct {
@@ -2373,6 +2329,72 @@ func (q *Queries) InsertChannelFeature(ctx context.Context, arg InsertChannelFea
 	return err
 }
 
+const insertChannelMig = `-- name: InsertChannelMig :one
+INSERT INTO graph_channels (
+    version, scid, node_id_1, node_id_2,
+    outpoint, capacity, bitcoin_key_1, bitcoin_key_2,
+    node_1_signature, node_2_signature, bitcoin_1_signature,
+    bitcoin_2_signature
+) VALUES (
+    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12
+) ON CONFLICT (scid, version)
+    -- If a conflict occurs, we have already migrated this channel. However, we
+    -- still need to do an "UPDATE SET" here instead of "DO NOTHING" because
+    -- otherwise, the "RETURNING id" part does not work.
+    DO UPDATE SET
+        node_id_1 = EXCLUDED.node_id_1,
+        node_id_2 = EXCLUDED.node_id_2,
+        outpoint = EXCLUDED.outpoint,
+        capacity = EXCLUDED.capacity,
+        bitcoin_key_1 = EXCLUDED.bitcoin_key_1,
+        bitcoin_key_2 = EXCLUDED.bitcoin_key_2,
+        node_1_signature = EXCLUDED.node_1_signature,
+        node_2_signature = EXCLUDED.node_2_signature,
+        bitcoin_1_signature = EXCLUDED.bitcoin_1_signature,
+        bitcoin_2_signature = EXCLUDED.bitcoin_2_signature
+RETURNING id
+`
+
+type InsertChannelMigParams struct {
+	Version           int16
+	Scid              []byte
+	NodeID1           int64
+	NodeID2           int64
+	Outpoint          string
+	Capacity          sql.NullInt64
+	BitcoinKey1       []byte
+	BitcoinKey2       []byte
+	Node1Signature    []byte
+	Node2Signature    []byte
+	Bitcoin1Signature []byte
+	Bitcoin2Signature []byte
+}
+
+// NOTE: This query is only meant to be used by the graph SQL migration since
+// for that migration, in order to be retry-safe, we don't want to error out if
+// we re-insert the same channel again (which would error if the normal
+// CreateChannel query is used because of the uniqueness constraint on the scid
+// and version columns).
+func (q *Queries) InsertChannelMig(ctx context.Context, arg InsertChannelMigParams) (int64, error) {
+	row := q.db.QueryRowContext(ctx, insertChannelMig,
+		arg.Version,
+		arg.Scid,
+		arg.NodeID1,
+		arg.NodeID2,
+		arg.Outpoint,
+		arg.Capacity,
+		arg.BitcoinKey1,
+		arg.BitcoinKey2,
+		arg.Node1Signature,
+		arg.Node2Signature,
+		arg.Bitcoin1Signature,
+		arg.Bitcoin2Signature,
+	)
+	var id int64
+	err := row.Scan(&id)
+	return id, err
+}
+
 const insertClosedChannel = `-- name: InsertClosedChannel :exec
 /* ─────────────────────────────────────────────
    graph_closed_scid table queries
@@ -2389,37 +2411,80 @@ func (q *Queries) InsertClosedChannel(ctx context.Context, scid []byte) error {
 	return err
 }
 
-const insertNodeAddress = `-- name: InsertNodeAddress :exec
+const insertEdgePolicyMig = `-- name: InsertEdgePolicyMig :one
-/* ─────────────────────────────────────────────
+INSERT INTO graph_channel_policies (
-   graph_node_addresses table queries
+    version, channel_id, node_id, timelock, fee_ppm,
-   ───────────────────────────────────��─────────
+    base_fee_msat, min_htlc_msat, last_update, disabled,
-*/
+    max_htlc_msat, inbound_base_fee_msat,
-
+    inbound_fee_rate_milli_msat, message_flags, channel_flags,
-INSERT INTO graph_node_addresses (
+    signature
-    node_id,
+) VALUES  (
-    type,
+    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15
-    address,
+)
-    position
+ON CONFLICT (channel_id, node_id, version)
-) VALUES (
+    -- If a conflict occurs, we have already migrated this policy. However, we
-    $1, $2, $3, $4
+    -- still need to do an "UPDATE SET" here instead of "DO NOTHING" because
- )
+    -- otherwise, the "RETURNING id" part does not work.
+    DO UPDATE SET
+        timelock = EXCLUDED.timelock,
+        fee_ppm = EXCLUDED.fee_ppm,
+        base_fee_msat = EXCLUDED.base_fee_msat,
+        min_htlc_msat = EXCLUDED.min_htlc_msat,
+        last_update = EXCLUDED.last_update,
+        disabled = EXCLUDED.disabled,
+        max_htlc_msat = EXCLUDED.max_htlc_msat,
+        inbound_base_fee_msat = EXCLUDED.inbound_base_fee_msat,
+        inbound_fee_rate_milli_msat = EXCLUDED.inbound_fee_rate_milli_msat,
+        message_flags = EXCLUDED.message_flags,
+        channel_flags = EXCLUDED.channel_flags,
+        signature = EXCLUDED.signature
+RETURNING id
 `
 
-type InsertNodeAddressParams struct {
+type InsertEdgePolicyMigParams struct {
-	NodeID   int64
+	Version                 int16
-	Type     int16
+	ChannelID               int64
-	Address  string
+	NodeID                  int64
-	Position int32
+	Timelock                int32
+	FeePpm                  int64
+	BaseFeeMsat             int64
+	MinHtlcMsat             int64
+	LastUpdate              sql.NullInt64
+	Disabled                sql.NullBool
+	MaxHtlcMsat             sql.NullInt64
+	InboundBaseFeeMsat      sql.NullInt64
+	InboundFeeRateMilliMsat sql.NullInt64
+	MessageFlags            sql.NullInt16
+	ChannelFlags            sql.NullInt16
+	Signature               []byte
 }
 
-func (q *Queries) InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error {
+// NOTE: This query is only meant to be used by the graph SQL migration since
-	_, err := q.db.ExecContext(ctx, insertNodeAddress,
+// for that migration, in order to be retry-safe, we don't want to error out if
+// we re-insert the same policy (which would error if the normal
+// UpsertEdgePolicy query is used because of the constraint in that query that
+// requires a policy update to have a newer last_update than the existing one).
+func (q *Queries) InsertEdgePolicyMig(ctx context.Context, arg InsertEdgePolicyMigParams) (int64, error) {
+	row := q.db.QueryRowContext(ctx, insertEdgePolicyMig,
+		arg.Version,
+		arg.ChannelID,
 		arg.NodeID,
-		arg.Type,
+		arg.Timelock,
-		arg.Address,
+		arg.FeePpm,
-		arg.Position,
+		arg.BaseFeeMsat,
+		arg.MinHtlcMsat,
+		arg.LastUpdate,
+		arg.Disabled,
+		arg.MaxHtlcMsat,
+		arg.InboundBaseFeeMsat,
+		arg.InboundFeeRateMilliMsat,
+		arg.MessageFlags,
+		arg.ChannelFlags,
+		arg.Signature,
 	)
-	return err
+	var id int64
+	err := row.Scan(&id)
+	return id, err
 }
 
 const insertNodeFeature = `-- name: InsertNodeFeature :exec
@@ -2432,7 +2497,9 @@ INSERT INTO graph_node_features (
     node_id, feature_bit
 ) VALUES (
     $1, $2
-)
+) ON CONFLICT (node_id, feature_bit)
+    -- Do nothing if the feature already exists for the node.
+    DO NOTHING
 `
 
 type InsertNodeFeatureParams struct {
@@ -2445,6 +2512,61 @@ func (q *Queries) InsertNodeFeature(ctx context.Context, arg InsertNodeFeaturePa
 	return err
 }
 
+const insertNodeMig = `-- name: InsertNodeMig :one
+/* ─────────────────────────────────────────────
+   Migration specific queries
+
+   NOTE: once sqldbv2 is in place, these queries can be contained to a package
+   dedicated to the migration that requires it, and so we can then remove
+   it from the main set of "live" queries that the code-base has access to.
+   ────────────────────────────────────────────-
+*/
+
+INSERT INTO graph_nodes (
+    version, pub_key, alias, last_update, color, signature
+) VALUES (
+    $1, $2, $3, $4, $5, $6
+)
+ON CONFLICT (pub_key, version)
+    -- If a conflict occurs, we have already migrated this node. However, we
+    -- still need to do an "UPDATE SET" here instead of "DO NOTHING" because
+    -- otherwise, the "RETURNING id" part does not work.
+    DO UPDATE SET
+        alias = EXCLUDED.alias,
+        last_update = EXCLUDED.last_update,
+        color = EXCLUDED.color,
+        signature = EXCLUDED.signature
+RETURNING id
+`
+
+type InsertNodeMigParams struct {
+	Version    int16
+	PubKey     []byte
+	Alias      sql.NullString
+	LastUpdate sql.NullInt64
+	Color      sql.NullString
+	Signature  []byte
+}
+
+// NOTE: This query is only meant to be used by the graph SQL migration since
+// for that migration, in order to be retry-safe, we don't want to error out if
+// we re-insert the same node (which would error if the normal UpsertNode query
+// is used because of the constraint in that query that requires a node update
+// to have a newer last_update than the existing node).
+func (q *Queries) InsertNodeMig(ctx context.Context, arg InsertNodeMigParams) (int64, error) {
+	row := q.db.QueryRowContext(ctx, insertNodeMig,
+		arg.Version,
+		arg.PubKey,
+		arg.Alias,
+		arg.LastUpdate,
+		arg.Color,
+		arg.Signature,
+	)
+	var id int64
+	err := row.Scan(&id)
+	return id, err
+}
+
 const isClosedChannel = `-- name: IsClosedChannel :one
 SELECT EXISTS (
     SELECT 1
@@ -3284,6 +3406,59 @@ func (q *Queries) ListNodesPaginated(ctx context.Context, arg ListNodesPaginated
 	return items, nil
 }
 
+const upsertChanPolicyExtraType = `-- name: UpsertChanPolicyExtraType :exec
+/* ─────────────────────────────────────────────
+   graph_channel_policy_extra_types table queries
+   ─────────────────────────────────────────────
+*/
+
+INSERT INTO graph_channel_policy_extra_types (
+    channel_policy_id, type, value
+)
+VALUES ($1, $2, $3)
+ON CONFLICT (channel_policy_id, type)
+    -- If a conflict occurs on channel_policy_id and type, then we update the
+    -- value.
+    DO UPDATE SET value = EXCLUDED.value
+`
+
+type UpsertChanPolicyExtraTypeParams struct {
+	ChannelPolicyID int64
+	Type            int64
+	Value           []byte
+}
+
+func (q *Queries) UpsertChanPolicyExtraType(ctx context.Context, arg UpsertChanPolicyExtraTypeParams) error {
+	_, err := q.db.ExecContext(ctx, upsertChanPolicyExtraType, arg.ChannelPolicyID, arg.Type, arg.Value)
+	return err
+}
+
+const upsertChannelExtraType = `-- name: UpsertChannelExtraType :exec
+/* ─────────────────────────────────────────────
+   graph_channel_extra_types table queries
+   ─────────────────────────────────────────────
+*/
+
+INSERT INTO graph_channel_extra_types (
+    channel_id, type, value
+)
+VALUES ($1, $2, $3)
+    ON CONFLICT (channel_id, type)
+    -- Update the value if a conflict occurs on channel_id and type.
+    DO UPDATE SET value = EXCLUDED.value
+`
+
+type UpsertChannelExtraTypeParams struct {
+	ChannelID int64
+	Type      int64
+	Value     []byte
+}
+
+func (q *Queries) UpsertChannelExtraType(ctx context.Context, arg UpsertChannelExtraTypeParams) error {
+	_, err := q.db.ExecContext(ctx, upsertChannelExtraType, arg.ChannelID, arg.Type, arg.Value)
+	return err
+}
+
 const upsertEdgePolicy = `-- name: UpsertEdgePolicy :one
 /* ─────────────────────────────────────────────
    graph_channel_policies table queries
@@ -3407,6 +3582,40 @@ func (q *Queries) UpsertNode(ctx context.Context, arg UpsertNodeParams) (int64,
 	return id, err
 }
 
+const upsertNodeAddress = `-- name: UpsertNodeAddress :exec
+/* ─────────────────────────────────────────────
+   graph_node_addresses table queries
+   ───────────────────────────────────��─────────
+*/
+
+INSERT INTO graph_node_addresses (
+    node_id,
+    type,
+    address,
+    position
+) VALUES (
+    $1, $2, $3, $4
+) ON CONFLICT (node_id, type, position)
+    DO UPDATE SET address = EXCLUDED.address
+`
+
+type UpsertNodeAddressParams struct {
+	NodeID   int64
+	Type     int16
+	Address  string
+	Position int32
+}
+
+func (q *Queries) UpsertNodeAddress(ctx context.Context, arg UpsertNodeAddressParams) error {
+	_, err := q.db.ExecContext(ctx, upsertNodeAddress,
+		arg.NodeID,
+		arg.Type,
+		arg.Address,
+		arg.Position,
+	)
+	return err
+}
+
 const upsertNodeExtraType = `-- name: UpsertNodeExtraType :exec
 /* ─────────────────────────────────────────────
    graph_node_extra_types table queries

sqldb/sqlc/querier.go

@@ -16,7 +16,6 @@ type Querier interface {
 	ClearKVInvoiceHashIndex(ctx context.Context) error
 	CountZombieChannels(ctx context.Context, version int16) (int64, error)
 	CreateChannel(ctx context.Context, arg CreateChannelParams) (int64, error)
-	CreateChannelExtraType(ctx context.Context, arg CreateChannelExtraTypeParams) error
 	DeleteCanceledInvoices(ctx context.Context) (sql.Result, error)
 	DeleteChannelPolicyExtraTypes(ctx context.Context, channelPolicyID int64) error
 	DeleteChannels(ctx context.Context, ids []int64) error
@@ -88,17 +87,33 @@ type Querier interface {
 	HighestSCID(ctx context.Context, version int16) ([]byte, error)
 	InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error
 	InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error
-	InsertChanPolicyExtraType(ctx context.Context, arg InsertChanPolicyExtraTypeParams) error
 	InsertChannelFeature(ctx context.Context, arg InsertChannelFeatureParams) error
+	// NOTE: This query is only meant to be used by the graph SQL migration since
+	// for that migration, in order to be retry-safe, we don't want to error out if
+	// we re-insert the same channel again (which would error if the normal
+	// CreateChannel query is used because of the uniqueness constraint on the scid
+	// and version columns).
+	InsertChannelMig(ctx context.Context, arg InsertChannelMigParams) (int64, error)
 	InsertClosedChannel(ctx context.Context, scid []byte) error
+	// NOTE: This query is only meant to be used by the graph SQL migration since
+	// for that migration, in order to be retry-safe, we don't want to error out if
+	// we re-insert the same policy (which would error if the normal
+	// UpsertEdgePolicy query is used because of the constraint in that query that
+	// requires a policy update to have a newer last_update than the existing one).
+	InsertEdgePolicyMig(ctx context.Context, arg InsertEdgePolicyMigParams) (int64, error)
 	InsertInvoice(ctx context.Context, arg InsertInvoiceParams) (int64, error)
 	InsertInvoiceFeature(ctx context.Context, arg InsertInvoiceFeatureParams) error
 	InsertInvoiceHTLC(ctx context.Context, arg InsertInvoiceHTLCParams) (int64, error)
 	InsertInvoiceHTLCCustomRecord(ctx context.Context, arg InsertInvoiceHTLCCustomRecordParams) error
 	InsertKVInvoiceKeyAndAddIndex(ctx context.Context, arg InsertKVInvoiceKeyAndAddIndexParams) error
 	InsertMigratedInvoice(ctx context.Context, arg InsertMigratedInvoiceParams) (int64, error)
-	InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error
 	InsertNodeFeature(ctx context.Context, arg InsertNodeFeatureParams) error
+	// NOTE: This query is only meant to be used by the graph SQL migration since
+	// for that migration, in order to be retry-safe, we don't want to error out if
+	// we re-insert the same node (which would error if the normal UpsertNode query
+	// is used because of the constraint in that query that requires a node update
+	// to have a newer last_update than the existing node).
+	InsertNodeMig(ctx context.Context, arg InsertNodeMigParams) (int64, error)
 	IsClosedChannel(ctx context.Context, scid []byte) (bool, error)
 	IsPublicV1Node(ctx context.Context, pubKey []byte) (bool, error)
 	IsZombieChannel(ctx context.Context, arg IsZombieChannelParams) (bool, error)
@@ -125,8 +140,11 @@ type Querier interface {
 	UpdateInvoiceHTLCs(ctx context.Context, arg UpdateInvoiceHTLCsParams) error
 	UpdateInvoiceState(ctx context.Context, arg UpdateInvoiceStateParams) (sql.Result, error)
 	UpsertAMPSubInvoice(ctx context.Context, arg UpsertAMPSubInvoiceParams) (sql.Result, error)
+	UpsertChanPolicyExtraType(ctx context.Context, arg UpsertChanPolicyExtraTypeParams) error
+	UpsertChannelExtraType(ctx context.Context, arg UpsertChannelExtraTypeParams) error
 	UpsertEdgePolicy(ctx context.Context, arg UpsertEdgePolicyParams) (int64, error)
 	UpsertNode(ctx context.Context, arg UpsertNodeParams) (int64, error)
+	UpsertNodeAddress(ctx context.Context, arg UpsertNodeAddressParams) error
 	UpsertNodeExtraType(ctx context.Context, arg UpsertNodeExtraTypeParams) error
 	UpsertPruneLogEntry(ctx context.Context, arg UpsertPruneLogEntryParams) error
 	UpsertZombieChannel(ctx context.Context, arg UpsertZombieChannelParams) error

sqldb/sqlc/queries/graph.sql

@@ -105,7 +105,9 @@ INSERT INTO graph_node_features (
     node_id, feature_bit
 ) VALUES (
     $1, $2
-);
+) ON CONFLICT (node_id, feature_bit)
+    -- Do nothing if the feature already exists for the node.
+    DO NOTHING;
 
 -- name: GetNodeFeatures :many
 SELECT *
@@ -135,7 +137,7 @@ WHERE node_id = $1
    ───────────────────────────────────��─────────
 */
 
--- name: InsertNodeAddress :exec
+-- name: UpsertNodeAddress :exec
 INSERT INTO graph_node_addresses (
     node_id,
     type,
@@ -143,7 +145,8 @@ INSERT INTO graph_node_addresses (
     position
 ) VALUES (
     $1, $2, $3, $4
- );
+) ON CONFLICT (node_id, type, position)
+    DO UPDATE SET address = EXCLUDED.address;
 
 -- name: GetNodeAddresses :many
 SELECT type, address
@@ -735,7 +738,9 @@ INSERT INTO graph_channel_features (
     channel_id, feature_bit
 ) VALUES (
     $1, $2
-);
+) ON CONFLICT (channel_id, feature_bit)
+    -- Do nothing if the channel_id and feature_bit already exist.
+    DO NOTHING;
 
 -- name: GetChannelFeaturesBatch :many
 SELECT
@@ -750,11 +755,14 @@ ORDER BY channel_id, feature_bit;
    ─────────────────────────────────────────────
 */
 
--- name: CreateChannelExtraType :exec
+-- name: UpsertChannelExtraType :exec
 INSERT INTO graph_channel_extra_types (
     channel_id, type, value
 )
-VALUES ($1, $2, $3);
+VALUES ($1, $2, $3)
+    ON CONFLICT (channel_id, type)
+    -- Update the value if a conflict occurs on channel_id and type.
+    DO UPDATE SET value = EXCLUDED.value;
 
 -- name: GetChannelExtrasBatch :many
 SELECT
@@ -861,11 +869,15 @@ WHERE c.scid = @scid
    ─────────────────────────────────────────────
 */
 
--- name: InsertChanPolicyExtraType :exec
+-- name: UpsertChanPolicyExtraType :exec
 INSERT INTO graph_channel_policy_extra_types (
     channel_policy_id, type, value
 )
-VALUES ($1, $2, $3);
+VALUES ($1, $2, $3)
+ON CONFLICT (channel_policy_id, type)
+    -- If a conflict occurs on channel_policy_id and type, then we update the
+    -- value.
+    DO UPDATE SET value = EXCLUDED.value;
 
 -- name: GetChannelPolicyExtraTypesBatch :many
 SELECT
@@ -995,3 +1007,98 @@ SELECT EXISTS (
 SELECT scid
 FROM graph_closed_scids
 WHERE scid IN (sqlc.slice('scids')/*SLICE:scids*/);
+
+/* ─────────────────────────────────────────────
+   Migration specific queries
+
+   NOTE: once sqldbv2 is in place, these queries can be contained to a package
+   dedicated to the migration that requires it, and so we can then remove
+   it from the main set of "live" queries that the code-base has access to.
+   ────────────────────────────────────────────-
+*/
+
+-- NOTE: This query is only meant to be used by the graph SQL migration since
+-- for that migration, in order to be retry-safe, we don't want to error out if
+-- we re-insert the same node (which would error if the normal UpsertNode query
+-- is used because of the constraint in that query that requires a node update
+-- to have a newer last_update than the existing node).
+-- name: InsertNodeMig :one
+INSERT INTO graph_nodes (
+    version, pub_key, alias, last_update, color, signature
+) VALUES (
+    $1, $2, $3, $4, $5, $6
+)
+ON CONFLICT (pub_key, version)
+    -- If a conflict occurs, we have already migrated this node. However, we
+    -- still need to do an "UPDATE SET" here instead of "DO NOTHING" because
+    -- otherwise, the "RETURNING id" part does not work.
+    DO UPDATE SET
+        alias = EXCLUDED.alias,
+        last_update = EXCLUDED.last_update,
+        color = EXCLUDED.color,
+        signature = EXCLUDED.signature
+RETURNING id;
+
+-- NOTE: This query is only meant to be used by the graph SQL migration since
+-- for that migration, in order to be retry-safe, we don't want to error out if
+-- we re-insert the same channel again (which would error if the normal
+-- CreateChannel query is used because of the uniqueness constraint on the scid
+-- and version columns).
+-- name: InsertChannelMig :one
+INSERT INTO graph_channels (
+    version, scid, node_id_1, node_id_2,
+    outpoint, capacity, bitcoin_key_1, bitcoin_key_2,
+    node_1_signature, node_2_signature, bitcoin_1_signature,
+    bitcoin_2_signature
+) VALUES (
+    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12
+) ON CONFLICT (scid, version)
+    -- If a conflict occurs, we have already migrated this channel. However, we
+    -- still need to do an "UPDATE SET" here instead of "DO NOTHING" because
+    -- otherwise, the "RETURNING id" part does not work.
+    DO UPDATE SET
+        node_id_1 = EXCLUDED.node_id_1,
+        node_id_2 = EXCLUDED.node_id_2,
+        outpoint = EXCLUDED.outpoint,
+        capacity = EXCLUDED.capacity,
+        bitcoin_key_1 = EXCLUDED.bitcoin_key_1,
+        bitcoin_key_2 = EXCLUDED.bitcoin_key_2,
+        node_1_signature = EXCLUDED.node_1_signature,
+        node_2_signature = EXCLUDED.node_2_signature,
+        bitcoin_1_signature = EXCLUDED.bitcoin_1_signature,
+        bitcoin_2_signature = EXCLUDED.bitcoin_2_signature
+RETURNING id;
+
+-- NOTE: This query is only meant to be used by the graph SQL migration since
+-- for that migration, in order to be retry-safe, we don't want to error out if
+-- we re-insert the same policy (which would error if the normal
+-- UpsertEdgePolicy query is used because of the constraint in that query that
+-- requires a policy update to have a newer last_update than the existing one).
+-- name: InsertEdgePolicyMig :one
+INSERT INTO graph_channel_policies (
+    version, channel_id, node_id, timelock, fee_ppm,
+    base_fee_msat, min_htlc_msat, last_update, disabled,
+    max_htlc_msat, inbound_base_fee_msat,
+    inbound_fee_rate_milli_msat, message_flags, channel_flags,
+    signature
+) VALUES  (
+    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15
+)
+ON CONFLICT (channel_id, node_id, version)
+    -- If a conflict occurs, we have already migrated this policy. However, we
+    -- still need to do an "UPDATE SET" here instead of "DO NOTHING" because
+    -- otherwise, the "RETURNING id" part does not work.
+    DO UPDATE SET
+        timelock = EXCLUDED.timelock,
+        fee_ppm = EXCLUDED.fee_ppm,
+        base_fee_msat = EXCLUDED.base_fee_msat,
+        min_htlc_msat = EXCLUDED.min_htlc_msat,
+        last_update = EXCLUDED.last_update,
+        disabled = EXCLUDED.disabled,
+        max_htlc_msat = EXCLUDED.max_htlc_msat,
+        inbound_base_fee_msat = EXCLUDED.inbound_base_fee_msat,
+        inbound_fee_rate_milli_msat = EXCLUDED.inbound_fee_rate_milli_msat,
+        message_flags = EXCLUDED.message_flags,
+        channel_flags = EXCLUDED.channel_flags,
+        signature = EXCLUDED.signature
+RETURNING id;