From c4e6f23c5bda72dbaa1384d958d1dc8665dcac58 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 27 May 2025 18:53:31 +0200 Subject: [PATCH] multi: add a re-usable TxOptions type Add a re-usable implementation of the sqldb.TxOptions interface and make use of this in the various spots (invoices, batch and graph/db) where we had previously defined individual implementations that were all doing the same thing. --- batch/batch.go | 17 +-------- batch/batch_test.go | 17 ++++----- batch/scheduler.go | 10 ++--- graph/db/sql_store.go | 61 +++++++------------------------ invoices/kv_sql_migration_test.go | 3 +- invoices/sql_migration_test.go | 4 +- invoices/sql_store.go | 57 +++++++++-------------------- sqldb/interfaces.go | 29 +++++++++++++++ sqldb/migrations.go | 15 +------- 9 files changed, 79 insertions(+), 134 deletions(-) diff --git a/batch/batch.go b/batch/batch.go index e31701439..bd5cda5e2 100644 --- a/batch/batch.go +++ b/batch/batch.go @@ -14,19 +14,6 @@ var errSolo = errors.New( "batch function returned an error and should be re-run solo", ) -// txOpts implements the sqldb.TxOptions interface. It is used to indicate that -// the transaction can be read-only or not transaction. -type txOpts struct { - readOnly bool -} - -// ReadOnly returns true if the transaction should be read only. -// -// NOTE: This is part of the sqldb.TxOptions interface. -func (t *txOpts) ReadOnly() bool { - return t.readOnly -} - type request[Q any] struct { *Request[Q] errChan chan error @@ -38,7 +25,7 @@ type batch[Q any] struct { reqs []*request[Q] clear func(b *batch[Q]) locker sync.Locker - txOpts txOpts + txOpts sqldb.TxOptions } // trigger is the entry point for the batch and ensures that run is started at @@ -68,7 +55,7 @@ func (b *batch[Q]) run(ctx context.Context) { // that fail will be retried individually. for len(b.reqs) > 0 { var failIdx = -1 - err := b.db.ExecTx(ctx, &b.txOpts, func(tx Q) error { + err := b.db.ExecTx(ctx, b.txOpts, func(tx Q) error { for i, req := range b.reqs { err := req.Do(tx) if err != nil { diff --git a/batch/batch_test.go b/batch/batch_test.go index dd9a5fb4f..49ce0eabc 100644 --- a/batch/batch_test.go +++ b/batch/batch_test.go @@ -550,7 +550,7 @@ func benchmarkSQLBatching(b *testing.B, sqlite bool) { } ctx := context.Background() - var opts txOpts + opts := sqldb.WriteTxOpt() // writeRecord is a helper that adds a single new invoice to the // database. It uses the 'i' argument to create a unique hash for the @@ -578,13 +578,12 @@ func benchmarkSQLBatching(b *testing.B, sqlite bool) { var hash [8]byte binary.BigEndian.PutUint64(hash[:], uint64(N-1)) - err := tx.ExecTx( - ctx, &txOpts{}, func(queries *sqlc.Queries) error { - _, err := queries.GetInvoiceByHash(ctx, hash[:]) - require.NoError(b, err) + err := tx.ExecTx(ctx, opts, func(queries *sqlc.Queries) error { + _, err := queries.GetInvoiceByHash(ctx, hash[:]) + require.NoError(b, err) - return nil - }, func() {}, + return nil + }, func() {}, ) require.NoError(b, err) } @@ -602,7 +601,7 @@ func benchmarkSQLBatching(b *testing.B, sqlite bool) { defer wg.Done() err := db.ExecTx( - ctx, &opts, + ctx, opts, func(tx *sqlc.Queries) error { writeRecord(b, tx, int64(j)) return nil @@ -624,7 +623,7 @@ func benchmarkSQLBatching(b *testing.B, sqlite bool) { b.ResetTimer() err := db.ExecTx( - ctx, &opts, + ctx, opts, func(tx *sqlc.Queries) error { for i := 0; i < b.N; i++ { writeRecord(b, tx, int64(i)) diff --git a/batch/scheduler.go b/batch/scheduler.go index dbd319f30..a9258f735 100644 --- a/batch/scheduler.go +++ b/batch/scheduler.go @@ -65,9 +65,7 @@ func (s *TimeScheduler[Q]) Execute(ctx context.Context, r *Request[Q]) error { // By default, we assume that the batch is read-only, // and we only upgrade it to read-write if a request // is added that is not read-only. - txOpts: txOpts{ - readOnly: true, - }, + txOpts: sqldb.ReadTxOpt(), } trigger := s.b.trigger time.AfterFunc(s.duration, func() { @@ -78,8 +76,8 @@ func (s *TimeScheduler[Q]) Execute(ctx context.Context, r *Request[Q]) error { // We only upgrade the batch to read-write if the new request is not // read-only. If it is already read-write, we don't need to do anything. - if s.b.txOpts.readOnly && !r.Opts.ReadOnly { - s.b.txOpts.readOnly = false + if s.b.txOpts.ReadOnly() && !r.Opts.ReadOnly { + s.b.txOpts = sqldb.WriteTxOpt() } // If this is a non-lazy request, we'll execute the batch immediately. @@ -109,7 +107,7 @@ func (s *TimeScheduler[Q]) Execute(ctx context.Context, r *Request[Q]) error { } // Otherwise, run the request on its own. - commitErr := s.db.ExecTx(ctx, &txOpts, func(tx Q) error { + commitErr := s.db.ExecTx(ctx, txOpts, func(tx Q) error { return req.Do(tx) }, func() { if req.Reset != nil { diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 1b4cbc566..331ff9223 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -139,27 +139,6 @@ func NewSQLStore(db BatchedSQLQueries, kvStore *KVStore, return s, nil } -// TxOptions defines the set of db txn options the SQLQueries -// understands. -type TxOptions struct { - // readOnly governs if a read only transaction is needed or not. - readOnly bool -} - -// ReadOnly returns true if the transaction should be read only. -// -// NOTE: This implements the TxOptions. -func (a *TxOptions) ReadOnly() bool { - return a.readOnly -} - -// NewReadTx creates a new read transaction option set. -func NewReadTx() *TxOptions { - return &TxOptions{ - readOnly: true, - } -} - // AddLightningNode adds a vertex/node to the graph database. If the node is not // in the database from before, this will add a new, unconnected one to the // graph. If it is present from before, this will update that node's @@ -192,11 +171,8 @@ func (s *SQLStore) FetchLightningNode(pubKey route.Vertex) ( ctx := context.TODO() - var ( - readTx = NewReadTx() - node *models.LightningNode - ) - err := s.db.ExecTx(ctx, readTx, func(db SQLQueries) error { + var node *models.LightningNode + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { var err error _, node, err = getNodeByPubKey(ctx, db, pubKey) @@ -222,11 +198,10 @@ func (s *SQLStore) HasLightningNode(pubKey [33]byte) (time.Time, bool, ctx := context.TODO() var ( - readTx = NewReadTx() exists bool lastUpdate time.Time ) - err := s.db.ExecTx(ctx, readTx, func(db SQLQueries) error { + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( ctx, sqlc.GetNodeByPubKeyParams{ Version: int16(ProtocolV1), @@ -266,11 +241,10 @@ func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, ctx := context.TODO() var ( - readTx = NewReadTx() addresses []net.Addr known bool ) - err := s.db.ExecTx(ctx, readTx, func(db SQLQueries) error { + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { var err error known, addresses, err = getNodeAddresses( ctx, db, nodePub.SerializeCompressed(), @@ -297,8 +271,7 @@ func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr, func (s *SQLStore) DeleteLightningNode(pubKey route.Vertex) error { ctx := context.TODO() - var writeTxOpts TxOptions - err := s.db.ExecTx(ctx, &writeTxOpts, func(db SQLQueries) error { + err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { res, err := db.DeleteNodeByPubKey( ctx, sqlc.DeleteNodeByPubKeyParams{ Version: int16(ProtocolV1), @@ -346,11 +319,10 @@ func (s *SQLStore) FetchNodeFeatures(nodePub route.Vertex) ( // NOTE: part of the V1Store interface. func (s *SQLStore) LookupAlias(pub *btcec.PublicKey) (string, error) { var ( - ctx = context.TODO() - readTx = NewReadTx() - alias string + ctx = context.TODO() + alias string ) - err := s.db.ExecTx(ctx, readTx, func(db SQLQueries) error { + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( ctx, sqlc.GetNodeByPubKeyParams{ Version: int16(ProtocolV1), @@ -387,11 +359,8 @@ func (s *SQLStore) LookupAlias(pub *btcec.PublicKey) (string, error) { func (s *SQLStore) SourceNode() (*models.LightningNode, error) { ctx := context.TODO() - var ( - readTx = NewReadTx() - node *models.LightningNode - ) - err := s.db.ExecTx(ctx, readTx, func(db SQLQueries) error { + var node *models.LightningNode + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { _, nodePub, err := getSourceNode(ctx, db, ProtocolV1) if err != nil { return fmt.Errorf("unable to fetch V1 source node: %w", @@ -416,9 +385,8 @@ func (s *SQLStore) SourceNode() (*models.LightningNode, error) { // NOTE: part of the V1Store interface. func (s *SQLStore) SetSourceNode(node *models.LightningNode) error { ctx := context.TODO() - var writeTxOpts TxOptions - return s.db.ExecTx(ctx, &writeTxOpts, func(db SQLQueries) error { + return s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { id, err := upsertNode(ctx, db, node) if err != nil { return fmt.Errorf("unable to upsert source node: %w", @@ -456,11 +424,8 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime, ctx := context.TODO() - var ( - readTx = NewReadTx() - nodes []models.LightningNode - ) - err := s.db.ExecTx(ctx, readTx, func(db SQLQueries) error { + var nodes []models.LightningNode + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNodes, err := db.GetNodesByLastUpdateRange( ctx, sqlc.GetNodesByLastUpdateRangeParams{ StartTime: sqldb.SQLInt64(startTime.Unix()), diff --git a/invoices/kv_sql_migration_test.go b/invoices/kv_sql_migration_test.go index b3048a17b..6a7804b15 100644 --- a/invoices/kv_sql_migration_test.go +++ b/invoices/kv_sql_migration_test.go @@ -70,9 +70,8 @@ func TestMigrationWithChannelDB(t *testing.T) { ctxb := context.Background() const batchSize = 11 - var opts sqldb.MigrationTxOptions err := sqlStore.ExecTx( - ctxb, &opts, func(tx *sqlc.Queries) error { + ctxb, sqldb.WriteTxOpt(), func(tx *sqlc.Queries) error { return invpkg.MigrateInvoicesToSQL( ctxb, kvStore.Backend, kvStore, tx, batchSize, diff --git a/invoices/sql_migration_test.go b/invoices/sql_migration_test.go index 3a6d9cc51..721f7ff59 100644 --- a/invoices/sql_migration_test.go +++ b/invoices/sql_migration_test.go @@ -317,8 +317,8 @@ func testMigrateSingleInvoiceRapid(t *rapid.T, store *SQLStore, mpp bool, invoices[hash] = invoice } - var ops SQLInvoiceQueriesTxOptions - err := store.db.ExecTx(ctxb, &ops, func(tx SQLInvoiceQueries) error { + ops := sqldb.WriteTxOpt() + err := store.db.ExecTx(ctxb, ops, func(tx SQLInvoiceQueries) error { for hash, invoice := range invoices { err := MigrateSingleInvoice(ctxb, tx, invoice, hash) require.NoError(t, err) diff --git a/invoices/sql_store.go b/invoices/sql_store.go index 1372a3131..c99629828 100644 --- a/invoices/sql_store.go +++ b/invoices/sql_store.go @@ -151,27 +151,6 @@ type SQLInvoiceQueries interface { //nolint:interfacebloat var _ InvoiceDB = (*SQLStore)(nil) -// SQLInvoiceQueriesTxOptions defines the set of db txn options the -// SQLInvoiceQueries understands. -type SQLInvoiceQueriesTxOptions struct { - // readOnly governs if a read only transaction is needed or not. - readOnly bool -} - -// ReadOnly returns true if the transaction should be read only. -// -// NOTE: This implements the TxOptions. -func (a *SQLInvoiceQueriesTxOptions) ReadOnly() bool { - return a.readOnly -} - -// NewSQLInvoiceQueryReadTx creates a new read transaction option set. -func NewSQLInvoiceQueryReadTx() SQLInvoiceQueriesTxOptions { - return SQLInvoiceQueriesTxOptions{ - readOnly: true, - } -} - // BatchedSQLInvoiceQueries is a version of the SQLInvoiceQueries that's capable // of batched database operations. type BatchedSQLInvoiceQueries interface { @@ -304,7 +283,7 @@ func (i *SQLStore) AddInvoice(ctx context.Context, } var ( - writeTxOpts SQLInvoiceQueriesTxOptions + writeTxOpts = sqldb.WriteTxOpt() invoiceID int64 ) @@ -315,7 +294,7 @@ func (i *SQLStore) AddInvoice(ctx context.Context, return 0, err } - err = i.db.ExecTx(ctx, &writeTxOpts, func(db SQLInvoiceQueries) error { + err = i.db.ExecTx(ctx, writeTxOpts, func(db SQLInvoiceQueries) error { var err error invoiceID, err = db.InsertInvoice(ctx, insertInvoiceParams) if err != nil { @@ -718,8 +697,8 @@ func (i *SQLStore) LookupInvoice(ctx context.Context, err error ) - readTxOpt := NewSQLInvoiceQueryReadTx() - txErr := i.db.ExecTx(ctx, &readTxOpt, func(db SQLInvoiceQueries) error { + readTxOpt := sqldb.ReadTxOpt() + txErr := i.db.ExecTx(ctx, readTxOpt, func(db SQLInvoiceQueries) error { invoice, err = fetchInvoice(ctx, db, ref) return err @@ -739,8 +718,8 @@ func (i *SQLStore) FetchPendingInvoices(ctx context.Context) ( var invoices map[lntypes.Hash]Invoice - readTxOpt := NewSQLInvoiceQueryReadTx() - err := i.db.ExecTx(ctx, &readTxOpt, func(db SQLInvoiceQueries) error { + readTxOpt := sqldb.ReadTxOpt() + err := i.db.ExecTx(ctx, readTxOpt, func(db SQLInvoiceQueries) error { return queryWithLimit(func(offset int) (int, error) { params := sqlc.FilterInvoicesParams{ PendingOnly: true, @@ -800,8 +779,8 @@ func (i *SQLStore) InvoicesSettledSince(ctx context.Context, idx uint64) ( return invoices, nil } - readTxOpt := NewSQLInvoiceQueryReadTx() - err := i.db.ExecTx(ctx, &readTxOpt, func(db SQLInvoiceQueries) error { + readTxOpt := sqldb.ReadTxOpt() + err := i.db.ExecTx(ctx, readTxOpt, func(db SQLInvoiceQueries) error { err := queryWithLimit(func(offset int) (int, error) { params := sqlc.FilterInvoicesParams{ SettleIndexGet: sqldb.SQLInt64(idx + 1), @@ -946,8 +925,8 @@ func (i *SQLStore) InvoicesAddedSince(ctx context.Context, idx uint64) ( return result, nil } - readTxOpt := NewSQLInvoiceQueryReadTx() - err := i.db.ExecTx(ctx, &readTxOpt, func(db SQLInvoiceQueries) error { + readTxOpt := sqldb.ReadTxOpt() + err := i.db.ExecTx(ctx, readTxOpt, func(db SQLInvoiceQueries) error { return queryWithLimit(func(offset int) (int, error) { params := sqlc.FilterInvoicesParams{ AddIndexGet: sqldb.SQLInt64(idx + 1), @@ -1017,8 +996,8 @@ func (i *SQLStore) QueryInvoices(ctx context.Context, "be non-zero") } - readTxOpt := NewSQLInvoiceQueryReadTx() - err := i.db.ExecTx(ctx, &readTxOpt, func(db SQLInvoiceQueries) error { + readTxOpt := sqldb.ReadTxOpt() + err := i.db.ExecTx(ctx, readTxOpt, func(db SQLInvoiceQueries) error { return queryWithLimit(func(offset int) (int, error) { params := sqlc.FilterInvoicesParams{ NumOffset: int32(offset), @@ -1465,8 +1444,8 @@ func (i *SQLStore) UpdateInvoice(ctx context.Context, ref InvoiceRef, var updatedInvoice *Invoice - txOpt := SQLInvoiceQueriesTxOptions{readOnly: false} - txErr := i.db.ExecTx(ctx, &txOpt, func(db SQLInvoiceQueries) error { + txOpt := sqldb.WriteTxOpt() + txErr := i.db.ExecTx(ctx, txOpt, func(db SQLInvoiceQueries) error { switch { // For the default case we fetch all HTLCs. case setID == nil: @@ -1539,8 +1518,8 @@ func (i *SQLStore) DeleteInvoice(ctx context.Context, } } - var writeTxOpt SQLInvoiceQueriesTxOptions - err := i.db.ExecTx(ctx, &writeTxOpt, func(db SQLInvoiceQueries) error { + writeTxOpt := sqldb.WriteTxOpt() + err := i.db.ExecTx(ctx, writeTxOpt, func(db SQLInvoiceQueries) error { for _, ref := range invoicesToDelete { params := sqlc.DeleteInvoiceParams{ AddIndex: sqldb.SQLInt64(ref.AddIndex), @@ -1584,8 +1563,8 @@ func (i *SQLStore) DeleteInvoice(ctx context.Context, // DeleteCanceledInvoices removes all canceled invoices from the database. func (i *SQLStore) DeleteCanceledInvoices(ctx context.Context) error { - var writeTxOpt SQLInvoiceQueriesTxOptions - err := i.db.ExecTx(ctx, &writeTxOpt, func(db SQLInvoiceQueries) error { + writeTxOpt := sqldb.WriteTxOpt() + err := i.db.ExecTx(ctx, writeTxOpt, func(db SQLInvoiceQueries) error { _, err := db.DeleteCanceledInvoices(ctx) if err != nil { return fmt.Errorf("unable to delete canceled "+ diff --git a/sqldb/interfaces.go b/sqldb/interfaces.go index 6f3278c78..12ce63a1b 100644 --- a/sqldb/interfaces.go +++ b/sqldb/interfaces.go @@ -39,6 +39,35 @@ type TxOptions interface { ReadOnly() bool } +// txOptions is a concrete implementation of the TxOptions interface. +type txOptions struct { + // readOnly indicates if the transaction should be read-only. + readOnly bool +} + +// ReadOnly returns true if the transaction should be read only. +// +// NOTE: This is part of the TxOptions interface. +func (t *txOptions) ReadOnly() bool { + return t.readOnly +} + +// WriteTxOpt returns a TxOptions that indicates that the transaction +// should be a write transaction. +func WriteTxOpt() TxOptions { + return &txOptions{ + readOnly: false, + } +} + +// ReadTxOpt returns a TxOptions that indicates that the transaction +// should be a read-only transaction. +func ReadTxOpt() TxOptions { + return &txOptions{ + readOnly: true, + } +} + // BatchedTx is a generic interface that represents the ability to execute // several operations to a given storage interface in a single atomic // transaction. Typically, Q here will be some subset of the main sqlc.Querier diff --git a/sqldb/migrations.go b/sqldb/migrations.go index c14579c46..53d71e390 100644 --- a/sqldb/migrations.go +++ b/sqldb/migrations.go @@ -330,17 +330,6 @@ func (t *replacerFile) Close() error { return nil } -// MigrationTxOptions is the implementation of the TxOptions interface for -// migration transactions. -type MigrationTxOptions struct { -} - -// ReadOnly returns false to indicate that migration transactions are not read -// only. -func (m *MigrationTxOptions) ReadOnly() bool { - return false -} - // ApplyMigrations applies the provided migrations to the database in sequence. // It ensures migrations are executed in the correct order, applying both custom // migration functions and SQL migrations as needed. @@ -433,12 +422,12 @@ func ApplyMigrations(ctx context.Context, db *BaseDB, migration.SchemaVersion, err) } - var opts MigrationTxOptions + opts := WriteTxOpt() // Run the custom migration as a transaction to ensure // atomicity. If successful, mark the migration as complete in // the migration tracker table. - err = executor.ExecTx(ctx, &opts, func(tx *sqlc.Queries) error { + err = executor.ExecTx(ctx, opts, func(tx *sqlc.Queries) error { // Apply the migration function if one is provided. if migration.MigrationFn != nil { log.Infof("Applying custom migration '%v' "+