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

@@ -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 {

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,8 +578,7 @@ 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 := tx.ExecTx(ctx, opts, func(queries *sqlc.Queries) error {
 			_, err := queries.GetInvoiceByHash(ctx, hash[:])
 			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))

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 {

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),
@@ -347,10 +320,9 @@ func (s *SQLStore) FetchNodeFeatures(nodePub route.Vertex) (
 func (s *SQLStore) LookupAlias(pub *btcec.PublicKey) (string, error) {
 	var (
 		ctx   = context.TODO()
-		readTx = NewReadTx()
 		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()),

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,

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)

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 "+

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

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' "+