diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 74cf68c84..bdb7954cd 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") @@ -791,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, } } @@ -809,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, } } @@ -831,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 9b22fbf03..5aea3d335 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" @@ -61,19 +62,21 @@ 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) } // 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) } // 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) } @@ -107,14 +110,16 @@ 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 { // 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 @@ -125,6 +130,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 +279,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) >= int(cfg.MaxBatchSize) { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch validation failed: %w", + err) + } } s.Do(func() { @@ -239,8 +310,18 @@ func migrateNodes(ctx context.Context, cfg *sqldb.QueryConfig, return fmt.Errorf("could not migrate nodes: %w", err) } - log.Infof("Migrated %d nodes from KV to SQL (skipped %d nodes due to "+ - "invalid TLV streams)", count, skipped) + // 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 in %v (skipped %d nodes "+ + "due to invalid TLV streams)", count, time.Since(totalTime), + skipped) return nil } @@ -319,12 +400,32 @@ 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, kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( + totalTime = time.Now() + channelCount uint64 skippedChanCount uint64 policyCount uint64 @@ -359,6 +460,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, @@ -414,14 +520,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) >= int(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() @@ -443,78 +582,161 @@ func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, err) } - log.Infof("Migrated %d channels and %d policies from KV to SQL "+ + 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 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 } -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) } @@ -564,13 +786,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) @@ -580,14 +802,15 @@ func migrateSingleChannel(ctx context.Context, 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 ( + totalTime = time.Now() + count uint64 pruneTipHeight uint32 pruneTipHash chainhash.Hash @@ -599,55 +822,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) >= int(cfg.MaxBatchSize) { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch "+ + "validation failed: %w", err) + } } s.Do(func() { @@ -668,6 +928,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) @@ -696,54 +965,13 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, 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 } -// 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, @@ -773,14 +1001,15 @@ 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 ( + totalTime = time.Now() + count uint64 t0 = time.Now() @@ -789,6 +1018,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++ @@ -800,17 +1066,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) >= int(cfg.MaxBatchSize) { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch validation failed: %w", + err) + } } s.Do(func() { @@ -832,25 +1097,36 @@ func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend, err) } - log.Infof("Migrated %d closed SCIDs from KV to SQL", count) + // 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 in %s", count, + time.Since(totalTime)) 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 ( + totalTime = time.Now() + count uint64 t0 = time.Now() @@ -859,6 +1135,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 { @@ -895,22 +1244,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) >= int(cfg.MaxBatchSize) { + err := validateBatch() + if err != nil { + return fmt.Errorf("batch validation failed: %w", + err) + } } s.Do(func() { @@ -929,7 +1275,17 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, return fmt.Errorf("could not migrate zombie index: %w", err) } - log.Infof("Migrated %d zombie channels from KV to SQL", count) + // 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 in %s", count, + time.Since(totalTime)) return nil } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1c913b78c..956b8406d 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) @@ -97,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) @@ -131,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) @@ -140,6 +143,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 @@ -148,6 +152,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 @@ -591,15 +596,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 +3786,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 +3805,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(), @@ -4522,6 +4523,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/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 diff --git a/sqldb/sqlc/graph.sql.go b/sqldb/sqlc/graph.sql.go index 6db5d32c8..d27143ca8 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, @@ -1429,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 @@ -1729,6 +1941,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 @@ -1772,6 +2031,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 @@ -1962,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 55fd74b75..eaef3c777 100644 --- a/sqldb/sqlc/querier.go +++ b/sqldb/sqlc/querier.go @@ -42,11 +42,13 @@ 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) 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 @@ -68,7 +70,9 @@ 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) + 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) @@ -80,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 ae276a5cb..04f814083 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 @@ -330,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), @@ -850,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 @@ -894,6 +958,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 @@ -920,3 +990,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*/);