diff --git a/config_builder.go b/config_builder.go index 47eb96b50..5a2406057 100644 --- a/config_builder.go +++ b/config_builder.go @@ -1128,9 +1128,8 @@ func (d *DefaultDatabaseBuilder) BuildDatabase( continue } - migFn, ok := getSQLMigration( + migFn, ok := d.getSQLMigration( ctx, version, dbs.ChanStateDB.Backend, - *d.cfg.ActiveNetParams.GenesisHash, ) if !ok { continue diff --git a/config_prod.go b/config_prod.go index f593340c4..e9f320256 100644 --- a/config_prod.go +++ b/config_prod.go @@ -5,7 +5,6 @@ package lnd import ( "context" - "github.com/btcsuite/btcd/chaincfg/chainhash" graphdb "github.com/lightningnetwork/lnd/graph/db" "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/sqldb" @@ -30,9 +29,9 @@ func (d *DefaultDatabaseBuilder) getGraphStore(_ *sqldb.BaseDB, // NOTE: this is a no-op for the production build since all migrations that are // in production will also be in development builds, and so they are not // defined behind a build tag. -func getSQLMigration(ctx context.Context, version int, - kvBackend kvdb.Backend, - chain chainhash.Hash) (func(tx *sqlc.Queries) error, bool) { +func (d *DefaultDatabaseBuilder) getSQLMigration(ctx context.Context, + version int, kvBackend kvdb.Backend) (func(tx *sqlc.Queries) error, + bool) { return nil, false } diff --git a/config_test_native_sql.go b/config_test_native_sql.go index 92e8353bd..01baacfae 100644 --- a/config_test_native_sql.go +++ b/config_test_native_sql.go @@ -7,9 +7,9 @@ import ( "database/sql" "fmt" - "github.com/btcsuite/btcd/chaincfg/chainhash" graphdb "github.com/lightningnetwork/lnd/graph/db" "github.com/lightningnetwork/lnd/kvdb" + "github.com/lightningnetwork/lnd/lncfg" "github.com/lightningnetwork/lnd/sqldb" "github.com/lightningnetwork/lnd/sqldb/sqlc" ) @@ -30,10 +30,15 @@ func (d *DefaultDatabaseBuilder) getGraphStore(baseDB *sqldb.BaseDB, }, ) + queryConfig := d.cfg.DB.Sqlite.QueryConfig + if d.cfg.DB.Backend == lncfg.PostgresBackend { + queryConfig = d.cfg.DB.Postgres.QueryConfig + } + return graphdb.NewSQLStore( &graphdb.SQLStoreConfig{ ChainHash: *d.cfg.ActiveNetParams.GenesisHash, - QueryCfg: sqldb.DefaultQueryConfig(), + QueryCfg: &queryConfig, }, graphExecutor, opts..., ) @@ -44,15 +49,23 @@ func (d *DefaultDatabaseBuilder) getGraphStore(baseDB *sqldb.BaseDB, const graphSQLMigration = 9 // getSQLMigration returns a migration function for the given version. -func getSQLMigration(ctx context.Context, version int, - kvBackend kvdb.Backend, - chain chainhash.Hash) (func(tx *sqlc.Queries) error, bool) { +func (d *DefaultDatabaseBuilder) getSQLMigration(ctx context.Context, + version int, kvBackend kvdb.Backend) (func(tx *sqlc.Queries) error, + bool) { + + cfg := &graphdb.SQLStoreConfig{ + ChainHash: *d.cfg.ActiveNetParams.GenesisHash, + QueryCfg: &d.cfg.DB.Sqlite.QueryConfig, + } + if d.cfg.DB.Backend == lncfg.PostgresBackend { + cfg.QueryCfg = &d.cfg.DB.Postgres.QueryConfig + } switch version { case graphSQLMigration: return func(tx *sqlc.Queries) error { err := graphdb.MigrateGraphToSQL( - ctx, kvBackend, tx, chain, + ctx, cfg, kvBackend, tx, ) if err != nil { return fmt.Errorf("failed to migrate graph "+ diff --git a/graph/builder.go b/graph/builder.go index 5cfb79e4c..08016dd69 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1037,9 +1037,6 @@ func (b *Builder) AddEdge(ctx context.Context, edge *models.ChannelEdgeInfo, // Chain View is updated with the new edge if it is successfully added to the // graph. We only persist the channel if we currently dont have it at all in // our graph. -// -// TODO(elle): this currently also does funding-transaction validation. But this -// should be moved to the gossiper instead. func (b *Builder) addEdge(ctx context.Context, edge *models.ChannelEdgeInfo, op ...batch.SchedulerOption) error { diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 1022197c5..74cf68c84 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -60,10 +60,6 @@ var ( WithBatchCommitInterval(500 * time.Millisecond), } - // testSQLPaginationCfg is used to configure the pagination settings for - // the SQL stores we open for testing. - testSQLPaginationCfg = sqldb.DefaultQueryConfig() - // testSqlitePragmaOpts is a set of SQLite pragma options that we apply // to the SQLite databases we open for testing. testSqlitePragmaOpts = []string{ @@ -111,7 +107,8 @@ var ( name: "native-sqlite", open: func(b testing.TB) V1Store { return connectNativeSQLite( - b, nativeSQLSqlitePath, nativeSQLSqliteFile, + b, sqldb.DefaultSQLiteConfig(), + nativeSQLSqlitePath, nativeSQLSqliteFile, ) }, } @@ -130,15 +127,20 @@ var ( nativeSQLPostgresConn = dbConnection{ name: "native-postgres", open: func(b testing.TB) V1Store { - return connectNativePostgres(b, nativeSQLPostgresDNS) + return connectNativePostgres( + b, sqldb.DefaultPostgresConfig(), + nativeSQLPostgresDNS, + ) }, } ) // connectNativePostgres creates a V1Store instance backed by a native Postgres // database for testing purposes. -func connectNativePostgres(t testing.TB, dsn string) V1Store { - return newSQLStore(t, sqlPostgres(t, dsn)) +func connectNativePostgres(t testing.TB, cfg *sqldb.QueryConfig, + dsn string) V1Store { + + return newSQLStore(t, cfg, sqlPostgres(t, dsn)) } // sqlPostgres creates a sqldb.DB instance backed by a native Postgres database @@ -158,8 +160,10 @@ func sqlPostgres(t testing.TB, dsn string) BatchedSQLQueries { // connectNativeSQLite creates a V1Store instance backed by a native SQLite // database for testing purposes. -func connectNativeSQLite(t testing.TB, dbPath, file string) V1Store { - return newSQLStore(t, sqlSQLite(t, dbPath, file)) +func connectNativeSQLite(t testing.TB, cfg *sqldb.QueryConfig, dbPath, + file string) V1Store { + + return newSQLStore(t, cfg, sqlSQLite(t, dbPath, file)) } // sqlSQLite creates a sqldb.DB instance backed by a native SQLite database for @@ -277,11 +281,13 @@ func newSQLExecutor(t testing.TB, db sqldb.DB) BatchedSQLQueries { // newSQLStore creates a new SQLStore instance for testing using a provided // sqldb.DB instance. -func newSQLStore(t testing.TB, db BatchedSQLQueries) V1Store { +func newSQLStore(t testing.TB, cfg *sqldb.QueryConfig, + db BatchedSQLQueries) V1Store { + store, err := NewSQLStore( &SQLStoreConfig{ ChainHash: dbTestChain, - QueryCfg: testSQLPaginationCfg, + QueryCfg: cfg, }, db, testStoreOptions..., ) @@ -443,10 +449,12 @@ func TestPopulateViaMigration(t *testing.T) { log.SetLevel(btclog.LevelDebug) var ( + cfg = sqldb.DefaultSQLiteConfig() srcKVDB = kvdbSqlite(t, kvdbSqlitePath, kvdbSqliteFile) dstSQL = sqlSQLite(t, nativeSQLSqlitePath, nativeSQLSqliteFile) ) if testPostgres { + cfg = sqldb.DefaultPostgresConfig() srcKVDB = kvdbPostgres(t, kvdbPostgresDNS) dstSQL = sqlPostgres(t, nativeSQLPostgresDNS) } @@ -456,7 +464,10 @@ func TestPopulateViaMigration(t *testing.T) { err := dstSQL.ExecTx( ctx, sqldb.WriteTxOpt(), func(queries SQLQueries) error { return MigrateGraphToSQL( - ctx, srcKVDB, queries, dbTestChain, + ctx, &SQLStoreConfig{ + QueryCfg: cfg, + ChainHash: dbTestChain, + }, srcKVDB, queries, ) }, func() {}, ) @@ -763,3 +774,111 @@ func BenchmarkGraphReadMethods(b *testing.B) { } } } + +// BenchmarkFindOptimalSQLQueryConfig uses the ForEachNode and ForEachChannel +// methods to find the optimal maximum sqldb QueryConfig values for a given +// database backend. This is useful for determining the best default values for +// each backend. The ForEachNode and ForEachChannel methods are used since +// they make use of both batching and pagination. +func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) { + // NOTE: Set this to true if you want to test with a postgres backend. + testPostgres := false + + // NOTE: Set this to true if you want to test with various batch sizes. + // By default, page sizes will be tested. + testBatching := false + + // Set the various page sizes we want to test. + // + // NOTE: these are the sqlite paging testing values. + testSizes := []int{20, 50, 100, 150, 500} + + configOption := "MaxPageSize" + if testBatching { + configOption = "MaxBatchSize" + + testSizes = []int{ + 50, 100, 150, 200, 250, 300, 350, + } + } + + dbName := "sqlite" + if testPostgres { + dbName = "postgres" + + // Set the various page sizes we want to test. + // + // NOTE: these are the postgres paging values. + testSizes = []int{5000, 7000, 10000, 12000} + + if testBatching { + testSizes = []int{ + 1000, 2000, 5000, 7000, 10000, + } + } + } + + for _, size := range testSizes { + b.Run(fmt.Sprintf("%s-%s-%d", configOption, dbName, size), + func(b *testing.B) { + ctx := context.Background() + + cfg := sqldb.DefaultSQLiteConfig() + if testPostgres { + cfg = sqldb.DefaultPostgresConfig() + } + + if testBatching { + cfg.MaxBatchSize = size + } else { + cfg.MaxPageSize = int32(size) + } + + store := connectNativeSQLite( + b, cfg, nativeSQLSqlitePath, + nativeSQLSqliteFile, + ) + if testPostgres { + store = connectNativePostgres( + b, cfg, nativeSQLPostgresDNS, + ) + } + + // Reset timer to exclude setup time. + b.ResetTimer() + + for i := 0; i < b.N; i++ { + var ( + numNodes = 0 + numChannels = 0 + ) + + //nolint:ll + err := store.ForEachNode( + ctx, + func(_ *models.LightningNode) error { + numNodes++ + + return nil + }, func() {}, + ) + require.NoError(b, err) + + //nolint:ll + err = store.ForEachChannel( + ctx, + func(_ *models.ChannelEdgeInfo, + _, + _ *models.ChannelEdgePolicy) error { + + numChannels++ + + return nil + }, func() {}, + ) + require.NoError(b, err) + } + }, + ) + } +} diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 0b06e7cac..9b22fbf03 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -26,8 +26,8 @@ import ( // NOTE: this is currently not called from any code path. It is called via tests // only for now and will be called from the main lnd binary once the // migration is fully implemented and tested. -func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend, - sqlDB SQLQueries, chain chainhash.Hash) error { +func MigrateGraphToSQL(ctx context.Context, cfg *SQLStoreConfig, + kvBackend kvdb.Backend, sqlDB SQLQueries) error { log.Infof("Starting migration of the graph store from KV to SQL") t0 := time.Now() @@ -43,7 +43,8 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend, } // 1) Migrate all the nodes. - if err := migrateNodes(ctx, kvBackend, sqlDB); err != nil { + err = migrateNodes(ctx, cfg.QueryCfg, kvBackend, sqlDB) + if err != nil { return fmt.Errorf("could not migrate nodes: %w", err) } @@ -53,7 +54,7 @@ func MigrateGraphToSQL(ctx context.Context, kvBackend kvdb.Backend, } // 3) Migrate all the channels and channel policies. - err = migrateChannelsAndPolicies(ctx, kvBackend, sqlDB, chain) + err = migrateChannelsAndPolicies(ctx, cfg, kvBackend, sqlDB) if err != nil { return fmt.Errorf("could not migrate channels and policies: %w", err) @@ -108,8 +109,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. -func migrateNodes(ctx context.Context, kvBackend kvdb.Backend, - sqlDB SQLQueries) error { +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. @@ -192,7 +193,7 @@ func migrateNodes(ctx context.Context, kvBackend kvdb.Backend, pub, id, dbNode.ID) } - migratedNode, err := buildNode(ctx, sqlDB, dbNode) + 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", @@ -320,8 +321,8 @@ func migrateSourceNode(ctx context.Context, kvdb kvdb.Backend, // migrateChannelsAndPolicies migrates all channels and their policies // from the KV backend to the SQL database. -func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend, - sqlDB SQLQueries, chain chainhash.Hash) error { +func migrateChannelsAndPolicies(ctx context.Context, cfg *SQLStoreConfig, + kvBackend kvdb.Backend, sqlDB SQLQueries) error { var ( channelCount uint64 @@ -373,9 +374,10 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend, // info, but rather rely on the chain hash LND is running with. // So this is our way of ensuring that LND is running on the // correct network at migration time. - if channel.ChainHash != chain { + if channel.ChainHash != cfg.ChainHash { return fmt.Errorf("channel %d has chain hash %s, "+ - "expected %s", scid, channel.ChainHash, chain) + "expected %s", scid, channel.ChainHash, + cfg.ChainHash) } // Sanity check to ensure that the channel has valid extra @@ -413,7 +415,8 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend, chunk++ err = migrateSingleChannel( - ctx, sqlDB, channel, policy1, policy2, migChanPolicy, + ctx, cfg, sqlDB, channel, policy1, policy2, + migChanPolicy, ) if err != nil { return fmt.Errorf("could not migrate channel %d: %w", @@ -448,8 +451,8 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend, return nil } -func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries, - channel *models.ChannelEdgeInfo, +func migrateSingleChannel(ctx context.Context, cfg *SQLStoreConfig, + sqlDB SQLQueries, channel *models.ChannelEdgeInfo, policy1, policy2 *models.ChannelEdgePolicy, migChanPolicy func(*models.ChannelEdgePolicy) error) error { @@ -508,7 +511,7 @@ func migrateSingleChannel(ctx context.Context, sqlDB SQLQueries, } migChan, migPol1, migPol2, err := getAndBuildChanAndPolicies( - ctx, sqlDB, row, channel.ChainHash, + ctx, cfg, sqlDB, row, ) if err != nil { return fmt.Errorf("could not build migrated channel and "+ @@ -603,6 +606,7 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, hash *chainhash.Hash) error { count++ + chunk++ // Keep track of the prune tip height and hash. if height > pruneTipHeight { @@ -702,9 +706,9 @@ func migratePruneLog(ctx context.Context, kvBackend kvdb.Backend, // 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, db SQLQueries, - row sqlc.GetChannelBySCIDWithPoliciesRow, - chain chainhash.Hash) (*models.ChannelEdgeInfo, +func getAndBuildChanAndPolicies(ctx context.Context, cfg *SQLStoreConfig, + db SQLQueries, + row sqlc.GetChannelBySCIDWithPoliciesRow) (*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { node1, node2, err := buildNodeVertices( @@ -715,7 +719,7 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries, } edge, err := getAndBuildEdgeInfo( - ctx, db, chain, row.GraphChannel, node1, node2, + ctx, cfg, db, row.GraphChannel, node1, node2, ) if err != nil { return nil, nil, nil, fmt.Errorf("unable to build channel "+ @@ -729,7 +733,8 @@ func getAndBuildChanAndPolicies(ctx context.Context, db SQLQueries, } policy1, policy2, err := getAndBuildChanPolicies( - ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2, + ctx, cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID, node1, + node2, ) if err != nil { return nil, nil, nil, fmt.Errorf("unable to build channel "+ @@ -786,6 +791,7 @@ func migrateClosedSCIDIndex(ctx context.Context, kvBackend kvdb.Backend, ) migrateSingleClosedSCID := func(scid lnwire.ShortChannelID) error { count++ + chunk++ chanIDB := channelIDToBytes(scid.ToUint64()) err := sqlDB.InsertClosedChannel(ctx, chanIDB) @@ -874,6 +880,7 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, } count++ + chunk++ err = sqlDB.UpsertZombieChannel( ctx, sqlc.UpsertZombieChannelParams{ diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index d134b2fb1..c555e433d 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -333,9 +333,7 @@ func TestMigrateGraphToSQL(t *testing.T) { require.True(t, ok) // Run the migration. - err := MigrateGraphToSQL( - ctx, kvDB.db, sql.db, testChain, - ) + err := MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db) require.NoError(t, err) // Validate that the two databases are now in sync. @@ -753,6 +751,18 @@ func TestMigrationWithChannelDB(t *testing.T) { // bbolt channel.db file, set this to "channel.db". const fileName = "channel.sqlite" + cfg := &SQLStoreConfig{ + ChainHash: chain, + QueryCfg: sqldb.DefaultPostgresConfig(), + } + + // Determine if we are using a SQLite file or a Bolt DB file. + var isSqlite bool + if strings.HasSuffix(fileName, ".sqlite") { + isSqlite = true + cfg.QueryCfg = sqldb.DefaultSQLiteConfig() + } + // Set up logging for the test. UseLogger(btclog.NewSLogger(btclog.NewDefaultHandler(os.Stdout))) @@ -763,7 +773,7 @@ func TestMigrationWithChannelDB(t *testing.T) { err := graphStore.ExecTx( ctx, sqldb.WriteTxOpt(), func(tx SQLQueries) error { return MigrateGraphToSQL( - ctx, kvBackend, tx, chain, + ctx, cfg, kvBackend, tx, ) }, sqldb.NoOpReset, ) @@ -825,12 +835,6 @@ func TestMigrationWithChannelDB(t *testing.T) { }, } - // Determine if we are using a SQLite file or a Bolt DB file. - var isSqlite bool - if strings.HasSuffix(fileName, ".sqlite") { - isSqlite = true - } - for _, test := range tests { t.Run(test.name, func(t *testing.T) { chanDBPath := path.Join(test.dbPath, fileName) @@ -1125,7 +1129,7 @@ func runTestMigration(t *testing.T, populateKV func(t *testing.T, db *KVStore), // Run the migration. err := MigrateGraphToSQL( - ctx, kvDB.db, sql.db, testChain, + ctx, sql.cfg, kvDB.db, sql.db, ) require.NoError(t, err) @@ -1312,7 +1316,7 @@ func testMigrateGraphToSQLRapidOnce(t *testing.T, rt *rapid.T, } // Run the migration. - err := MigrateGraphToSQL(ctx, kvDB.db, sql.db, testChain) + err := MigrateGraphToSQL(ctx, sql.cfg, kvDB.db, sql.db) require.NoError(t, err) // Create a slice of all nodes. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 3276f67d7..1c913b78c 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -255,7 +255,7 @@ func (s *SQLStore) FetchLightningNode(ctx context.Context, var node *models.LightningNode err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { var err error - _, node, err = getNodeByPubKey(ctx, db, pubKey) + _, node, err = getNodeByPubKey(ctx, s.cfg.QueryCfg, db, pubKey) return err }, sqldb.NoOpReset) @@ -483,7 +483,7 @@ func (s *SQLStore) SourceNode(ctx context.Context) (*models.LightningNode, err) } - _, node, err = getNodeByPubKey(ctx, db, nodePub) + _, node, err = getNodeByPubKey(ctx, s.cfg.QueryCfg, db, nodePub) return err }, sqldb.NoOpReset) @@ -779,7 +779,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, } _, otherNode, err := getNodeByPubKey( - ctx, db, otherNodePub, + ctx, s.cfg.QueryCfg, db, otherNodePub, ) if err != nil { return fmt.Errorf("unable to fetch "+ @@ -1761,8 +1761,7 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) ( } edge, err = getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, - node2, + ctx, s.cfg, db, row.GraphChannel, node1, node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1776,7 +1775,8 @@ func (s *SQLStore) FetchChannelEdgesByID(chanID uint64) ( } policy1, policy2, err = getAndBuildChanPolicies( - ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2, + ctx, s.cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID, + node1, node2, ) if err != nil { return fmt.Errorf("unable to build channel "+ @@ -1833,8 +1833,7 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) ( } edge, err = getAndBuildEdgeInfo( - ctx, db, s.cfg.ChainHash, row.GraphChannel, node1, - node2, + ctx, s.cfg, db, row.GraphChannel, node1, node2, ) if err != nil { return fmt.Errorf("unable to build channel info: %w", @@ -1848,7 +1847,8 @@ func (s *SQLStore) FetchChannelEdgesByOutpoint(op *wire.OutPoint) ( } policy1, policy2, err = getAndBuildChanPolicies( - ctx, db, dbPol1, dbPol2, edge.ChannelID, node1, node2, + ctx, s.cfg.QueryCfg, db, dbPol1, dbPol2, edge.ChannelID, + node1, node2, ) if err != nil { return fmt.Errorf("unable to build channel "+ @@ -3171,7 +3171,7 @@ func updateChanEdgePolicy(ctx context.Context, tx SQLQueries, } // getNodeByPubKey attempts to look up a target node by its public key. -func getNodeByPubKey(ctx context.Context, db SQLQueries, +func getNodeByPubKey(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries, pubKey route.Vertex) (int64, *models.LightningNode, error) { dbNode, err := db.GetNodeByPubKey( @@ -3186,7 +3186,7 @@ func getNodeByPubKey(ctx context.Context, db SQLQueries, return 0, nil, fmt.Errorf("unable to fetch node: %w", err) } - node, err := buildNode(ctx, db, dbNode) + node, err := buildNode(ctx, cfg, db, dbNode) if err != nil { return 0, nil, fmt.Errorf("unable to build node: %w", err) } @@ -3210,14 +3210,9 @@ func buildCacheableChannelInfo(scid []byte, capacity int64, node1Pub, // buildNode constructs a LightningNode instance from the given database node // record. The node's features, addresses and extra signed fields are also // fetched from the database and set on the node. -func buildNode(ctx context.Context, db SQLQueries, +func buildNode(ctx context.Context, cfg *sqldb.QueryConfig, db SQLQueries, dbNode sqlc.GraphNode) (*models.LightningNode, error) { - // NOTE: buildNode is only used to load the data for a single node, and - // so no paged queries will be performed. This means that it's ok to - // used pass in default config values here. - cfg := sqldb.DefaultQueryConfig() - data, err := batchLoadNodeData(ctx, cfg, db, []int64{dbNode.ID}) if err != nil { return nil, fmt.Errorf("unable to batch load node data: %w", @@ -3945,22 +3940,21 @@ func upsertChanPolicyExtraSignedFields(ctx context.Context, db SQLQueries, // getAndBuildEdgeInfo builds a models.ChannelEdgeInfo instance from the // provided dbChanRow and also fetches any other required information // to construct the edge info. -func getAndBuildEdgeInfo(ctx context.Context, db SQLQueries, - chain chainhash.Hash, dbChan sqlc.GraphChannel, node1, +func getAndBuildEdgeInfo(ctx context.Context, cfg *SQLStoreConfig, + db SQLQueries, dbChan sqlc.GraphChannel, node1, node2 route.Vertex) (*models.ChannelEdgeInfo, error) { - // NOTE: getAndBuildEdgeInfo is only used to load the data for a single - // edge, and so no paged queries will be performed. This means that - // it's ok to used pass in default config values here. - cfg := sqldb.DefaultQueryConfig() - - data, err := batchLoadChannelData(ctx, cfg, db, []int64{dbChan.ID}, nil) + data, err := batchLoadChannelData( + ctx, cfg.QueryCfg, db, []int64{dbChan.ID}, nil, + ) if err != nil { return nil, fmt.Errorf("unable to batch load channel data: %w", err) } - return buildEdgeInfoWithBatchData(chain, dbChan, node1, node2, data) + return buildEdgeInfoWithBatchData( + cfg.ChainHash, dbChan, node1, node2, data, + ) } // buildEdgeInfoWithBatchData builds edge info using pre-loaded batch data. @@ -4059,9 +4053,9 @@ func buildNodeVertices(node1Pub, node2Pub []byte) (route.Vertex, // retrieves all the extra info required to build the complete // models.ChannelEdgePolicy types. It returns two policies, which may be nil if // the provided sqlc.GraphChannelPolicy records are nil. -func getAndBuildChanPolicies(ctx context.Context, db SQLQueries, - dbPol1, dbPol2 *sqlc.GraphChannelPolicy, channelID uint64, node1, - node2 route.Vertex) (*models.ChannelEdgePolicy, +func getAndBuildChanPolicies(ctx context.Context, cfg *sqldb.QueryConfig, + db SQLQueries, dbPol1, dbPol2 *sqlc.GraphChannelPolicy, + channelID uint64, node1, node2 route.Vertex) (*models.ChannelEdgePolicy, *models.ChannelEdgePolicy, error) { if dbPol1 == nil && dbPol2 == nil { @@ -4076,12 +4070,6 @@ func getAndBuildChanPolicies(ctx context.Context, db SQLQueries, policyIDs = append(policyIDs, dbPol2.ID) } - // NOTE: getAndBuildChanPolicies is only used to load the data for - // a maximum of two policies, and so no paged queries will be - // performed (unless the page size is one). So it's ok to use - // the default config values here. - cfg := sqldb.DefaultQueryConfig() - batchData, err := batchLoadChannelData(ctx, cfg, db, nil, policyIDs) if err != nil { return nil, nil, fmt.Errorf("unable to batch load channel "+ diff --git a/graph/db/test_postgres.go b/graph/db/test_postgres.go index 463a89db2..6134f0114 100644 --- a/graph/db/test_postgres.go +++ b/graph/db/test_postgres.go @@ -43,7 +43,7 @@ func NewTestDBWithFixture(t testing.TB, store, err := NewSQLStore( &SQLStoreConfig{ ChainHash: *chaincfg.MainNetParams.GenesisHash, - QueryCfg: sqldb.DefaultQueryConfig(), + QueryCfg: sqldb.DefaultPostgresConfig(), }, querier, ) require.NoError(t, err) diff --git a/graph/db/test_sqlite.go b/graph/db/test_sqlite.go index 3442bfd6e..c1c6d808f 100644 --- a/graph/db/test_sqlite.go +++ b/graph/db/test_sqlite.go @@ -28,7 +28,7 @@ func NewTestDBWithFixture(t testing.TB, _ *sqldb.TestPgFixture) V1Store { store, err := NewSQLStore( &SQLStoreConfig{ ChainHash: *chaincfg.MainNetParams.GenesisHash, - QueryCfg: sqldb.DefaultQueryConfig(), + QueryCfg: sqldb.DefaultSQLiteConfig(), }, newBatchQuerier(t), ) require.NoError(t, err) diff --git a/itest/lnd_graph_migration_test.go b/itest/lnd_graph_migration_test.go index 07abc3bb0..652b1e652 100644 --- a/itest/lnd_graph_migration_test.go +++ b/itest/lnd_graph_migration_test.go @@ -142,10 +142,15 @@ func openNativeSQLGraphDB(ht *lntest.HarnessTest, }, ) + queryCfg := sqldb.DefaultSQLiteConfig() + if hn.Cfg.DBBackend != node.BackendSqlite { + queryCfg = sqldb.DefaultPostgresConfig() + } + store, err := graphdb.NewSQLStore( &graphdb.SQLStoreConfig{ ChainHash: *ht.Miner().ActiveNet.GenesisHash, - QueryCfg: sqldb.DefaultQueryConfig(), + QueryCfg: queryCfg, }, executor, ) diff --git a/lncfg/db.go b/lncfg/db.go index 3da1f4d3b..0f55c736e 100644 --- a/lncfg/db.go +++ b/lncfg/db.go @@ -115,10 +115,12 @@ func DefaultDB() *DB { }, Postgres: &sqldb.PostgresConfig{ MaxConnections: defaultPostgresMaxConnections, + QueryConfig: *sqldb.DefaultPostgresConfig(), }, Sqlite: &sqldb.SqliteConfig{ MaxConnections: defaultSqliteMaxConnections, BusyTimeout: defaultSqliteBusyTimeout, + QueryConfig: *sqldb.DefaultSQLiteConfig(), }, UseNativeSQL: false, SkipNativeSQLMigration: false, @@ -135,6 +137,9 @@ func (db *DB) Validate() error { } case SqliteBackend: + if err := db.Sqlite.Validate(); err != nil { + return err + } case PostgresBackend: if err := db.Postgres.Validate(); err != nil { return err diff --git a/sample-lnd.conf b/sample-lnd.conf index cba40213c..22ffc3916 100644 --- a/sample-lnd.conf +++ b/sample-lnd.conf @@ -1601,6 +1601,12 @@ ; Whether to skip executing schema migrations. ; db.postgres.skipmigrations=false +; The maximum number of elements to use in a native-SQL batch query IN clause. +; db.postgres.query.max-batch-size=5000 + +; The maximum number of records to fetch at a time in a paginated query during +; native-SQL calls. +; db.postgres.query.max-page-size=10500 [sqlite] @@ -1617,6 +1623,13 @@ ; The maximum amount of time to wait to execute a query if the db is locked. ; db.sqlite.busytimeout=5s +; The maximum number of elements to use in a native-SQL batch query IN clause. +; db.sqlite.query.max-batch-size=250 + +; The maximum number of records to fetch at a time in a paginated query during +; native-SQL calls. +; db.sqlite.query.max-page-size=100 + ; Raw pragma option pairs to be used when opening the sqlite db. The flag ; can be specified multiple times to set multiple options. ; Default: diff --git a/sqldb/config.go b/sqldb/config.go index 24e58b498..34de293c1 100644 --- a/sqldb/config.go +++ b/sqldb/config.go @@ -28,6 +28,16 @@ type SqliteConfig struct { MaxConnections int `long:"maxconnections" description:"The maximum number of open connections to the database. Set to zero for unlimited."` PragmaOptions []string `long:"pragmaoptions" description:"A list of pragma options to set on a database connection. For example, 'auto_vacuum=incremental'. Note that the flag must be specified multiple times if multiple options are to be set."` SkipMigrations bool `long:"skipmigrations" description:"Skip applying migrations on startup."` + QueryConfig `group:"query" namespace:"query"` +} + +// Validate checks that the SqliteConfig values are valid. +func (p *SqliteConfig) Validate() error { + if err := p.QueryConfig.Validate(true); err != nil { + return fmt.Errorf("invalid query config: %w", err) + } + + return nil } // PostgresConfig holds the postgres database configuration. @@ -38,8 +48,10 @@ type PostgresConfig struct { Timeout time.Duration `long:"timeout" description:"Database connection timeout. Set to zero to disable."` MaxConnections int `long:"maxconnections" description:"The maximum number of open connections to the database. Set to zero for unlimited."` SkipMigrations bool `long:"skipmigrations" description:"Skip applying migrations on startup."` + QueryConfig `group:"query" namespace:"query"` } +// Validate checks that the PostgresConfig values are valid. func (p *PostgresConfig) Validate() error { if p.Dsn == "" { return fmt.Errorf("DSN is required") @@ -51,5 +63,9 @@ func (p *PostgresConfig) Validate() error { return fmt.Errorf("invalid DSN: %w", err) } + if err := p.QueryConfig.Validate(false); err != nil { + return fmt.Errorf("invalid query config: %w", err) + } + return nil } diff --git a/sqldb/paginate.go b/sqldb/paginate.go index 2a51218c5..17e2fd40e 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -5,25 +5,88 @@ import ( "fmt" ) +const ( + // maxSQLiteBatchSize is the maximum number of items that can be + // included in a batch query IN clause for SQLite. This was determined + // using the TestSQLSliceQueries test. + maxSQLiteBatchSize = 32766 + + // maxPostgresBatchSize is the maximum number of items that can be + // included in a batch query IN clause for Postgres. This was determined + // using the TestSQLSliceQueries test. + maxPostgresBatchSize = 65535 + + // defaultSQLitePageSize is the default page size for SQLite queries. + defaultSQLitePageSize = 100 + + // defaultPostgresPageSize is the default page size for Postgres + // queries. + defaultPostgresPageSize = 10500 + + // defaultSQLiteBatchSize is the default batch size for SQLite queries. + defaultSQLiteBatchSize = 250 + + // defaultPostgresBatchSize is the default batch size for Postgres + // queries. + defaultPostgresBatchSize = 5000 +) + // QueryConfig holds configuration values for SQL queries. +// +//nolint:ll type QueryConfig struct { // MaxBatchSize is the maximum number of items included in a batch // query IN clauses list. - MaxBatchSize int + 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."` // MaxPageSize is the maximum number of items returned in a single page // of results. This is used for paginated queries. - MaxPageSize int32 + 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."` } -// DefaultQueryConfig returns a default configuration for SQL queries. -// -// TODO(elle): make configurable & have different defaults for SQLite and -// Postgres. -func DefaultQueryConfig() *QueryConfig { +// Validate checks that the QueryConfig values are valid. +func (c *QueryConfig) Validate(sqlite bool) error { + if c.MaxBatchSize <= 0 { + return fmt.Errorf("max batch size must be greater than "+ + "zero, got %d", c.MaxBatchSize) + } + if c.MaxPageSize <= 0 { + return fmt.Errorf("max page size must be greater than "+ + "zero, got %d", c.MaxPageSize) + } + + if sqlite { + if c.MaxBatchSize > maxSQLiteBatchSize { + return fmt.Errorf("max batch size for SQLite cannot "+ + "exceed %d, got %d", maxSQLiteBatchSize, + c.MaxBatchSize) + } + } else { + if c.MaxBatchSize > maxPostgresBatchSize { + return fmt.Errorf("max batch size for Postgres cannot "+ + "exceed %d, got %d", maxPostgresBatchSize, + c.MaxBatchSize) + } + } + + return nil +} + +// DefaultSQLiteConfig returns a default configuration for SQL queries to a +// SQLite backend. +func DefaultSQLiteConfig() *QueryConfig { return &QueryConfig{ - MaxBatchSize: 250, - MaxPageSize: 10000, + MaxBatchSize: defaultSQLiteBatchSize, + MaxPageSize: defaultSQLitePageSize, + } +} + +// DefaultPostgresConfig returns a default configuration for SQL queries to a +// Postgres backend. +func DefaultPostgresConfig() *QueryConfig { + return &QueryConfig{ + MaxBatchSize: defaultPostgresBatchSize, + MaxPageSize: defaultPostgresPageSize, } } diff --git a/sqldb/paginate_test.go b/sqldb/paginate_test.go index dff0a16b1..f62bf65b4 100644 --- a/sqldb/paginate_test.go +++ b/sqldb/paginate_test.go @@ -22,7 +22,7 @@ func TestExecuteBatchQuery(t *testing.T) { t.Run("empty input returns nil", func(t *testing.T) { var ( - cfg = DefaultQueryConfig() + cfg = DefaultSQLiteConfig() inputItems []int ) @@ -144,7 +144,7 @@ func TestExecuteBatchQuery(t *testing.T) { t.Run("query function error is propagated", func(t *testing.T) { var ( - cfg = DefaultQueryConfig() + cfg = DefaultSQLiteConfig() inputItems = []int{1, 2, 3} ) @@ -174,7 +174,7 @@ func TestExecuteBatchQuery(t *testing.T) { t.Run("callback error is propagated", func(t *testing.T) { var ( - cfg = DefaultQueryConfig() + cfg = DefaultSQLiteConfig() inputItems = []int{1, 2, 3} ) @@ -281,11 +281,17 @@ func TestSQLSliceQueries(t *testing.T) { break } - x *= 10 + // If it succeeded, we expect it to be under the maximum that + // we expect for this DB. + if isSQLite { + require.LessOrEqual(t, x, maxSQLiteBatchSize, + "SQLite should not exceed 32766 parameters") + } else { + require.LessOrEqual(t, x, maxPostgresBatchSize, + "Postgres should not exceed 65535 parameters") + } - // Just to make sure that the test doesn't carry on too long, - // we assert that we don't exceed a reasonable limit. - require.LessOrEqual(t, x, 100000) + x *= 10 } // Now that we have found the limit that the raw query can handle, we @@ -301,7 +307,7 @@ func TestSQLSliceQueries(t *testing.T) { err := ExecuteBatchQuery( ctx, - DefaultQueryConfig(), + DefaultSQLiteConfig(), queryParams, func(s string) string { return s