From 9019bcaddff51438fda8919f9afd48614938b0e3 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 10:28:30 +0200 Subject: [PATCH 01/10] graph/db: let test policy have some extra opaque data This will help us test idempotency later on, but it also ensures that TestMigrateGraphToSQL is properly testing writes to the graph_channel_policy_extra_types table. --- graph/db/sql_migration_test.go | 1 + 1 file changed, 1 insertion(+) diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index c555e433d..84f7e99ea 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -716,6 +716,7 @@ func makeTestPolicy(chanID uint64, toNode route.Vertex, isNode1 bool, FeeBaseMSat: math.MaxUint64, FeeProportionalMillionths: math.MaxUint64, ToNode: toNode, + ExtraOpaqueData: testExtraData, } for _, opt := range opts { From 68e4970fbbac284a7773e8fffc3e2b62be80f40a Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 09:00:29 +0200 Subject: [PATCH 02/10] graph/db: let migration test test retry safety Currently, the graph SQL migration is not retry safe. Meaning that if the source DB exeutes a retry under the hood, this could result in the migration failing. In preparation for fixing this, we adust the migration test accordingly. --- graph/db/sql_migration_test.go | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 84f7e99ea..a752be424 100644 --- a/graph/db/sql_migration_test.go +++ b/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() @@ -98,6 +105,15 @@ func TestMigrateGraphToSQL(t *testing.T) { write func(t *testing.T, db *KVStore, object any) objects []any expGraphStats graphStats + + // expNotRetrySafety is true if we expect an error to occur for + // the test if the migration is run twice. In other-words, if + // the specific case in question is currently not idempotent. + // + // NOTE: we want _all_ the cases here to be idempotent, so this + // is a temporary field which will be removed once we have + // properly made the migration retry-safe. + expNotRetrySafety bool }{ { name: "empty", @@ -139,6 +155,7 @@ func TestMigrateGraphToSQL(t *testing.T) { expGraphStats: graphStats{ numNodes: 6, }, + expNotRetrySafety: true, }, { name: "source node", @@ -156,6 +173,7 @@ func TestMigrateGraphToSQL(t *testing.T) { numNodes: 1, srcNodeSet: true, }, + expNotRetrySafety: true, }, { name: "channels and policies", @@ -228,6 +246,7 @@ func TestMigrateGraphToSQL(t *testing.T) { numChannels: 3, numPolicies: 3, }, + expNotRetrySafety: true, }, { name: "prune log", @@ -338,6 +357,20 @@ func TestMigrateGraphToSQL(t *testing.T) { // Validate that the two databases are now in sync. assertInSync(t, kvDB, sql, test.expGraphStats) + + // NOTE: for now, not all the cases in the test are + // retry safe! The aim is to completely remove this + // field once we have made the migration retry-safe. + err = MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db) + if !test.expNotRetrySafety { + // The migration should be retry-safe, so + // running it again should not change the state + // of the databases. + require.NoError(t, err) + assertInSync(t, kvDB, sql, test.expGraphStats) + } else { + require.Error(t, err) + } }) } } From aefc9118a404655c392371af58cdd837cfd4bdc4 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 09:59:21 +0200 Subject: [PATCH 03/10] graph/db: migration test for channels with no policies In preparation for making the channel & policy migration logic idempotent in a step-by-step manner, we add a test here that only tests the migration of channels _without_ policies so that we can first focus on just making the channel migration idempotent. --- graph/db/sql_migration_test.go | 54 ++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index a752be424..13e3bff80 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -175,6 +175,60 @@ func TestMigrateGraphToSQL(t *testing.T) { }, expNotRetrySafety: 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, + }, + expNotRetrySafety: true, + }, { name: "channels and policies", write: writeUpdate, From 2c8ac0c92ccee5804235d888a1a3091aec649256 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 09:05:57 +0200 Subject: [PATCH 04/10] graph/db: thread through `reset` call-backs In preparation for handling retries on the source DB side, we thread through the `reset` call-backs properly so that we can reset appropriate variables. --- graph/db/sql_migration.go | 34 +++++++++++++++++++++++++++------- graph/db/sql_migration_test.go | 5 +++-- 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 5aea3d335..45656e64e 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -923,6 +923,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 +983,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 +1005,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 +1099,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 +1285,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 +1313,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 +1332,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 +1351,5 @@ func forEachClosedSCID(db kvdb.Backend, byteOrder.Uint64(k), )) }) - }, func() {}) + }, reset) } diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 13e3bff80..d4bba602d 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -600,6 +600,7 @@ func checkKVPruneLogEntries(t *testing.T, kv *KVStore, sql *SQLStore, return nil }, + func() {}, ) require.NoError(t, err) @@ -631,7 +632,7 @@ func checkClosedSCIDIndex(t *testing.T, kv kvdb.Backend, sql *SQLStore) { require.True(t, closed) return nil - }) + }, func() {}) require.NoError(t, err) } @@ -663,7 +664,7 @@ func checkZombieIndex(t *testing.T, kv kvdb.Backend, sql *SQLStore) { } return nil - }) + }, func() {}) require.NoError(t, err) } From ddea6d59ce44c551c83671e62323322306ffb999 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 09:32:15 +0200 Subject: [PATCH 05/10] graph/db+sqldb: make node migration idempotent In this commit, the graph SQL migration is updated so that the node migration step is retry-safe. This is done by using migration specific logic & queries that do not use the same node-update-constraint as the normal node upsert logic. For normal "run-time" logic, we always expect a node update to have a newer timestamp than any previously stored one. But for the migration, we will only ever be dealing with a single announcement for a given node & to make things retry-safe, we dont want the query to error if we re-insert the exact same node. --- graph/db/sql_migration.go | 76 ++++++++++++++++++++++++++++++++-- graph/db/sql_migration_test.go | 2 - graph/db/sql_store.go | 10 +++++ sqldb/sqlc/graph.sql.go | 55 ++++++++++++++++++++++++ sqldb/sqlc/querier.go | 6 +++ sqldb/sqlc/queries/graph.sql | 31 ++++++++++++++ 6 files changed, 175 insertions(+), 5 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 45656e64e..ae45aa9f5 100644 --- a/graph/db/sql_migration.go +++ b/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 - // migration will be retried from the start. + count = 0 + 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) @@ -1353,3 +1356,70 @@ func forEachClosedSCID(db kvdb.Backend, }) }, 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. +// +// TODO(elle): update the upsert calls in this function to be more efficient. +// since no data collection steps should be required during the migration. +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 + } + + // NOTE: The upserts here will be updated to be more efficient in the + // following commits. + + // Update the node's features. + err = upsertNodeFeatures(ctx, db, nodeID, node.Features) + if err != nil { + return 0, fmt.Errorf("inserting node features: %w", err) + } + + // Update the node's addresses. + err = upsertNodeAddresses(ctx, db, nodeID, node.Addresses) + if err != nil { + return 0, fmt.Errorf("inserting node addresses: %w", 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) + } + + // Update the node's extra signed fields. + err = upsertNodeExtraSignedFields(ctx, db, nodeID, extra) + if err != nil { + return 0, fmt.Errorf("inserting node extra TLVs: %w", err) + } + + return nodeID, nil +} diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index d4bba602d..7826fd1c9 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -155,7 +155,6 @@ func TestMigrateGraphToSQL(t *testing.T) { expGraphStats: graphStats{ numNodes: 6, }, - expNotRetrySafety: true, }, { name: "source node", @@ -173,7 +172,6 @@ func TestMigrateGraphToSQL(t *testing.T) { numNodes: 1, srcNodeSet: true, }, - expNotRetrySafety: true, }, { name: "channel with no policies", diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 956b8406d..449c4f35d 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -153,6 +153,16 @@ 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) } // BatchedSQLQueries is a version of SQLQueries that's capable of batched diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index d27143ca8..7afe8b1b0 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -2445,6 +2445,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 diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index eaef3c777..7696e3651 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -99,6 +99,12 @@ type Querier interface { 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) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 04f814083..82e9b8736 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -995,3 +995,34 @@ 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; From a291d6f1a6bf9c30e41034d2d1ae1f1e7e09d54b Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 09:43:40 +0200 Subject: [PATCH 06/10] graph/db+sqldb: improve efficiency of node migration There is no need to use the "collect-then-update" pattern for node insertion during the SQL migration since if we do have any previously persisted data for the node and happen to re-run the insertion for that node, the data will be exactly the same. So we can make use of "On conflict, no nothing" here too. --- graph/db/sql_migration.go | 61 +++++++++++++++++++++++-------- graph/db/sql_store.go | 65 ++++++++++++++++++++------------- sqldb/sqlc/graph.sql.go | 71 +++++++++++++++++++----------------- sqldb/sqlc/querier.go | 2 +- sqldb/sqlc/queries/graph.sql | 9 +++-- 5 files changed, 129 insertions(+), 79 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index ae45aa9f5..2f4ede4f8 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -1363,9 +1363,6 @@ func forEachClosedSCID(db kvdb.Backend, // 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. -// -// TODO(elle): update the upsert calls in this function to be more efficient. -// since no data collection steps should be required during the migration. func insertNodeSQLMig(ctx context.Context, db SQLQueries, node *models.LightningNode) (int64, error) { @@ -1392,19 +1389,42 @@ func insertNodeSQLMig(ctx context.Context, db SQLQueries, return nodeID, nil } - // NOTE: The upserts here will be updated to be more efficient in the - // following commits. - - // Update the node's features. - err = upsertNodeFeatures(ctx, db, nodeID, node.Features) - if err != nil { - return 0, fmt.Errorf("inserting node features: %w", err) + // 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. - err = upsertNodeAddresses(ctx, db, nodeID, node.Addresses) + newAddresses, err := collectAddressRecords(node.Addresses) if err != nil { - return 0, fmt.Errorf("inserting node addresses: %w", err) + 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 @@ -1415,10 +1435,19 @@ func insertNodeSQLMig(ctx context.Context, db SQLQueries, err) } - // Update the node's extra signed fields. - err = upsertNodeExtraSignedFields(ctx, db, nodeID, extra) - if err != nil { - return 0, fmt.Errorf("inserting node extra TLVs: %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 diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 449c4f35d..ef903ec67 100644 --- a/graph/db/sql_store.go +++ b/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 @@ -3533,23 +3533,11 @@ const ( addressTypeOpaque dbAddressType = math.MaxInt8 ) -// 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) - } +// collectAddressRecords collects the addresses from the provided +// net.Addr slice and returns a map of dbAddressType to a slice of address +// strings. +func collectAddressRecords(addresses []net.Addr) (map[dbAddressType][]string, + error) { // Copy the nodes latest set of addresses. newAddresses := map[dbAddressType][]string{ @@ -3571,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", - addr) + return nil, fmt.Errorf("unhandled IP "+ + "address: %v", addr) } case *tor.OnionAddr: @@ -3582,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 " + - "address") + return nil, fmt.Errorf("invalid length for " + + "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( - ctx, sqlc.InsertNodeAddressParams{ + err := db.UpsertNodeAddress( + ctx, sqlc.UpsertNodeAddressParams{ NodeID: nodeID, Type: int16(addrType), Address: addr, diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 7afe8b1b0..ecf100e44 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -2389,39 +2389,6 @@ func (q *Queries) InsertClosedChannel(ctx context.Context, scid []byte) error { return err } -const insertNodeAddress = `-- name: InsertNodeAddress :exec -/* ───────────────────────────────────────────── - graph_node_addresses table queries - ───────────────────────────────────��───────── -*/ - -INSERT INTO graph_node_addresses ( - node_id, - type, - address, - position -) VALUES ( - $1, $2, $3, $4 - ) -` - -type InsertNodeAddressParams struct { - NodeID int64 - Type int16 - Address string - Position int32 -} - -func (q *Queries) InsertNodeAddress(ctx context.Context, arg InsertNodeAddressParams) error { - _, err := q.db.ExecContext(ctx, insertNodeAddress, - arg.NodeID, - arg.Type, - arg.Address, - arg.Position, - ) - return err -} - const insertNodeFeature = `-- name: InsertNodeFeature :exec /* ───────────────────────────────────────────── graph_node_features table queries @@ -2432,7 +2399,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 { @@ -3462,6 +3431,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 diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 7696e3651..1123d6bc2 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -97,7 +97,6 @@ type Querier interface { 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 @@ -133,6 +132,7 @@ type Querier interface { UpsertAMPSubInvoice(ctx context.Context, arg UpsertAMPSubInvoiceParams) (sql.Result, 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 diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 82e9b8736..8a897003d 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/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 From 8736fcafa885fe5ee66671f94494aad34a3fc4a4 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 10:11:20 +0200 Subject: [PATCH 07/10] 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. --- graph/db/sql_migration.go | 114 ++++++++++++++++++++++++++++++- graph/db/sql_migration_test.go | 1 - graph/db/sql_store.go | 45 +++++-------- sqldb/sqlc/graph.sql.go | 119 ++++++++++++++++++++++++++------- sqldb/sqlc/querier.go | 8 ++- sqldb/sqlc/queries/graph.sql | 41 +++++++++++- 6 files changed, 267 insertions(+), 61 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 2f4ede4f8..61e61526c 100644 --- a/graph/db/sql_migration.go +++ b/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 +} diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 7826fd1c9..aaf70990b 100644 --- a/graph/db/sql_migration_test.go +++ b/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", diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index ef903ec67..1879f6f38 100644 --- a/graph/db/sql_store.go +++ b/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,7 +3854,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) } } @@ -3871,30 +3862,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 "+ - "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 diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index ecf100e44..1e32bc33b 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/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 @@ -2360,7 +2337,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 +2352,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 @@ -3308,6 +3353,32 @@ func (q *Queries) ListNodesPaginated(ctx context.Context, arg ListNodesPaginated return items, nil } +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 diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 1123d6bc2..0011f086a 100644 --- a/sqldb/sqlc/querier.go +++ b/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 @@ -90,6 +89,12 @@ type Querier interface { 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 InsertInvoice(ctx context.Context, arg InsertInvoiceParams) (int64, error) InsertInvoiceFeature(ctx context.Context, arg InsertInvoiceFeatureParams) error @@ -130,6 +135,7 @@ 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) + 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 diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 8a897003d..d8215b22e 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -738,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 @@ -753,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 @@ -1029,3 +1034,33 @@ ON CONFLICT (pub_key, version) 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; From 22bf88e900dc6290194f4f6fb16c4444181434ef Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 10:21:11 +0200 Subject: [PATCH 08/10] graph/db+sqldb: make policy migration idempotent Finally, we make the channel-policy part of the SQL migration idempotent by adding a migration-only policy insert query which will not error out if the policy already exists and does not have a timestamp that is newer than the existing records timestamp. To keep the commit simple, a insertChanEdgePolicyMig function is added which is basically identical to the updateChanEdgePolicy function except for the fact that it uses the newly added query. In the next commit, it will be simplified even more. --- graph/db/sql_migration.go | 101 ++++++++++++++++++++++++++++++++- graph/db/sql_migration_test.go | 26 ++------- graph/db/sql_store.go | 1 + sqldb/sqlc/graph.sql.go | 76 +++++++++++++++++++++++++ sqldb/sqlc/querier.go | 6 ++ sqldb/sqlc/queries/graph.sql | 34 +++++++++++ 6 files changed, 221 insertions(+), 23 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 61e61526c..30e516a6a 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -454,7 +454,7 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, policyCount++ - _, _, _, err := updateChanEdgePolicy(ctx, sqlDB, policy) + _, _, _, err := insertChanEdgePolicyMig(ctx, sqlDB, policy) if err != nil { return fmt.Errorf("could not migrate channel "+ "policy %d: %w", policy.ChannelID, err) @@ -1560,3 +1560,102 @@ func insertChannelMig(ctx context.Context, db SQLQueries, 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. +// +// TODO(elle): update this function to be more performant in the migration +// setting. For the sake of keeping the commit that introduced this function +// simple, this is for now mostly the same as updateChanEdgePolicy. +func insertChanEdgePolicyMig(ctx context.Context, tx SQLQueries, + edge *models.ChannelEdgePolicy) (route.Vertex, route.Vertex, bool, + error) { + + var ( + node1Pub, node2Pub route.Vertex + isNode1 bool + chanIDB = channelIDToBytes(edge.ChannelID) + ) + + // Check that this edge policy refers to a channel that we already + // know of. We do this explicitly so that we can return the appropriate + // ErrEdgeNotFound error if the channel doesn't exist, rather than + // abort the transaction which would abort the entire batch. + dbChan, err := tx.GetChannelAndNodesBySCID( + ctx, sqlc.GetChannelAndNodesBySCIDParams{ + Scid: chanIDB, + Version: int16(ProtocolV1), + }, + ) + if errors.Is(err, sql.ErrNoRows) { + return node1Pub, node2Pub, false, ErrEdgeNotFound + } else if err != nil { + return node1Pub, node2Pub, false, fmt.Errorf("unable to "+ + "fetch channel(%v): %w", edge.ChannelID, err) + } + + copy(node1Pub[:], dbChan.Node1PubKey) + copy(node2Pub[:], dbChan.Node2PubKey) + + // Figure out which node this edge is from. + isNode1 = edge.ChannelFlags&lnwire.ChanUpdateDirection == 0 + nodeID := dbChan.NodeID1 + if !isNode1 { + nodeID = dbChan.NodeID2 + } + + 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.ID, + 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 node1Pub, node2Pub, isNode1, + 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 node1Pub, node2Pub, false, fmt.Errorf("unable to "+ + "marshal extra opaque data: %w", err) + } + + // Update the channel policy's extra signed fields. + err = upsertChanPolicyExtraSignedFields(ctx, tx, id, extra) + if err != nil { + return node1Pub, node2Pub, false, fmt.Errorf("inserting chan "+ + "policy extra TLVs: %w", err) + } + + return node1Pub, node2Pub, isNode1, nil +} diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index aaf70990b..3833b8cc6 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -105,15 +105,6 @@ func TestMigrateGraphToSQL(t *testing.T) { write func(t *testing.T, db *KVStore, object any) objects []any expGraphStats graphStats - - // expNotRetrySafety is true if we expect an error to occur for - // the test if the migration is run twice. In other-words, if - // the specific case in question is currently not idempotent. - // - // NOTE: we want _all_ the cases here to be idempotent, so this - // is a temporary field which will be removed once we have - // properly made the migration retry-safe. - expNotRetrySafety bool }{ { name: "empty", @@ -297,7 +288,6 @@ func TestMigrateGraphToSQL(t *testing.T) { numChannels: 3, numPolicies: 3, }, - expNotRetrySafety: true, }, { name: "prune log", @@ -409,19 +399,11 @@ func TestMigrateGraphToSQL(t *testing.T) { // Validate that the two databases are now in sync. assertInSync(t, kvDB, sql, test.expGraphStats) - // NOTE: for now, not all the cases in the test are - // retry safe! The aim is to completely remove this - // field once we have made the migration retry-safe. + // 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) - if !test.expNotRetrySafety { - // The migration should be retry-safe, so - // running it again should not change the state - // of the databases. - require.NoError(t, err) - assertInSync(t, kvDB, sql, test.expGraphStats) - } else { - require.Error(t, err) - } + require.NoError(t, err) + assertInSync(t, kvDB, sql, test.expGraphStats) }) } } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1879f6f38..60871e17b 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -164,6 +164,7 @@ type SQLQueries interface { */ 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 diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 1e32bc33b..bdc049d85 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -2434,6 +2434,82 @@ func (q *Queries) InsertClosedChannel(ctx context.Context, scid []byte) error { return err } +const insertEdgePolicyMig = `-- 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 +` + +type InsertEdgePolicyMigParams struct { + Version int16 + ChannelID int64 + NodeID int64 + 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 +} + +// 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). +func (q *Queries) InsertEdgePolicyMig(ctx context.Context, arg InsertEdgePolicyMigParams) (int64, error) { + row := q.db.QueryRowContext(ctx, insertEdgePolicyMig, + arg.Version, + arg.ChannelID, + arg.NodeID, + arg.Timelock, + arg.FeePpm, + arg.BaseFeeMsat, + arg.MinHtlcMsat, + arg.LastUpdate, + arg.Disabled, + arg.MaxHtlcMsat, + arg.InboundBaseFeeMsat, + arg.InboundFeeRateMilliMsat, + arg.MessageFlags, + arg.ChannelFlags, + arg.Signature, + ) + var id int64 + err := row.Scan(&id) + return id, err +} + const insertNodeFeature = `-- name: InsertNodeFeature :exec /* ───────────────────────────────────────────── graph_node_features table queries diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 0011f086a..6d38597a4 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -96,6 +96,12 @@ type Querier interface { // 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) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index d8215b22e..7f12cd0c5 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -1064,3 +1064,37 @@ INSERT INTO graph_channels ( 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; \ No newline at end of file From f2ed5564efec50f1714dc4fd68b6282627bf8543 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 10:30:01 +0200 Subject: [PATCH 09/10] graph/db+sqldb: improve performance of chan update sql migration This commit simplifies insertChanEdgePolicyMig. Much of the logic can be removed given that this method is only used in the context of the graph SQL migration. This should improve the performance of the migration quite a lot since it removes the extra GetChannelAndNodesBySCID call. --- graph/db/sql_migration.go | 79 +++++++++++++----------------------- graph/db/sql_store.go | 6 +-- sqldb/sqlc/graph.sql.go | 50 ++++++++++++----------- sqldb/sqlc/querier.go | 2 +- sqldb/sqlc/queries/graph.sql | 8 +++- 5 files changed, 66 insertions(+), 79 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 30e516a6a..b88518fd8 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -440,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 @@ -454,7 +456,7 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, policyCount++ - _, _, _, err := insertChanEdgePolicyMig(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) @@ -531,12 +533,12 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, } // 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) @@ -1564,45 +1566,14 @@ func insertChannelMig(ctx context.Context, db SQLQueries, // 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. -// -// TODO(elle): update this function to be more performant in the migration -// setting. For the sake of keeping the commit that introduced this function -// simple, this is for now mostly the same as updateChanEdgePolicy. func insertChanEdgePolicyMig(ctx context.Context, tx SQLQueries, - edge *models.ChannelEdgePolicy) (route.Vertex, route.Vertex, bool, - error) { - - var ( - node1Pub, node2Pub route.Vertex - isNode1 bool - chanIDB = channelIDToBytes(edge.ChannelID) - ) - - // Check that this edge policy refers to a channel that we already - // know of. We do this explicitly so that we can return the appropriate - // ErrEdgeNotFound error if the channel doesn't exist, rather than - // abort the transaction which would abort the entire batch. - dbChan, err := tx.GetChannelAndNodesBySCID( - ctx, sqlc.GetChannelAndNodesBySCIDParams{ - Scid: chanIDB, - Version: int16(ProtocolV1), - }, - ) - if errors.Is(err, sql.ErrNoRows) { - return node1Pub, node2Pub, false, ErrEdgeNotFound - } else if err != nil { - return node1Pub, node2Pub, false, fmt.Errorf("unable to "+ - "fetch channel(%v): %w", edge.ChannelID, err) - } - - copy(node1Pub[:], dbChan.Node1PubKey) - copy(node2Pub[:], dbChan.Node2PubKey) + dbChan *dbChanInfo, edge *models.ChannelEdgePolicy) error { // Figure out which node this edge is from. - isNode1 = edge.ChannelFlags&lnwire.ChanUpdateDirection == 0 - nodeID := dbChan.NodeID1 + isNode1 := edge.ChannelFlags&lnwire.ChanUpdateDirection == 0 + nodeID := dbChan.node1ID if !isNode1 { - nodeID = dbChan.NodeID2 + nodeID = dbChan.node2ID } var ( @@ -1616,7 +1587,7 @@ func insertChanEdgePolicyMig(ctx context.Context, tx SQLQueries, id, err := tx.InsertEdgePolicyMig(ctx, sqlc.InsertEdgePolicyMigParams{ Version: int16(ProtocolV1), - ChannelID: dbChan.ID, + ChannelID: dbChan.channelID, NodeID: nodeID, Timelock: int32(edge.TimeLockDelta), FeePpm: int64(edge.FeeProportionalMillionths), @@ -1638,24 +1609,32 @@ func insertChanEdgePolicyMig(ctx context.Context, tx SQLQueries, Signature: edge.SigBytes, }) if err != nil { - return node1Pub, node2Pub, isNode1, - fmt.Errorf("unable to upsert edge policy: %w", err) + 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 node1Pub, node2Pub, false, fmt.Errorf("unable to "+ - "marshal extra opaque data: %w", err) + return fmt.Errorf("unable to marshal extra opaque data: %w", + err) } - // Update the channel policy's extra signed fields. - err = upsertChanPolicyExtraSignedFields(ctx, tx, id, extra) - if err != nil { - return node1Pub, node2Pub, false, fmt.Errorf("inserting chan "+ - "policy extra TLVs: %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 node1Pub, node2Pub, isNode1, nil + return nil } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 60871e17b..1789c70c4 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -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 @@ -3934,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( - ctx, sqlc.InsertChanPolicyExtraTypeParams{ + err = db.UpsertChanPolicyExtraType( + ctx, sqlc.UpsertChanPolicyExtraTypeParams{ ChannelPolicyID: chanPolicyID, Type: int64(tlvType), Value: value, diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index bdc049d85..4c2be19ea 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -2304,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 @@ -3429,6 +3406,33 @@ 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 diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 6d38597a4..0087559be 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -87,7 +87,6 @@ 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 @@ -141,6 +140,7 @@ 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) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 7f12cd0c5..f2224c00f 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -869,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 From ce1df9da34293cc512b43dfbae91759c973e2768 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 15 Aug 2025 10:57:53 +0200 Subject: [PATCH 10/10] graph/db: let the rapid migration test also tests idempotency --- graph/db/sql_migration_test.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 3833b8cc6..e68c4431d 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -1273,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) { @@ -1423,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.