From 218aa9eaa810efe84ba661a4a1ca42c299fe8cef Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 12 Aug 2025 15:37:40 +0200 Subject: [PATCH 1/9] graph/db: move sanity check out of insertChannel We do this so that this lookup is only done in the situation it is actually needed. During a migration, we dont need to special case this AlreadyExists error since we will terminate the transaction either way. So there is no need for the extra lookup during the migration. A timing analysis showed that this query was significantly impacting the performance of the migration when run with a postgres backend. --- graph/db/sql_store.go | 42 +++++++++++++++++++----------------------- 1 file changed, 19 insertions(+), 23 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1c913b78c..40d5525d2 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -591,15 +591,29 @@ func (s *SQLStore) AddChannelEdge(ctx context.Context, alreadyExists = false }, Do: func(tx SQLQueries) error { - _, err := insertChannel(ctx, tx, edge) + chanIDB := channelIDToBytes(edge.ChannelID) - // Silence ErrEdgeAlreadyExist so that the batch can - // succeed, but propagate the error via local state. - if errors.Is(err, ErrEdgeAlreadyExist) { + // Make sure that the channel doesn't already exist. We + // do this explicitly instead of relying on catching a + // unique constraint error because relying on SQL to + // throw that error would abort the entire batch of + // transactions. + _, err := tx.GetChannelBySCID( + ctx, sqlc.GetChannelBySCIDParams{ + Scid: chanIDB, + Version: int16(ProtocolV1), + }, + ) + if err == nil { alreadyExists = true return nil + } else if !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("unable to fetch channel: %w", + err) } + _, err = insertChannel(ctx, tx, edge) + return err }, OnCommit: func(err error) error { @@ -3767,24 +3781,6 @@ type dbChanInfo struct { func insertChannel(ctx context.Context, db SQLQueries, edge *models.ChannelEdgeInfo) (*dbChanInfo, error) { - chanIDB := channelIDToBytes(edge.ChannelID) - - // Make sure that the channel doesn't already exist. We do this - // explicitly instead of relying on catching a unique constraint error - // because relying on SQL to throw that error would abort the entire - // batch of transactions. - _, err := db.GetChannelBySCID( - ctx, sqlc.GetChannelBySCIDParams{ - Scid: chanIDB, - Version: int16(ProtocolV1), - }, - ) - if err == nil { - return nil, ErrEdgeAlreadyExist - } else if !errors.Is(err, sql.ErrNoRows) { - return nil, fmt.Errorf("unable to fetch channel: %w", err) - } - // Make sure that at least a "shell" entry for each node is present in // the nodes table. node1DBID, err := maybeCreateShellNode(ctx, db, edge.NodeKey1Bytes) @@ -3804,7 +3800,7 @@ func insertChannel(ctx context.Context, db SQLQueries, createParams := sqlc.CreateChannelParams{ Version: int16(ProtocolV1), - Scid: chanIDB, + Scid: channelIDToBytes(edge.ChannelID), NodeID1: node1DBID, NodeID2: node2DBID, Outpoint: edge.ChannelPoint.String(), From 03ef2740a66034a7b58da8943aa1450e8d1f9eef Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 13:07:11 +0200 Subject: [PATCH 2/9] graph/db+sqldb: use batch validation for node migration Restructue the `migrateNodes` function so that it does the validation of migrated nodes in batches. So instead of fetching each node individually after migrating it, we wait for a minimum batch size to be reached and then validate a batch of nodes together. This lets us make way fewer DB round trips. --- graph/db/sql_migration.go | 169 +++++++++++++++++++++++++---------- graph/db/sql_store.go | 1 + sqldb/sqlc/graph.sql.go | 47 ++++++++++ sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 5 ++ 5 files changed, 176 insertions(+), 47 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 9b22fbf03..72947bd3f 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -107,8 +107,8 @@ func checkGraphExists(db kvdb.Backend) (bool, error) { } // migrateNodes migrates all nodes from the KV backend to the SQL database. -// This includes doing a sanity check after each migration to ensure that the -// migrated node matches the original node. +// It collects nodes in batches, inserts them individually, and then validates +// them in batches. func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, kvBackend kvdb.Backend, sqlDB SQLQueries) error { @@ -125,6 +125,108 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, } ) + // batch is a map that holds node objects that have been migrated to + // the native SQL store that have yet to be validated. The object's held + // by this map were derived from the KVDB store and so when they are + // validated, the map index (the SQL store node ID) will be used to + // fetch the corresponding node object in the SQL store, and it will + // then be compared against the original KVDB node object. + batch := make( + map[int64]*models.LightningNode, cfg.MaxBatchSize, + ) + + // validateBatch validates that the batch of nodes in the 'batch' map + // have been migrated successfully. + validateBatch := func() error { + if len(batch) == 0 { + return nil + } + + // Extract DB node IDs. + dbIDs := make([]int64, 0, len(batch)) + for dbID := range batch { + dbIDs = append(dbIDs, dbID) + } + + // Batch fetch all nodes from the database. + dbNodes, err := sqlDB.GetNodesByIDs(ctx, dbIDs) + if err != nil { + return fmt.Errorf("could not batch fetch nodes: %w", + err) + } + + // Make sure that the number of nodes fetched matches the number + // of nodes in the batch. + if len(dbNodes) != len(batch) { + return fmt.Errorf("expected to fetch %d nodes, "+ + "but got %d", len(batch), len(dbNodes)) + } + + // Now, batch fetch the normalised data for all the nodes in + // the batch. + batchData, err := batchLoadNodeData(ctx, cfg, sqlDB, dbIDs) + if err != nil { + return fmt.Errorf("unable to batch load node data: %w", + err) + } + + for _, dbNode := range dbNodes { + // Get the KVDB node info from the batch map. + node, ok := batch[dbNode.ID] + if !ok { + return fmt.Errorf("node with ID %d not found "+ + "in batch", dbNode.ID) + } + + // Build the migrated node from the DB node and the + // batch node data. + migNode, err := buildNodeWithBatchData( + dbNode, batchData, + ) + if err != nil { + return fmt.Errorf("could not build migrated "+ + "node from dbNode(db id: %d, node "+ + "pub: %x): %w", dbNode.ID, + node.PubKeyBytes, err) + } + + // Make sure that the node addresses are sorted before + // comparing them to ensure that the order of addresses + // does not affect the comparison. + slices.SortFunc( + node.Addresses, func(i, j net.Addr) int { + return cmp.Compare( + i.String(), j.String(), + ) + }, + ) + slices.SortFunc( + migNode.Addresses, func(i, j net.Addr) int { + return cmp.Compare( + i.String(), j.String(), + ) + }, + ) + + err = sqldb.CompareRecords( + node, migNode, + fmt.Sprintf("node %x", node.PubKeyBytes), + ) + if err != nil { + return fmt.Errorf("node mismatch after "+ + "migration for node %x: %w", + node.PubKeyBytes, err) + } + } + + // Clear the batch map for the next iteration. + batch = make( + map[int64]*models.LightningNode, cfg.MaxBatchSize, + ) + + return nil + } + // Loop through each node in the KV store and insert it into the SQL // database. err := forEachNode(kvBackend, func(_ kvdb.RTx, @@ -172,52 +274,16 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, err) } - // Fetch it from the SQL store and compare it against the - // original node object to ensure the migration was successful. - dbNode, err := sqlDB.GetNodeByPubKey( - ctx, sqlc.GetNodeByPubKeyParams{ - PubKey: node.PubKeyBytes[:], - Version: int16(ProtocolV1), - }, - ) - if err != nil { - return fmt.Errorf("could not get node by pubkey (%x)"+ - "after migration: %w", pub, err) - } + // Add to validation batch. + batch[id] = node - // Sanity check: ensure the migrated node ID matches the one we - // just inserted. - if dbNode.ID != id { - return fmt.Errorf("node ID mismatch for node (%x) "+ - "after migration: expected %d, got %d", - pub, id, dbNode.ID) - } - - migratedNode, err := buildNode(ctx, cfg, sqlDB, dbNode) - if err != nil { - return fmt.Errorf("could not build migrated node "+ - "from dbNode(db id: %d, node pub: %x): %w", - dbNode.ID, pub, err) - } - - // Make sure that the node addresses are sorted before - // comparing them to ensure that the order of addresses does - // not affect the comparison. - slices.SortFunc(node.Addresses, func(i, j net.Addr) int { - return cmp.Compare(i.String(), j.String()) - }) - slices.SortFunc( - migratedNode.Addresses, func(i, j net.Addr) int { - return cmp.Compare(i.String(), j.String()) - }, - ) - - err = sqldb.CompareRecords( - node, migratedNode, fmt.Sprintf("node %x", pub), - ) - if err != nil { - return fmt.Errorf("node mismatch after migration "+ - "for node %x: %w", pub, err) + // Validate batch when full. + if len(batch) >= cfg.MaxBatchSize { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch validation failed: %w", + err) + } } s.Do(func() { @@ -239,6 +305,15 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, return fmt.Errorf("could not migrate nodes: %w", err) } + // Validate any remaining nodes in the batch. + if len(batch) > 0 { + err := validateBatch() + if err != nil { + return fmt.Errorf("final batch validation failed: %w", + err) + } + } + log.Infof("Migrated %d nodes from KV to SQL (skipped %d nodes due to "+ "invalid TLV streams)", count, skipped) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 40d5525d2..de6c150b3 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -55,6 +55,7 @@ type SQLQueries interface { */ UpsertNode(ctx context.Context, arg sqlc.UpsertNodeParams) (int64, error) GetNodeByPubKey(ctx context.Context, arg sqlc.GetNodeByPubKeyParams) (sqlc.GraphNode, error) + GetNodesByIDs(ctx context.Context, ids []int64) ([]sqlc.GraphNode, error) GetNodeIDByPubKey(ctx context.Context, arg sqlc.GetNodeIDByPubKeyParams) (int64, error) GetNodesByLastUpdateRange(ctx context.Context, arg sqlc.GetNodesByLastUpdateRangeParams) ([]sqlc.GraphNode, error) ListNodesPaginated(ctx context.Context, arg sqlc.ListNodesPaginatedParams) ([]sqlc.GraphNode, error) diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 6db5d32c8..5a81e5217 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -1729,6 +1729,53 @@ func (q *Queries) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyPa return id, err } +const getNodesByIDs = `-- name: GetNodesByIDs :many +SELECT id, version, pub_key, alias, last_update, color, signature +FROM graph_nodes +WHERE id IN (/*SLICE:ids*/?) +` + +func (q *Queries) GetNodesByIDs(ctx context.Context, ids []int64) ([]GraphNode, error) { + query := getNodesByIDs + var queryParams []interface{} + if len(ids) > 0 { + for _, v := range ids { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:ids*/?", makeQueryParams(len(queryParams), len(ids)), 1) + } else { + query = strings.Replace(query, "/*SLICE:ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphNode + for rows.Next() { + var i GraphNode + if err := rows.Scan( + &i.ID, + &i.Version, + &i.PubKey, + &i.Alias, + &i.LastUpdate, + &i.Color, + &i.Signature, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getNodesByLastUpdateRange = `-- name: GetNodesByLastUpdateRange :many SELECT id, version, pub_key, alias, last_update, color, signature FROM graph_nodes diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 55fd74b75..7f1ad9abb 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -68,6 +68,7 @@ type Querier interface { GetNodeFeaturesBatch(ctx context.Context, ids []int64) ([]GraphNodeFeature, error) GetNodeFeaturesByPubKey(ctx context.Context, arg GetNodeFeaturesByPubKeyParams) ([]int32, error) GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) + GetNodesByIDs(ctx context.Context, ids []int64) ([]GraphNode, error) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error) GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error) GetPruneTip(ctx context.Context) (GraphPruneLog, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index ae276a5cb..55629a9ac 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -21,6 +21,11 @@ WHERE graph_nodes.last_update IS NULL OR EXCLUDED.last_update > graph_nodes.last_update RETURNING id; +-- name: GetNodesByIDs :many +SELECT * +FROM graph_nodes +WHERE id IN (sqlc.slice('ids')/*SLICE:ids*/); + -- name: GetNodeByPubKey :one SELECT * FROM graph_nodes From 81c54611c1d102d555c688e87d079997df5b7c58 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 13:37:25 +0200 Subject: [PATCH 3/9] graph/db+sqldb: use batch fetching during channel&policy migration Restructue the `migrateChannelsAndPolicies` function so that it does the validation of migrated channels and policies in batches. So instead of fetching channel and its policies individually after migrating it, we wait for a minimum batch size to be reached and then validate a batch of them together. This lets us make way fewer DB round trips. --- graph/db/sql_migration.go | 260 ++++++++++++++++++++++++----------- graph/db/sql_store.go | 46 +++++++ sqldb/sqlc/graph.sql.go | 173 +++++++++++++++++++++++ sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 53 +++++++ 5 files changed, 452 insertions(+), 81 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 72947bd3f..dcf664a1e 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -394,6 +394,24 @@ func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend, return nil } +// migChanInfo holds the information about a channel and its policies. +type migChanInfo struct { + // edge is the channel object as read from the KVDB source. + edge *models.ChannelEdgeInfo + + // policy1 is the first channel policy for the channel as read from + // the KVDB source. + policy1 *models.ChannelEdgePolicy + + // policy2 is the second channel policy for the channel as read + // from the KVDB source. + policy2 *models.ChannelEdgePolicy + + // dbInfo holds location info (in the form of DB IDs) of the channel + // and its policies in the native-SQL destination. + dbInfo *dbChanInfo +} + // migrateChannelsAndPolicies migrates all channels and their policies // from the KV backend to the SQL database. func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, @@ -434,6 +452,11 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, return nil } + // batch is used to collect migrated channel info that we will + // batch-validate. Each entry is indexed by the DB ID of the channel + // in the SQL database. + batch := make(map[int64]*migChanInfo, cfg.QueryCfg.MaxBatchSize) + // Iterate over each channel in the KV store and migrate it and its // policies to the SQL database. err := forEachChannel(kvBackend, func(channel *models.ChannelEdgeInfo, @@ -489,14 +512,47 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, channelCount++ chunk++ - err = migrateSingleChannel( - ctx, cfg, sqlDB, channel, policy1, policy2, - migChanPolicy, - ) + // Migrate the channel info along with its policies. + dbChanInfo, err := insertChannel(ctx, sqlDB, channel) if err != nil { - return fmt.Errorf("could not migrate channel %d: %w", + 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) + if err != nil { + return fmt.Errorf("could not migrate policy1(%d): %w", scid, err) } + err = migChanPolicy(policy2) + if err != nil { + return fmt.Errorf("could not migrate policy2(%d): %w", + scid, err) + } + + // Collect the migrated channel info and policies in a batch for + // later validation. + batch[dbChanInfo.channelID] = &migChanInfo{ + edge: channel, + policy1: policy1, + policy2: policy2, + dbInfo: dbChanInfo, + } + + if len(batch) >= cfg.QueryCfg.MaxBatchSize { + // Do batch validation. + err := validateMigratedChannels(ctx, cfg, sqlDB, batch) + if err != nil { + return fmt.Errorf("could not validate "+ + "channel batch: %w", err) + } + + batch = make( + map[int64]*migChanInfo, + cfg.QueryCfg.MaxBatchSize, + ) + } s.Do(func() { elapsed := time.Since(t0).Seconds() @@ -518,6 +574,17 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, err) } + if len(batch) > 0 { + // Do a final batch validation for any remaining channels. + err := validateMigratedChannels(ctx, cfg, sqlDB, batch) + if err != nil { + return fmt.Errorf("could not validate final channel "+ + "batch: %w", err) + } + + batch = make(map[int64]*migChanInfo, cfg.QueryCfg.MaxBatchSize) + } + log.Infof("Migrated %d channels and %d policies from KV to SQL "+ "(skipped %d channels and %d policies due to invalid TLV "+ "streams)", channelCount, policyCount, skippedChanCount, @@ -526,70 +593,142 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, return nil } -func migrateSingleChannel(ctx context.Context, cfg *SQLStoreConfig, - sqlDB SQLQueries, channel *models.ChannelEdgeInfo, - policy1, policy2 *models.ChannelEdgePolicy, - migChanPolicy func(*models.ChannelEdgePolicy) error) error { +// validateMigratedChannels validates the channels in the batch after they have +// been migrated to the SQL database. It batch fetches all channels by their IDs +// and compares the migrated channels and their policies with the original ones +// to ensure they match using batch construction patterns. +func validateMigratedChannels(ctx context.Context, cfg *SQLStoreConfig, + sqlDB SQLQueries, batch map[int64]*migChanInfo) error { - scid := channel.ChannelID - - // First, migrate the channel info along with its policies. - dbChanInfo, err := insertChannel(ctx, sqlDB, channel) - if err != nil { - return fmt.Errorf("could not insert record for channel %d "+ - "in SQL store: %w", scid, err) + // Convert batch keys (DB IDs) to an int slice for the batch query. + dbChanIDs := make([]int64, 0, len(batch)) + for id := range batch { + dbChanIDs = append(dbChanIDs, id) } - // Now, migrate the two channel policies. - err = migChanPolicy(policy1) + // Batch fetch all channels with their policies. + rows, err := sqlDB.GetChannelsByIDs(ctx, dbChanIDs) if err != nil { - return fmt.Errorf("could not migrate policy1(%d): %w", scid, - err) - } - err = migChanPolicy(policy2) - if err != nil { - return fmt.Errorf("could not migrate policy2(%d): %w", scid, + return fmt.Errorf("could not batch get channels by IDs: %w", err) } - // Now, fetch the channel and its policies from the SQL DB. - row, err := sqlDB.GetChannelBySCIDWithPolicies( - ctx, sqlc.GetChannelBySCIDWithPoliciesParams{ - Scid: channelIDToBytes(scid), - Version: int16(ProtocolV1), - }, + // Sanity check that the same number of channels were returned + // as requested. + if len(rows) != len(dbChanIDs) { + return fmt.Errorf("expected to fetch %d channels, "+ + "but got %d", len(dbChanIDs), len(rows)) + } + + // Collect all policy IDs needed for batch data loading. + dbPolicyIDs := make([]int64, 0, len(dbChanIDs)*2) + + for _, row := range rows { + scid := byteOrder.Uint64(row.GraphChannel.Scid) + + dbPol1, dbPol2, err := extractChannelPolicies(row) + if err != nil { + return fmt.Errorf("could not extract channel policies"+ + " for SCID %d: %w", scid, err) + } + if dbPol1 != nil { + dbPolicyIDs = append(dbPolicyIDs, dbPol1.ID) + } + if dbPol2 != nil { + dbPolicyIDs = append(dbPolicyIDs, dbPol2.ID) + } + } + + // Batch load all channel and policy data (features, extras). + batchData, err := batchLoadChannelData( + ctx, cfg.QueryCfg, sqlDB, dbChanIDs, dbPolicyIDs, ) if err != nil { - return fmt.Errorf("could not get channel by SCID(%d): %w", scid, - err) + return fmt.Errorf("could not batch load channel and policy "+ + "data: %w", err) } + // Validate each channel in the batch using pre-loaded data. + for _, row := range rows { + kvdbChan, ok := batch[row.GraphChannel.ID] + if !ok { + return fmt.Errorf("channel with ID %d not found "+ + "in batch", row.GraphChannel.ID) + } + + scid := byteOrder.Uint64(row.GraphChannel.Scid) + + err = validateMigratedChannelWithBatchData( + cfg, scid, kvdbChan, row, batchData, + ) + if err != nil { + return fmt.Errorf("channel %d validation failed "+ + "after migration: %w", scid, err) + } + } + + return nil +} + +// validateMigratedChannelWithBatchData validates a single migrated channel +// using pre-fetched batch data for optimal performance. +func validateMigratedChannelWithBatchData(cfg *SQLStoreConfig, + scid uint64, info *migChanInfo, row sqlc.GetChannelsByIDsRow, + batchData *batchChannelData) error { + + dbChanInfo := info.dbInfo + channel := info.edge + // Assert that the DB IDs for the channel and nodes are as expected // given the inserted channel info. - err = sqldb.CompareRecords( + err := sqldb.CompareRecords( dbChanInfo.channelID, row.GraphChannel.ID, "channel DB ID", ) if err != nil { return err } err = sqldb.CompareRecords( - dbChanInfo.node1ID, row.GraphNode.ID, "node1 DB ID", + dbChanInfo.node1ID, row.Node1ID, "node1 DB ID", ) if err != nil { return err } err = sqldb.CompareRecords( - dbChanInfo.node2ID, row.GraphNode_2.ID, "node2 DB ID", + dbChanInfo.node2ID, row.Node2ID, "node2 DB ID", ) if err != nil { return err } - migChan, migPol1, migPol2, err := getAndBuildChanAndPolicies( - ctx, cfg, sqlDB, row, + // Build node vertices from the row data. + node1, node2, err := buildNodeVertices( + row.Node1PubKey, row.Node2PubKey, ) if err != nil { - return fmt.Errorf("could not build migrated channel and "+ + return err + } + + // Build channel info using batch data. + migChan, err := buildEdgeInfoWithBatchData( + cfg.ChainHash, row.GraphChannel, node1, node2, batchData, + ) + if err != nil { + return fmt.Errorf("could not build migrated channel info: %w", + err) + } + + // Extract channel policies from the row. + dbPol1, dbPol2, err := extractChannelPolicies(row) + if err != nil { + return fmt.Errorf("could not extract channel policies: %w", err) + } + + // Build channel policies using batch data. + migPol1, migPol2, err := buildChanPoliciesWithBatchData( + dbPol1, dbPol2, scid, node1, node2, batchData, + ) + if err != nil { + return fmt.Errorf("could not build migrated channel "+ "policies: %w", err) } @@ -639,13 +778,13 @@ func migrateSingleChannel(ctx context.Context, cfg *SQLStoreConfig, ) } - err = checkPolicy(policy1, migPol1) + err = checkPolicy(info.policy1, migPol1) if err != nil { return fmt.Errorf("policy1 mismatch for channel %d: %w", scid, err) } - err = checkPolicy(policy2, migPol2) + err = checkPolicy(info.policy2, migPol2) if err != nil { return fmt.Errorf("policy2 mismatch for channel %d: %w", scid, err) @@ -778,47 +917,6 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, return nil } -// getAndBuildChanAndPolicies is a helper that builds the channel edge info -// and policies from the given row returned by the SQL query -// GetChannelBySCIDWithPolicies. -func getAndBuildChanAndPolicies(ctx context.Context, cfg *SQLStoreConfig, - db SQLQueries, - row sqlc.GetChannelBySCIDWithPoliciesRow) (*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { - - node1, node2, err := buildNodeVertices( - row.GraphNode.PubKey, row.GraphNode_2.PubKey, - ) - if err != nil { - return nil, nil, nil, err - } - - edge, err := getAndBuildEdgeInfo( - ctx, cfg, db, row.GraphChannel, node1, node2, - ) - if err != nil { - return nil, nil, nil, fmt.Errorf("unable to build channel "+ - "info: %w", err) - } - - dbPol1, dbPol2, err := extractChannelPolicies(row) - if err != nil { - return nil, nil, nil, fmt.Errorf("unable to extract channel "+ - "policies: %w", err) - } - - policy1, policy2, err := getAndBuildChanPolicies( - ctx, cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID, node1, - node2, - ) - if err != nil { - return nil, nil, nil, fmt.Errorf("unable to build channel "+ - "policies: %w", err) - } - - return edge, policy1, policy2, nil -} - // 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, diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index de6c150b3..c80d32a16 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -98,6 +98,7 @@ type SQLQueries interface { GetChannelsBySCIDRange(ctx context.Context, arg sqlc.GetChannelsBySCIDRangeParams) ([]sqlc.GetChannelsBySCIDRangeRow, error) GetChannelBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelBySCIDWithPoliciesParams) (sqlc.GetChannelBySCIDWithPoliciesRow, error) GetChannelsBySCIDWithPolicies(ctx context.Context, arg sqlc.GetChannelsBySCIDWithPoliciesParams) ([]sqlc.GetChannelsBySCIDWithPoliciesRow, error) + GetChannelsByIDs(ctx context.Context, ids []int64) ([]sqlc.GetChannelsByIDsRow, error) GetChannelAndNodesBySCID(ctx context.Context, arg sqlc.GetChannelAndNodesBySCIDParams) (sqlc.GetChannelAndNodesBySCIDRow, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) ListChannelsByNodeID(ctx context.Context, arg sqlc.ListChannelsByNodeIDParams) ([]sqlc.ListChannelsByNodeIDRow, error) @@ -4519,6 +4520,51 @@ func extractChannelPolicies(row any) (*sqlc.GraphChannelPolicy, } return policy1, policy2, nil + + case sqlc.GetChannelsByIDsRow: + if r.Policy1ID.Valid { + policy1 = &sqlc.GraphChannelPolicy{ + ID: r.Policy1ID.Int64, + Version: r.Policy1Version.Int16, + ChannelID: r.GraphChannel.ID, + NodeID: r.Policy1NodeID.Int64, + Timelock: r.Policy1Timelock.Int32, + FeePpm: r.Policy1FeePpm.Int64, + BaseFeeMsat: r.Policy1BaseFeeMsat.Int64, + MinHtlcMsat: r.Policy1MinHtlcMsat.Int64, + MaxHtlcMsat: r.Policy1MaxHtlcMsat, + LastUpdate: r.Policy1LastUpdate, + InboundBaseFeeMsat: r.Policy1InboundBaseFeeMsat, + InboundFeeRateMilliMsat: r.Policy1InboundFeeRateMilliMsat, + Disabled: r.Policy1Disabled, + MessageFlags: r.Policy1MessageFlags, + ChannelFlags: r.Policy1ChannelFlags, + Signature: r.Policy1Signature, + } + } + if r.Policy2ID.Valid { + policy2 = &sqlc.GraphChannelPolicy{ + ID: r.Policy2ID.Int64, + Version: r.Policy2Version.Int16, + ChannelID: r.GraphChannel.ID, + NodeID: r.Policy2NodeID.Int64, + Timelock: r.Policy2Timelock.Int32, + FeePpm: r.Policy2FeePpm.Int64, + BaseFeeMsat: r.Policy2BaseFeeMsat.Int64, + MinHtlcMsat: r.Policy2MinHtlcMsat.Int64, + MaxHtlcMsat: r.Policy2MaxHtlcMsat, + LastUpdate: r.Policy2LastUpdate, + InboundBaseFeeMsat: r.Policy2InboundBaseFeeMsat, + InboundFeeRateMilliMsat: r.Policy2InboundFeeRateMilliMsat, + Disabled: r.Policy2Disabled, + MessageFlags: r.Policy2MessageFlags, + ChannelFlags: r.Policy2ChannelFlags, + Signature: r.Policy2Signature, + } + } + + return policy1, policy2, nil + default: return nil, nil, fmt.Errorf("unexpected row type in "+ "extractChannelPolicies: %T", r) diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 5a81e5217..eb5eecf9c 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -873,6 +873,179 @@ func (q *Queries) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds return items, nil } +const getChannelsByIDs = `-- name: GetChannelsByIDs :many +SELECT + c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature, + + -- Minimal node data. + n1.id AS node1_id, + n1.pub_key AS node1_pub_key, + n2.id AS node2_id, + n2.pub_key AS node2_pub_key, + + -- Policy 1 + cp1.id AS policy1_id, + cp1.node_id AS policy1_node_id, + cp1.version AS policy1_version, + cp1.timelock AS policy1_timelock, + cp1.fee_ppm AS policy1_fee_ppm, + cp1.base_fee_msat AS policy1_base_fee_msat, + cp1.min_htlc_msat AS policy1_min_htlc_msat, + cp1.max_htlc_msat AS policy1_max_htlc_msat, + cp1.last_update AS policy1_last_update, + cp1.disabled AS policy1_disabled, + cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat, + cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat, + cp1.message_flags AS policy1_message_flags, + cp1.channel_flags AS policy1_channel_flags, + cp1.signature AS policy1_signature, + + -- Policy 2 + cp2.id AS policy2_id, + cp2.node_id AS policy2_node_id, + cp2.version AS policy2_version, + cp2.timelock AS policy2_timelock, + cp2.fee_ppm AS policy2_fee_ppm, + cp2.base_fee_msat AS policy2_base_fee_msat, + cp2.min_htlc_msat AS policy2_min_htlc_msat, + cp2.max_htlc_msat AS policy2_max_htlc_msat, + cp2.last_update AS policy2_last_update, + cp2.disabled AS policy2_disabled, + cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat, + cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat, + cp2.message_flags AS policy2_message_flags, + cp2.channel_flags AS policy2_channel_flags, + cp2.signature AS policy2_signature + +FROM graph_channels c + JOIN graph_nodes n1 ON c.node_id_1 = n1.id + JOIN graph_nodes n2 ON c.node_id_2 = n2.id + LEFT JOIN graph_channel_policies cp1 + ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version + LEFT JOIN graph_channel_policies cp2 + ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version +WHERE c.id IN (/*SLICE:ids*/?) +` + +type GetChannelsByIDsRow struct { + GraphChannel GraphChannel + Node1ID int64 + Node1PubKey []byte + Node2ID int64 + Node2PubKey []byte + Policy1ID sql.NullInt64 + Policy1NodeID sql.NullInt64 + Policy1Version sql.NullInt16 + Policy1Timelock sql.NullInt32 + Policy1FeePpm sql.NullInt64 + Policy1BaseFeeMsat sql.NullInt64 + Policy1MinHtlcMsat sql.NullInt64 + Policy1MaxHtlcMsat sql.NullInt64 + Policy1LastUpdate sql.NullInt64 + Policy1Disabled sql.NullBool + Policy1InboundBaseFeeMsat sql.NullInt64 + Policy1InboundFeeRateMilliMsat sql.NullInt64 + Policy1MessageFlags sql.NullInt16 + Policy1ChannelFlags sql.NullInt16 + Policy1Signature []byte + Policy2ID sql.NullInt64 + Policy2NodeID sql.NullInt64 + Policy2Version sql.NullInt16 + Policy2Timelock sql.NullInt32 + Policy2FeePpm sql.NullInt64 + Policy2BaseFeeMsat sql.NullInt64 + Policy2MinHtlcMsat sql.NullInt64 + Policy2MaxHtlcMsat sql.NullInt64 + Policy2LastUpdate sql.NullInt64 + Policy2Disabled sql.NullBool + Policy2InboundBaseFeeMsat sql.NullInt64 + Policy2InboundFeeRateMilliMsat sql.NullInt64 + Policy2MessageFlags sql.NullInt16 + Policy2ChannelFlags sql.NullInt16 + Policy2Signature []byte +} + +func (q *Queries) GetChannelsByIDs(ctx context.Context, ids []int64) ([]GetChannelsByIDsRow, error) { + query := getChannelsByIDs + var queryParams []interface{} + if len(ids) > 0 { + for _, v := range ids { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:ids*/?", makeQueryParams(len(queryParams), len(ids)), 1) + } else { + query = strings.Replace(query, "/*SLICE:ids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetChannelsByIDsRow + for rows.Next() { + var i GetChannelsByIDsRow + if err := rows.Scan( + &i.GraphChannel.ID, + &i.GraphChannel.Version, + &i.GraphChannel.Scid, + &i.GraphChannel.NodeID1, + &i.GraphChannel.NodeID2, + &i.GraphChannel.Outpoint, + &i.GraphChannel.Capacity, + &i.GraphChannel.BitcoinKey1, + &i.GraphChannel.BitcoinKey2, + &i.GraphChannel.Node1Signature, + &i.GraphChannel.Node2Signature, + &i.GraphChannel.Bitcoin1Signature, + &i.GraphChannel.Bitcoin2Signature, + &i.Node1ID, + &i.Node1PubKey, + &i.Node2ID, + &i.Node2PubKey, + &i.Policy1ID, + &i.Policy1NodeID, + &i.Policy1Version, + &i.Policy1Timelock, + &i.Policy1FeePpm, + &i.Policy1BaseFeeMsat, + &i.Policy1MinHtlcMsat, + &i.Policy1MaxHtlcMsat, + &i.Policy1LastUpdate, + &i.Policy1Disabled, + &i.Policy1InboundBaseFeeMsat, + &i.Policy1InboundFeeRateMilliMsat, + &i.Policy1MessageFlags, + &i.Policy1ChannelFlags, + &i.Policy1Signature, + &i.Policy2ID, + &i.Policy2NodeID, + &i.Policy2Version, + &i.Policy2Timelock, + &i.Policy2FeePpm, + &i.Policy2BaseFeeMsat, + &i.Policy2MinHtlcMsat, + &i.Policy2MaxHtlcMsat, + &i.Policy2LastUpdate, + &i.Policy2Disabled, + &i.Policy2InboundBaseFeeMsat, + &i.Policy2InboundFeeRateMilliMsat, + &i.Policy2MessageFlags, + &i.Policy2ChannelFlags, + &i.Policy2Signature, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getChannelsByOutpoints = `-- name: GetChannelsByOutpoints :many SELECT c.id, c.version, c.scid, c.node_id_1, c.node_id_2, c.outpoint, c.capacity, c.bitcoin_key_1, c.bitcoin_key_2, c.node_1_signature, c.node_2_signature, c.bitcoin_1_signature, c.bitcoin_2_signature, diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 7f1ad9abb..4e1e26a02 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -42,6 +42,7 @@ type Querier interface { GetChannelFeaturesBatch(ctx context.Context, chanIds []int64) ([]GraphChannelFeature, error) GetChannelPolicyByChannelAndNode(ctx context.Context, arg GetChannelPolicyByChannelAndNodeParams) (GraphChannelPolicy, error) GetChannelPolicyExtraTypesBatch(ctx context.Context, policyIds []int64) ([]GetChannelPolicyExtraTypesBatchRow, error) + GetChannelsByIDs(ctx context.Context, ids []int64) ([]GetChannelsByIDsRow, error) GetChannelsByOutpoints(ctx context.Context, outpoints []string) ([]GetChannelsByOutpointsRow, error) GetChannelsByPolicyLastUpdateRange(ctx context.Context, arg GetChannelsByPolicyLastUpdateRangeParams) ([]GetChannelsByPolicyLastUpdateRangeRow, error) GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index 55629a9ac..c0b890131 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -335,6 +335,59 @@ WHERE c.version = @version AND c.scid IN (sqlc.slice('scids')/*SLICE:scids*/); +-- name: GetChannelsByIDs :many +SELECT + sqlc.embed(c), + + -- Minimal node data. + n1.id AS node1_id, + n1.pub_key AS node1_pub_key, + n2.id AS node2_id, + n2.pub_key AS node2_pub_key, + + -- Policy 1 + cp1.id AS policy1_id, + cp1.node_id AS policy1_node_id, + cp1.version AS policy1_version, + cp1.timelock AS policy1_timelock, + cp1.fee_ppm AS policy1_fee_ppm, + cp1.base_fee_msat AS policy1_base_fee_msat, + cp1.min_htlc_msat AS policy1_min_htlc_msat, + cp1.max_htlc_msat AS policy1_max_htlc_msat, + cp1.last_update AS policy1_last_update, + cp1.disabled AS policy1_disabled, + cp1.inbound_base_fee_msat AS policy1_inbound_base_fee_msat, + cp1.inbound_fee_rate_milli_msat AS policy1_inbound_fee_rate_milli_msat, + cp1.message_flags AS policy1_message_flags, + cp1.channel_flags AS policy1_channel_flags, + cp1.signature AS policy1_signature, + + -- Policy 2 + cp2.id AS policy2_id, + cp2.node_id AS policy2_node_id, + cp2.version AS policy2_version, + cp2.timelock AS policy2_timelock, + cp2.fee_ppm AS policy2_fee_ppm, + cp2.base_fee_msat AS policy2_base_fee_msat, + cp2.min_htlc_msat AS policy2_min_htlc_msat, + cp2.max_htlc_msat AS policy2_max_htlc_msat, + cp2.last_update AS policy2_last_update, + cp2.disabled AS policy2_disabled, + cp2.inbound_base_fee_msat AS policy2_inbound_base_fee_msat, + cp2.inbound_fee_rate_milli_msat AS policy2_inbound_fee_rate_milli_msat, + cp2.message_flags AS policy2_message_flags, + cp2.channel_flags AS policy2_channel_flags, + cp2.signature AS policy2_signature + +FROM graph_channels c + JOIN graph_nodes n1 ON c.node_id_1 = n1.id + JOIN graph_nodes n2 ON c.node_id_2 = n2.id + LEFT JOIN graph_channel_policies cp1 + ON cp1.channel_id = c.id AND cp1.node_id = c.node_id_1 AND cp1.version = c.version + LEFT JOIN graph_channel_policies cp2 + ON cp2.channel_id = c.id AND cp2.node_id = c.node_id_2 AND cp2.version = c.version +WHERE c.id IN (sqlc.slice('ids')/*SLICE:ids*/); + -- name: GetChannelsByPolicyLastUpdateRange :many SELECT sqlc.embed(c), From 8554f17b3f8d6be205c5648b9832322ab372f221 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 13:50:20 +0200 Subject: [PATCH 4/9] graph/db+sqldb: validate prune log migration using batching As was done in the previous commits for nodes & channels, we update the migratePruneLog function here so that it validates migrated entries in batches rather than one-by-one. --- graph/db/sql_migration.go | 128 ++++++++++++++++++++++++----------- graph/db/sql_store.go | 1 + sqldb/sqlc/graph.sql.go | 40 +++++++++++ sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 6 ++ 5 files changed, 135 insertions(+), 41 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index dcf664a1e..d62fe74ac 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -61,7 +61,8 @@ func MigrateGraphToSQL(ctx context.Context, cfg *SQLStoreConfig, } // 4) Migrate the Prune log. - if err := migratePruneLog(ctx, kvBackend, sqlDB); err != nil { + err = migratePruneLog(ctx, cfg.QueryCfg, kvBackend, sqlDB) + if err != nil { return fmt.Errorf("could not migrate prune log: %w", err) } @@ -794,12 +795,11 @@ func validateMigratedChannelWithBatchData(cfg *SQLStoreConfig, } // migratePruneLog migrates the prune log from the KV backend to the SQL -// database. It iterates over each prune log entry in the KV store, inserts it -// into the SQL database, and then verifies that the entry was inserted -// correctly by fetching it back from the SQL database and comparing it to the -// original entry. -func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, - sqlDB SQLQueries) error { +// database. It collects entries in batches, inserts them individually, and then +// validates them in batches using GetPruneEntriesForHeights for better i +// performance. +func migratePruneLog(ctx context.Context, cfg *sqldb.QueryConfig, + kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( count uint64 @@ -813,55 +813,92 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, } ) - // migrateSinglePruneEntry is a helper function that inserts a single - // prune log entry into the SQL database and verifies that it was - // inserted correctly. - migrateSinglePruneEntry := func(height uint32, - hash *chainhash.Hash) error { + batch := make(map[uint32]chainhash.Hash, cfg.MaxBatchSize) - count++ - chunk++ - - // Keep track of the prune tip height and hash. - if height > pruneTipHeight { - pruneTipHeight = height - pruneTipHash = *hash + // validateBatch validates a batch of prune entries using batch query. + validateBatch := func() error { + if len(batch) == 0 { + return nil } - err := sqlDB.UpsertPruneLogEntry( - ctx, sqlc.UpsertPruneLogEntryParams{ - BlockHeight: int64(height), - BlockHash: hash[:], - }, - ) + // Extract heights for the batch query. + heights := make([]int64, 0, len(batch)) + for height := range batch { + heights = append(heights, int64(height)) + } + + // Batch fetch all entries from the database. + rows, err := sqlDB.GetPruneEntriesForHeights(ctx, heights) if err != nil { - return fmt.Errorf("unable to insert prune log "+ - "entry for height %d: %w", height, err) + return fmt.Errorf("could not batch get prune "+ + "entries: %w", err) } - // Now, check that the entry was inserted correctly. - migratedHash, err := sqlDB.GetPruneHashByHeight( - ctx, int64(height), - ) - if err != nil { - return fmt.Errorf("could not get prune hash "+ - "for height %d: %w", height, err) + if len(rows) != len(batch) { + return fmt.Errorf("expected to fetch %d prune "+ + "entries, but got %d", len(batch), + len(rows)) } - return sqldb.CompareRecords( - hash[:], migratedHash, "prune log entry", - ) + // Validate each entry in the batch. + for _, row := range rows { + kvdbHash, ok := batch[uint32(row.BlockHeight)] + if !ok { + return fmt.Errorf("prune entry for height %d "+ + "not found in batch", row.BlockHeight) + } + + err := sqldb.CompareRecords( + kvdbHash[:], row.BlockHash, + fmt.Sprintf("prune log entry at height %d", + row.BlockHash), + ) + if err != nil { + return err + } + } + + // Reset the batch map for the next iteration. + batch = make(map[uint32]chainhash.Hash, cfg.MaxBatchSize) + + return nil } // Iterate over each prune log entry in the KV store and migrate it to // the SQL database. err := forEachPruneLogEntry( kvBackend, func(height uint32, hash *chainhash.Hash) error { - err := migrateSinglePruneEntry(height, hash) + count++ + chunk++ + + // Keep track of the prune tip height and hash. + if height > pruneTipHeight { + pruneTipHeight = height + pruneTipHash = *hash + } + + // Insert the entry (individual inserts for now). + err := sqlDB.UpsertPruneLogEntry( + ctx, sqlc.UpsertPruneLogEntryParams{ + BlockHeight: int64(height), + BlockHash: hash[:], + }, + ) if err != nil { - return fmt.Errorf("could not migrate "+ - "prune log entry at height %d: %w", - height, err) + return fmt.Errorf("unable to insert prune log "+ + "entry for height %d: %w", height, err) + } + + // Add to validation batch. + batch[height] = *hash + + // Validate batch when full. + if len(batch) >= cfg.MaxBatchSize { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch "+ + "validation failed: %w", err) + } } s.Do(func() { @@ -882,6 +919,15 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, return fmt.Errorf("could not migrate prune log: %w", err) } + // Validate any remaining entries in the batch. + if len(batch) > 0 { + err := validateBatch() + if err != nil { + return fmt.Errorf("final batch validation failed: %w", + err) + } + } + // Check that the prune tip is set correctly in the SQL // database. pruneTip, err := sqlDB.GetPruneTip(ctx) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index c80d32a16..1de50165b 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -142,6 +142,7 @@ type SQLQueries interface { */ GetPruneTip(ctx context.Context) (sqlc.GraphPruneLog, error) GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error) + GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]sqlc.GraphPruneLog, error) UpsertPruneLogEntry(ctx context.Context, arg sqlc.UpsertPruneLogEntryParams) error DeletePruneLogEntriesInRange(ctx context.Context, arg sqlc.DeletePruneLogEntriesInRangeParams) error diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index eb5eecf9c..5718f25c1 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -1992,6 +1992,46 @@ func (q *Queries) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByL return items, nil } +const getPruneEntriesForHeights = `-- name: GetPruneEntriesForHeights :many +SELECT block_height, block_hash +FROM graph_prune_log +WHERE block_height + IN (/*SLICE:heights*/?) +` + +func (q *Queries) GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]GraphPruneLog, error) { + query := getPruneEntriesForHeights + var queryParams []interface{} + if len(heights) > 0 { + for _, v := range heights { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:heights*/?", makeQueryParams(len(queryParams), len(heights)), 1) + } else { + query = strings.Replace(query, "/*SLICE:heights*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphPruneLog + for rows.Next() { + var i GraphPruneLog + if err := rows.Scan(&i.BlockHeight, &i.BlockHash); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getPruneHashByHeight = `-- name: GetPruneHashByHeight :one SELECT block_hash FROM graph_prune_log diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index 4e1e26a02..e7b4f3f52 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -71,6 +71,7 @@ type Querier interface { GetNodeIDByPubKey(ctx context.Context, arg GetNodeIDByPubKeyParams) (int64, error) GetNodesByIDs(ctx context.Context, ids []int64) ([]GraphNode, error) GetNodesByLastUpdateRange(ctx context.Context, arg GetNodesByLastUpdateRangeParams) ([]GraphNode, error) + GetPruneEntriesForHeights(ctx context.Context, heights []int64) ([]GraphPruneLog, error) GetPruneHashByHeight(ctx context.Context, blockHeight int64) ([]byte, error) GetPruneTip(ctx context.Context) (GraphPruneLog, error) GetPublicV1ChannelsBySCID(ctx context.Context, arg GetPublicV1ChannelsBySCIDParams) ([]GraphChannel, error) diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index c0b890131..ed0eb8c50 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -952,6 +952,12 @@ FROM graph_prune_log ORDER BY block_height DESC LIMIT 1; +-- name: GetPruneEntriesForHeights :many +SELECT block_height, block_hash +FROM graph_prune_log +WHERE block_height + IN (sqlc.slice('heights')/*SLICE:heights*/); + -- name: GetPruneHashByHeight :one SELECT block_hash FROM graph_prune_log From a490e03479009789f3e39191358a8dd6cb4a2f31 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 13:56:50 +0200 Subject: [PATCH 5/9] graph/db+sqldb: use batch validation for closed SCID migration As was done in the previous commits for nodes & channels, we update the migrateClosedSCIDIndex function here so that it validates migrated entries in batches rather than one-by-one. --- graph/db/sql_migration.go | 78 ++++++++++++++++++++++++++++-------- graph/db/sql_store.go | 1 + sqldb/sqlc/graph.sql.go | 39 ++++++++++++++++++ sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 5 +++ 5 files changed, 107 insertions(+), 17 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index d62fe74ac..467f5436e 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -67,7 +67,7 @@ func MigrateGraphToSQL(ctx context.Context, cfg *SQLStoreConfig, } // 5) Migrate the closed SCID index. - err = migrateClosedSCIDIndex(ctx, kvBackend, sqlDB) + err = migrateClosedSCIDIndex(ctx, cfg.QueryCfg, kvBackend, sqlDB) if err != nil { return fmt.Errorf("could not migrate closed SCID index: %w", err) @@ -992,12 +992,11 @@ func forEachPruneLogEntry(db kvdb.Backend, cb func(height uint32, } // migrateClosedSCIDIndex migrates the closed SCID index from the KV backend to -// the SQL database. It iterates over each closed SCID in the KV store, inserts -// it into the SQL database, and then verifies that the SCID was inserted -// correctly by checking if the channel with the given SCID is seen as closed in -// the SQL database. -func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend, - sqlDB SQLQueries) error { +// the SQL database. It collects SCIDs in batches, inserts them individually, +// and then validates them in batches using GetClosedChannelsSCIDs for better +// performance. +func migrateClosedSCIDIndex(ctx context.Context, cfg *sqldb.QueryConfig, + kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( count uint64 @@ -1008,6 +1007,43 @@ func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend, Interval: 10 * time.Second, } ) + + batch := make([][]byte, 0, cfg.MaxBatchSize) + + // validateBatch validates a batch of closed SCIDs using batch query. + validateBatch := func() error { + if len(batch) == 0 { + return nil + } + + // Batch fetch all closed SCIDs from the database. + dbSCIDs, err := sqlDB.GetClosedChannelsSCIDs(ctx, batch) + if err != nil { + return fmt.Errorf("could not batch get closed "+ + "SCIDs: %w", err) + } + + // Create set of SCIDs that exist in the database for quick + // lookup. + dbSCIDSet := make(map[string]struct{}) + for _, scid := range dbSCIDs { + dbSCIDSet[string(scid)] = struct{}{} + } + + // Validate each SCID in the batch. + for _, expectedSCID := range batch { + if _, found := dbSCIDSet[string(expectedSCID)]; !found { + return fmt.Errorf("closed SCID %x not found "+ + "in database", expectedSCID) + } + } + + // Reset the batch for the next iteration. + batch = make([][]byte, 0, cfg.MaxBatchSize) + + return nil + } + migrateSingleClosedSCID := func(scid lnwire.ShortChannelID) error { count++ chunk++ @@ -1019,17 +1055,16 @@ func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend, "with SCID %s: %w", scid, err) } - // Now, verify that the channel with the given SCID is - // seen as closed. - isClosed, err := sqlDB.IsClosedChannel(ctx, chanIDB) - if err != nil { - return fmt.Errorf("could not check if channel %s "+ - "is closed: %w", scid, err) - } + // Add to validation batch. + batch = append(batch, chanIDB) - if !isClosed { - return fmt.Errorf("channel %s should be closed, "+ - "but is not", scid) + // Validate batch when full. + if len(batch) >= cfg.MaxBatchSize { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch validation failed: %w", + err) + } } s.Do(func() { @@ -1051,6 +1086,15 @@ func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend, err) } + // Validate any remaining SCIDs in the batch. + if len(batch) > 0 { + err := validateBatch() + if err != nil { + return fmt.Errorf("final batch validation failed: %w", + err) + } + } + log.Infof("Migrated %d closed SCIDs from KV to SQL", count) return nil diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1de50165b..fbfee848f 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -151,6 +151,7 @@ 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) } // 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 5718f25c1..7b9e472a1 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -1602,6 +1602,45 @@ func (q *Queries) GetChannelsBySCIDs(ctx context.Context, arg GetChannelsBySCIDs return items, nil } +const getClosedChannelsSCIDs = `-- name: GetClosedChannelsSCIDs :many +SELECT scid +FROM graph_closed_scids +WHERE scid IN (/*SLICE:scids*/?) +` + +func (q *Queries) GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error) { + query := getClosedChannelsSCIDs + var queryParams []interface{} + if len(scids) > 0 { + for _, v := range scids { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:scids*/?", makeQueryParams(len(queryParams), len(scids)), 1) + } else { + query = strings.Replace(query, "/*SLICE:scids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items [][]byte + for rows.Next() { + var scid []byte + if err := rows.Scan(&scid); err != nil { + return nil, err + } + items = append(items, scid) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getExtraNodeTypes = `-- name: GetExtraNodeTypes :many SELECT node_id, type, value FROM graph_node_extra_types diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index e7b4f3f52..b3c0857eb 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -48,6 +48,7 @@ type Querier interface { GetChannelsBySCIDRange(ctx context.Context, arg GetChannelsBySCIDRangeParams) ([]GetChannelsBySCIDRangeRow, error) GetChannelsBySCIDWithPolicies(ctx context.Context, arg GetChannelsBySCIDWithPoliciesParams) ([]GetChannelsBySCIDWithPoliciesRow, error) GetChannelsBySCIDs(ctx context.Context, arg GetChannelsBySCIDsParams) ([]GraphChannel, error) + GetClosedChannelsSCIDs(ctx context.Context, scids [][]byte) ([][]byte, error) GetDatabaseVersion(ctx context.Context) (int32, error) GetExtraNodeTypes(ctx context.Context, nodeID int64) ([]GraphNodeExtraType, error) // This method may return more than one invoice if filter using multiple fields diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index ed0eb8c50..f77e5044c 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -984,3 +984,8 @@ SELECT EXISTS ( FROM graph_closed_scids WHERE scid = $1 ); + +-- name: GetClosedChannelsSCIDs :many +SELECT scid +FROM graph_closed_scids +WHERE scid IN (sqlc.slice('scids')/*SLICE:scids*/); From 5b064747447c1108b6b308ef213d6109dc4965a8 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 14:11:40 +0200 Subject: [PATCH 6/9] graph/db+sqldb: batch validation for zombie index migration Finally, we update the migrateZombieIndex function to use batch validation just like was done in the previous commits. Here, we additionally make sure to validate the entire zombie index entry and not just the SCID. --- graph/db/sql_migration.go | 122 +++++++++++++++++++++++++++++------ graph/db/sql_store.go | 1 + sqldb/sqlc/graph.sql.go | 51 +++++++++++++++ sqldb/sqlc/querier.go | 1 + sqldb/sqlc/queries/graph.sql | 6 ++ 5 files changed, 160 insertions(+), 21 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 467f5436e..17a751f87 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -15,6 +15,7 @@ import ( "github.com/lightningnetwork/lnd/graph/db/models" "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lnwire" + "github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/sqldb" "github.com/lightningnetwork/lnd/sqldb/sqlc" "golang.org/x/time/rate" @@ -74,7 +75,8 @@ func MigrateGraphToSQL(ctx context.Context, cfg *SQLStoreConfig, } // 6) Migrate the zombie index. - if err := migrateZombieIndex(ctx, kvBackend, sqlDB); err != nil { + err = migrateZombieIndex(ctx, cfg.QueryCfg, kvBackend, sqlDB) + if err != nil { return fmt.Errorf("could not migrate zombie index: %w", err) } @@ -1100,18 +1102,17 @@ func migrateClosedSCIDIndex(ctx context.Context, cfg *sqldb.QueryConfig, return nil } -// migrateZombieIndex migrates the zombie index from the KV backend to -// the SQL database. It iterates over each zombie channel in the KV store, -// inserts it into the SQL database, and then verifies that the channel is -// indeed marked as a zombie channel in the SQL database. +// migrateZombieIndex migrates the zombie index from the KV backend to the SQL +// database. It collects zombie channels in batches, inserts them individually, +// and validates them in batches. // // NOTE: before inserting an entry into the zombie index, the function checks // if the channel is already marked as closed in the SQL store. If it is, // the entry is skipped. This means that the resulting zombie index count in // the SQL store may well be less than the count of zombie channels in the KV // store. -func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, - sqlDB SQLQueries) error { +func migrateZombieIndex(ctx context.Context, cfg *sqldb.QueryConfig, + kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( count uint64 @@ -1122,6 +1123,79 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, Interval: 10 * time.Second, } ) + + type zombieEntry struct { + pub1 route.Vertex + pub2 route.Vertex + } + + batch := make(map[uint64]*zombieEntry, cfg.MaxBatchSize) + + // validateBatch validates a batch of zombie SCIDs using batch query. + validateBatch := func() error { + if len(batch) == 0 { + return nil + } + + scids := make([][]byte, 0, len(batch)) + for scid := range batch { + scids = append(scids, channelIDToBytes(scid)) + } + + // Batch fetch all zombie channels from the database. + rows, err := sqlDB.GetZombieChannelsSCIDs( + ctx, sqlc.GetZombieChannelsSCIDsParams{ + Version: int16(ProtocolV1), + Scids: scids, + }, + ) + if err != nil { + return fmt.Errorf("could not batch get zombie "+ + "SCIDs: %w", err) + } + + // Make sure that the number of rows returned matches + // the number of SCIDs we requested. + if len(rows) != len(scids) { + return fmt.Errorf("expected to fetch %d zombie "+ + "SCIDs, but got %d", len(scids), len(rows)) + } + + // Validate each row is in the batch. + for _, row := range rows { + scid := byteOrder.Uint64(row.Scid) + + kvdbZombie, ok := batch[scid] + if !ok { + return fmt.Errorf("zombie SCID %x not found "+ + "in batch", scid) + } + + err = sqldb.CompareRecords( + kvdbZombie.pub1[:], row.NodeKey1, + fmt.Sprintf("zombie pub key 1 (%s) for "+ + "channel %d", kvdbZombie.pub1, scid), + ) + if err != nil { + return err + } + + err = sqldb.CompareRecords( + kvdbZombie.pub2[:], row.NodeKey2, + fmt.Sprintf("zombie pub key 2 (%s) for "+ + "channel %d", kvdbZombie.pub2, scid), + ) + if err != nil { + return err + } + } + + // Reset the batch for the next iteration. + batch = make(map[uint64]*zombieEntry, cfg.MaxBatchSize) + + return nil + } + err := forEachZombieEntry(kvBackend, func(chanID uint64, pubKey1, pubKey2 [33]byte) error { @@ -1158,22 +1232,19 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, "channel %d: %w", chanID, err) } - // Finally, verify that the channel is indeed marked as a - // zombie channel. - isZombie, err := sqlDB.IsZombieChannel( - ctx, sqlc.IsZombieChannelParams{ - Version: int16(ProtocolV1), - Scid: chanIDB, - }, - ) - if err != nil { - return fmt.Errorf("could not check if "+ - "channel %d is zombie: %w", chanID, err) + // Add to validation batch only after successful insertion. + batch[chanID] = &zombieEntry{ + pub1: pubKey1, + pub2: pubKey2, } - if !isZombie { - return fmt.Errorf("channel %d should be "+ - "a zombie, but is not", chanID) + // Validate batch when full. + if len(batch) >= cfg.MaxBatchSize { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch validation failed: %w", + err) + } } s.Do(func() { @@ -1192,6 +1263,15 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, return fmt.Errorf("could not migrate zombie index: %w", err) } + // Validate any remaining zombie SCIDs in the batch. + if len(batch) > 0 { + err := validateBatch() + if err != nil { + return fmt.Errorf("final batch validation failed: %w", + err) + } + } + log.Infof("Migrated %d zombie channels from KV to SQL", count) return nil diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index fbfee848f..956b8406d 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -133,6 +133,7 @@ type SQLQueries interface { */ UpsertZombieChannel(ctx context.Context, arg sqlc.UpsertZombieChannelParams) error GetZombieChannel(ctx context.Context, arg sqlc.GetZombieChannelParams) (sqlc.GraphZombieChannel, error) + GetZombieChannelsSCIDs(ctx context.Context, arg sqlc.GetZombieChannelsSCIDsParams) ([]sqlc.GraphZombieChannel, error) CountZombieChannels(ctx context.Context, version int16) (int64, error) DeleteZombieChannel(ctx context.Context, arg sqlc.DeleteZombieChannelParams) (sql.Result, error) IsZombieChannel(ctx context.Context, arg sqlc.IsZombieChannelParams) (bool, error) diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 7b9e472a1..d27143ca8 100644 --- a/sqldb/sqlc/graph.sql.go +++ b/sqldb/sqlc/graph.sql.go @@ -2261,6 +2261,57 @@ func (q *Queries) GetZombieChannel(ctx context.Context, arg GetZombieChannelPara return i, err } +const getZombieChannelsSCIDs = `-- name: GetZombieChannelsSCIDs :many +SELECT scid, version, node_key_1, node_key_2 +FROM graph_zombie_channels +WHERE version = $1 + AND scid IN (/*SLICE:scids*/?) +` + +type GetZombieChannelsSCIDsParams struct { + Version int16 + Scids [][]byte +} + +func (q *Queries) GetZombieChannelsSCIDs(ctx context.Context, arg GetZombieChannelsSCIDsParams) ([]GraphZombieChannel, error) { + query := getZombieChannelsSCIDs + var queryParams []interface{} + queryParams = append(queryParams, arg.Version) + if len(arg.Scids) > 0 { + for _, v := range arg.Scids { + queryParams = append(queryParams, v) + } + query = strings.Replace(query, "/*SLICE:scids*/?", makeQueryParams(len(queryParams), len(arg.Scids)), 1) + } else { + query = strings.Replace(query, "/*SLICE:scids*/?", "NULL", 1) + } + rows, err := q.db.QueryContext(ctx, query, queryParams...) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GraphZombieChannel + for rows.Next() { + var i GraphZombieChannel + if err := rows.Scan( + &i.Scid, + &i.Version, + &i.NodeKey1, + &i.NodeKey2, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const highestSCID = `-- name: HighestSCID :one SELECT scid FROM graph_channels diff --git a/sqldb/sqlc/querier.go b/sqldb/sqlc/querier.go index b3c0857eb..eaef3c777 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -84,6 +84,7 @@ type Querier interface { // and so the query for V2 may differ. GetV1DisabledSCIDs(ctx context.Context) ([][]byte, error) GetZombieChannel(ctx context.Context, arg GetZombieChannelParams) (GraphZombieChannel, error) + GetZombieChannelsSCIDs(ctx context.Context, arg GetZombieChannelsSCIDsParams) ([]GraphZombieChannel, error) HighestSCID(ctx context.Context, version int16) ([]byte, error) InsertAMPSubInvoice(ctx context.Context, arg InsertAMPSubInvoiceParams) error InsertAMPSubInvoiceHTLC(ctx context.Context, arg InsertAMPSubInvoiceHTLCParams) error diff --git a/sqldb/sqlc/queries/graph.sql b/sqldb/sqlc/queries/graph.sql index f77e5044c..04f814083 100644 --- a/sqldb/sqlc/queries/graph.sql +++ b/sqldb/sqlc/queries/graph.sql @@ -908,6 +908,12 @@ DO UPDATE SET node_key_1 = COALESCE(EXCLUDED.node_key_1, graph_zombie_channels.node_key_1), node_key_2 = COALESCE(EXCLUDED.node_key_2, graph_zombie_channels.node_key_2); +-- name: GetZombieChannelsSCIDs :many +SELECT * +FROM graph_zombie_channels +WHERE version = @version + AND scid IN (sqlc.slice('scids')/*SLICE:scids*/); + -- name: DeleteZombieChannel :execresult DELETE FROM graph_zombie_channels WHERE scid = $1 From b1c643f4f159195b24e265b41c88c037b8510287 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 14:15:59 +0200 Subject: [PATCH 7/9] graph/db: add migration timing logs Time the full duration of each graph migration step for the purposes of logging. --- graph/db/sql_migration.go | 33 +++++++++++++++++++++++---------- 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 17a751f87..0584914c3 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -118,6 +118,8 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, // Keep track of the number of nodes migrated and the number of // nodes skipped due to errors. var ( + totalTime = time.Now() + count uint64 skipped uint64 @@ -317,8 +319,9 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, } } - log.Infof("Migrated %d nodes from KV to SQL (skipped %d nodes due to "+ - "invalid TLV streams)", count, skipped) + log.Infof("Migrated %d nodes from KV to SQL in %v (skipped %d nodes "+ + "due to invalid TLV streams)", count, time.Since(totalTime), + skipped) return nil } @@ -421,6 +424,8 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( + totalTime = time.Now() + channelCount uint64 skippedChanCount uint64 policyCount uint64 @@ -588,10 +593,10 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, batch = make(map[int64]*migChanInfo, cfg.QueryCfg.MaxBatchSize) } - log.Infof("Migrated %d channels and %d policies from KV to SQL "+ + log.Infof("Migrated %d channels and %d policies from KV to SQL in %s"+ "(skipped %d channels and %d policies due to invalid TLV "+ - "streams)", channelCount, policyCount, skippedChanCount, - skippedPolicyCount) + "streams)", channelCount, policyCount, time.Since(totalTime), + skippedChanCount, skippedPolicyCount) return nil } @@ -804,6 +809,8 @@ func migratePruneLog(ctx context.Context, cfg *sqldb.QueryConfig, kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( + totalTime = time.Now() + count uint64 pruneTipHeight uint32 pruneTipHash chainhash.Hash @@ -958,9 +965,9 @@ func migratePruneLog(ctx context.Context, cfg *sqldb.QueryConfig, chainhash.Hash(pruneTip.BlockHash)) } - log.Infof("Migrated %d prune log entries from KV to SQL. The prune "+ - "tip is: height %d, hash: %s", count, pruneTipHeight, - pruneTipHash) + log.Infof("Migrated %d prune log entries from KV to SQL in %s. "+ + "The prune tip is: height %d, hash: %s", count, + time.Since(totalTime), pruneTipHeight, pruneTipHash) return nil } @@ -1001,6 +1008,8 @@ func migrateClosedSCIDIndex(ctx context.Context, cfg *sqldb.QueryConfig, kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( + totalTime = time.Now() + count uint64 t0 = time.Now() @@ -1097,7 +1106,8 @@ func migrateClosedSCIDIndex(ctx context.Context, cfg *sqldb.QueryConfig, } } - log.Infof("Migrated %d closed SCIDs from KV to SQL", count) + log.Infof("Migrated %d closed SCIDs from KV to SQL in %s", count, + time.Since(totalTime)) return nil } @@ -1115,6 +1125,8 @@ func migrateZombieIndex(ctx context.Context, cfg *sqldb.QueryConfig, kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( + totalTime = time.Now() + count uint64 t0 = time.Now() @@ -1272,7 +1284,8 @@ func migrateZombieIndex(ctx context.Context, cfg *sqldb.QueryConfig, } } - log.Infof("Migrated %d zombie channels from KV to SQL", count) + log.Infof("Migrated %d zombie channels from KV to SQL in %s", count, + time.Since(totalTime)) return nil } From d5729845d00baa277c7c2b6b2122cf6e947d1770 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 14:16:13 +0200 Subject: [PATCH 8/9] graph/db: remove outdated TODO --- graph/db/benchmark_test.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 74cf68c84..58b8cb39a 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -431,9 +431,6 @@ func TestPopulateDBs(t *testing.T) { // postgres backend instead of the kvdb-sqlite backend. // // NOTE: this is a helper test and is not run by default. -// -// TODO(elle): this test reveals tht there may be an issue with the postgres -// migration as it is super slow. func TestPopulateViaMigration(t *testing.T) { t.Skipf("Skipping local helper test") From f560c4d95b0127f569813b91fca949dae8461d94 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Thu, 14 Aug 2025 08:03:28 +0200 Subject: [PATCH 9/9] sqldb: use uint32 for config values --- graph/db/benchmark_test.go | 10 +++++----- graph/db/sql_migration.go | 10 +++++----- sqldb/paginate.go | 12 ++++++------ sqldb/paginate_test.go | 4 ++-- 4 files changed, 18 insertions(+), 18 deletions(-) diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 58b8cb39a..bdb7954cd 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -788,13 +788,13 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) { // Set the various page sizes we want to test. // // NOTE: these are the sqlite paging testing values. - testSizes := []int{20, 50, 100, 150, 500} + testSizes := []uint32{20, 50, 100, 150, 500} configOption := "MaxPageSize" if testBatching { configOption = "MaxBatchSize" - testSizes = []int{ + testSizes = []uint32{ 50, 100, 150, 200, 250, 300, 350, } } @@ -806,10 +806,10 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) { // Set the various page sizes we want to test. // // NOTE: these are the postgres paging values. - testSizes = []int{5000, 7000, 10000, 12000} + testSizes = []uint32{5000, 7000, 10000, 12000} if testBatching { - testSizes = []int{ + testSizes = []uint32{ 1000, 2000, 5000, 7000, 10000, } } @@ -828,7 +828,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) { if testBatching { cfg.MaxBatchSize = size } else { - cfg.MaxPageSize = int32(size) + cfg.MaxPageSize = size } store := connectNativeSQLite( diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 0584914c3..5aea3d335 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -283,7 +283,7 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, batch[id] = node // Validate batch when full. - if len(batch) >= cfg.MaxBatchSize { + if len(batch) >= int(cfg.MaxBatchSize) { err := validateBatch() if err != nil { return fmt.Errorf("batch validation failed: %w", @@ -548,7 +548,7 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, dbInfo: dbChanInfo, } - if len(batch) >= cfg.QueryCfg.MaxBatchSize { + if len(batch) >= int(cfg.QueryCfg.MaxBatchSize) { // Do batch validation. err := validateMigratedChannels(ctx, cfg, sqlDB, batch) if err != nil { @@ -902,7 +902,7 @@ func migratePruneLog(ctx context.Context, cfg *sqldb.QueryConfig, batch[height] = *hash // Validate batch when full. - if len(batch) >= cfg.MaxBatchSize { + if len(batch) >= int(cfg.MaxBatchSize) { err := validateBatch() if err != nil { return fmt.Errorf("batch "+ @@ -1070,7 +1070,7 @@ func migrateClosedSCIDIndex(ctx context.Context, cfg *sqldb.QueryConfig, batch = append(batch, chanIDB) // Validate batch when full. - if len(batch) >= cfg.MaxBatchSize { + if len(batch) >= int(cfg.MaxBatchSize) { err := validateBatch() if err != nil { return fmt.Errorf("batch validation failed: %w", @@ -1251,7 +1251,7 @@ func migrateZombieIndex(ctx context.Context, cfg *sqldb.QueryConfig, } // Validate batch when full. - if len(batch) >= cfg.MaxBatchSize { + if len(batch) >= int(cfg.MaxBatchSize) { err := validateBatch() if err != nil { return fmt.Errorf("batch validation failed: %w", diff --git a/sqldb/paginate.go b/sqldb/paginate.go index 17e2fd40e..4fd2a9d4d 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -37,11 +37,11 @@ const ( type QueryConfig struct { // MaxBatchSize is the maximum number of items included in a batch // query IN clauses list. - MaxBatchSize int `long:"max-batch-size" description:"The maximum number of items to include in a batch query IN clause. This is used for queries that fetch results based on a list of identifiers."` + MaxBatchSize uint32 `long:"max-batch-size" description:"The maximum number of items to include in a batch query IN clause. This is used for queries that fetch results based on a list of identifiers."` // MaxPageSize is the maximum number of items returned in a single page // of results. This is used for paginated queries. - MaxPageSize int32 `long:"max-page-size" description:"The maximum number of items to return in a single page of results. This is used for paginated queries."` + MaxPageSize uint32 `long:"max-page-size" description:"The maximum number of items to return in a single page of results. This is used for paginated queries."` } // Validate checks that the QueryConfig values are valid. @@ -121,9 +121,9 @@ func ExecuteBatchQuery[I any, T any, R any](ctx context.Context, } // Process items in pages. - for i := 0; i < len(inputItems); i += cfg.MaxBatchSize { + for i := 0; i < len(inputItems); i += int(cfg.MaxBatchSize) { // Calculate the end index for this page. - end := i + cfg.MaxBatchSize + end := i + int(cfg.MaxBatchSize) if end > len(inputItems) { end = len(inputItems) } @@ -189,7 +189,7 @@ func ExecutePaginatedQuery[C any, T any](ctx context.Context, cfg *QueryConfig, for { // Fetch the next page. - items, err := queryFunc(ctx, cursor, cfg.MaxPageSize) + items, err := queryFunc(ctx, cursor, int32(cfg.MaxPageSize)) if err != nil { return fmt.Errorf("failed to fetch page with "+ "cursor %v: %w", cursor, err) @@ -265,7 +265,7 @@ func ExecuteCollectAndBatchWithSharedDataQuery[C any, T any, I any, D any]( for { // Step 1: Fetch the next page of items. - items, err := pageQueryFunc(ctx, cursor, cfg.MaxPageSize) + items, err := pageQueryFunc(ctx, cursor, int32(cfg.MaxPageSize)) if err != nil { return fmt.Errorf("failed to fetch page with "+ "cursor %v: %w", cursor, err) diff --git a/sqldb/paginate_test.go b/sqldb/paginate_test.go index f62bf65b4..20abf422e 100644 --- a/sqldb/paginate_test.go +++ b/sqldb/paginate_test.go @@ -341,7 +341,7 @@ func TestExecutePaginatedQuery(t *testing.T) { tests := []struct { name string - pageSize int32 + pageSize uint32 allItems []testItem initialCursor int64 queryError error @@ -592,7 +592,7 @@ func TestExecuteCollectAndBatchWithSharedDataQuery(t *testing.T) { tests := []struct { name string - maxPageSize int32 + maxPageSize uint32 allRows []channelRow initialCursor int64 pageQueryError error