From 75691163a91f984e78b9a97ff49c9b541e9e72ca Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 11 Aug 2025 10:45:41 +0200 Subject: [PATCH 1/6] graph: remove outdated TODO This todo has been addressed. --- graph/builder.go | 3 --- 1 file changed, 3 deletions(-) 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 { From ee292786b18415048785bb3e9532974259a10983 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 11 Aug 2025 11:17:17 +0200 Subject: [PATCH 2/6] config+sqldb: make native SQL query config options configurable Here, we make the sql query option params (batch size and pagination size) configurable. The defaults for SQLite vs Postgres are still the same but will be changed in an upcoming commit. --- config_test_native_sql.go | 8 +++++++- lncfg/db.go | 8 ++++++++ sample-lnd.conf | 13 +++++++++++++ sqldb/config.go | 2 ++ sqldb/paginate.go | 9 ++++----- 5 files changed, 34 insertions(+), 6 deletions(-) diff --git a/config_test_native_sql.go b/config_test_native_sql.go index 92e8353bd..8ee51d4e7 100644 --- a/config_test_native_sql.go +++ b/config_test_native_sql.go @@ -10,6 +10,7 @@ import ( "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 +31,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..., ) diff --git a/lncfg/db.go b/lncfg/db.go index 3da1f4d3b..388d51803 100644 --- a/lncfg/db.go +++ b/lncfg/db.go @@ -115,10 +115,18 @@ func DefaultDB() *DB { }, Postgres: &sqldb.PostgresConfig{ MaxConnections: defaultPostgresMaxConnections, + QueryConfig: sqldb.QueryConfig{ + MaxBatchSize: 250, + MaxPageSize: 10000, + }, }, Sqlite: &sqldb.SqliteConfig{ MaxConnections: defaultSqliteMaxConnections, BusyTimeout: defaultSqliteBusyTimeout, + QueryConfig: sqldb.QueryConfig{ + MaxBatchSize: 250, + MaxPageSize: 10000, + }, }, UseNativeSQL: false, SkipNativeSQLMigration: false, 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..c5f49a98f 100644 --- a/sqldb/config.go +++ b/sqldb/config.go @@ -28,6 +28,7 @@ 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"` } // PostgresConfig holds the postgres database configuration. @@ -38,6 +39,7 @@ 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"` } func (p *PostgresConfig) Validate() error { diff --git a/sqldb/paginate.go b/sqldb/paginate.go index 2a51218c5..d81032508 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -6,20 +6,19 @@ import ( ) // 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 { return &QueryConfig{ MaxBatchSize: 250, From 185166b8d3c21c483d7422aebf85d83af1ce9516 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 11 Aug 2025 11:27:06 +0200 Subject: [PATCH 3/6] sqldb+config: validate maximum batch size config value Now that the SQL query config values are configurable, we add some validation to make sure that the user doesnt set a max batch size that is larger than the limits for sqlite/postgres that have been determined by the TestSQLSliceQueries test. --- lncfg/db.go | 3 +++ sqldb/config.go | 14 ++++++++++++++ sqldb/paginate.go | 40 ++++++++++++++++++++++++++++++++++++++++ sqldb/paginate_test.go | 14 ++++++++++---- 4 files changed, 67 insertions(+), 4 deletions(-) diff --git a/lncfg/db.go b/lncfg/db.go index 388d51803..927e76759 100644 --- a/lncfg/db.go +++ b/lncfg/db.go @@ -143,6 +143,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/sqldb/config.go b/sqldb/config.go index c5f49a98f..34de293c1 100644 --- a/sqldb/config.go +++ b/sqldb/config.go @@ -31,6 +31,15 @@ type SqliteConfig struct { 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. // //nolint:ll @@ -42,6 +51,7 @@ type PostgresConfig struct { 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") @@ -53,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 d81032508..e71f2e1c3 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -5,6 +5,18 @@ 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 +) + // QueryConfig holds configuration values for SQL queries. // //nolint:ll @@ -18,6 +30,34 @@ type QueryConfig struct { 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."` } +// 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 +} + // DefaultQueryConfig returns a default configuration for SQL queries. func DefaultQueryConfig() *QueryConfig { return &QueryConfig{ diff --git a/sqldb/paginate_test.go b/sqldb/paginate_test.go index dff0a16b1..5b0683a06 100644 --- a/sqldb/paginate_test.go +++ b/sqldb/paginate_test.go @@ -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 From 6a31e06817d051a922d7b3790d8870dff79f1f36 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 11 Aug 2025 13:09:01 +0200 Subject: [PATCH 4/6] graph/db+sqldb: find best default query cfg values for sqlite & postgres This commit adds a BenchmarkFindOptimalSQLQueryConfig test in the graph/db package which runs ForEachNode and ForEachChannel queries against a local backend using various different values for the sql QueryConfig struct. This is done to determine good default values to use for the config options for sqlite vs postgres. --- graph/db/benchmark_test.go | 138 +++++++++++++++++++++++++++++++++---- graph/db/test_postgres.go | 2 +- graph/db/test_sqlite.go | 2 +- lncfg/db.go | 10 +-- sqldb/paginate.go | 32 +++++++++ 5 files changed, 162 insertions(+), 22 deletions(-) diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index 1022197c5..783a2ce08 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..., ) @@ -763,3 +769,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/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/lncfg/db.go b/lncfg/db.go index 927e76759..0f55c736e 100644 --- a/lncfg/db.go +++ b/lncfg/db.go @@ -115,18 +115,12 @@ func DefaultDB() *DB { }, Postgres: &sqldb.PostgresConfig{ MaxConnections: defaultPostgresMaxConnections, - QueryConfig: sqldb.QueryConfig{ - MaxBatchSize: 250, - MaxPageSize: 10000, - }, + QueryConfig: *sqldb.DefaultPostgresConfig(), }, Sqlite: &sqldb.SqliteConfig{ MaxConnections: defaultSqliteMaxConnections, BusyTimeout: defaultSqliteBusyTimeout, - QueryConfig: sqldb.QueryConfig{ - MaxBatchSize: 250, - MaxPageSize: 10000, - }, + QueryConfig: *sqldb.DefaultSQLiteConfig(), }, UseNativeSQL: false, SkipNativeSQLMigration: false, diff --git a/sqldb/paginate.go b/sqldb/paginate.go index e71f2e1c3..279893add 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -15,6 +15,20 @@ const ( // 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. @@ -66,6 +80,24 @@ func DefaultQueryConfig() *QueryConfig { } } +// DefaultSQLiteConfig returns a default configuration for SQL queries to a +// SQLite backend. +func DefaultSQLiteConfig() *QueryConfig { + return &QueryConfig{ + 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, + } +} + // BatchQueryFunc represents a function that takes a batch of converted items // and returns results. type BatchQueryFunc[T any, R any] func(context.Context, []T) ([]R, error) From 1082eaaeb38fc55ea1c2a135d8b68259ddc1130a Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 12 Aug 2025 15:34:34 +0200 Subject: [PATCH 5/6] graph/db: fix progress logs --- graph/db/sql_migration.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 0b06e7cac..4d337ad94 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -603,6 +603,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 { @@ -786,6 +787,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 +876,7 @@ func migrateZombieIndex(ctx context.Context, kvBackend kvdb.Backend, } count++ + chunk++ err = sqlDB.UpsertZombieChannel( ctx, sqlc.UpsertZombieChannelParams{ From b1deddec446adc72c8d585b0f28b451d9a3fde88 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Wed, 13 Aug 2025 09:30:09 +0200 Subject: [PATCH 6/6] multi: remove DefaultQueryConfig And always make use of either the new DefaultSQLiteConfig or DefaultPostgresConfig. --- config_builder.go | 3 +- config_prod.go | 7 ++-- config_test_native_sql.go | 17 ++++++--- graph/db/benchmark_test.go | 7 +++- graph/db/sql_migration.go | 44 ++++++++++++----------- graph/db/sql_migration_test.go | 28 ++++++++------- graph/db/sql_store.go | 58 ++++++++++++------------------- itest/lnd_graph_migration_test.go | 7 +++- sqldb/paginate.go | 8 ----- sqldb/paginate_test.go | 8 ++--- 10 files changed, 95 insertions(+), 92 deletions(-) diff --git a/config_builder.go b/config_builder.go index 35a44e2c6..28e8dd426 100644 --- a/config_builder.go +++ b/config_builder.go @@ -1130,9 +1130,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 8ee51d4e7..01baacfae 100644 --- a/config_test_native_sql.go +++ b/config_test_native_sql.go @@ -7,7 +7,6 @@ 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" @@ -50,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/db/benchmark_test.go b/graph/db/benchmark_test.go index 783a2ce08..74cf68c84 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -449,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) } @@ -462,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() {}, ) diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index 4d337ad94..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 "+ @@ -703,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( @@ -716,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 "+ @@ -730,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 "+ 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/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/sqldb/paginate.go b/sqldb/paginate.go index 279893add..17e2fd40e 100644 --- a/sqldb/paginate.go +++ b/sqldb/paginate.go @@ -72,14 +72,6 @@ func (c *QueryConfig) Validate(sqlite bool) error { return nil } -// DefaultQueryConfig returns a default configuration for SQL queries. -func DefaultQueryConfig() *QueryConfig { - return &QueryConfig{ - MaxBatchSize: 250, - MaxPageSize: 10000, - } -} - // DefaultSQLiteConfig returns a default configuration for SQL queries to a // SQLite backend. func DefaultSQLiteConfig() *QueryConfig { diff --git a/sqldb/paginate_test.go b/sqldb/paginate_test.go index 5b0683a06..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} ) @@ -307,7 +307,7 @@ func TestSQLSliceQueries(t *testing.T) { err := ExecuteBatchQuery( ctx, - DefaultQueryConfig(), + DefaultSQLiteConfig(), queryParams, func(s string) string { return s