Merge pull request #9873 from ellemouton/sqldbHelpers

sqldb: re-usable TxOptions and NoOpReset

Author: guggero

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,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))

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,16 +171,13 @@ 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)
 
 		return err
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return nil, fmt.Errorf("unable to fetch node: %w", err)
 	}
@@ -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),
@@ -246,7 +221,7 @@ func (s *SQLStore) HasLightningNode(pubKey [33]byte) (time.Time, bool,
 		}
 
 		return nil
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return time.Time{}, false,
 			fmt.Errorf("unable to fetch node: %w", err)
@@ -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(),
@@ -281,7 +255,7 @@ func (s *SQLStore) AddrsForNode(nodePub *btcec.PublicKey) (bool, []net.Addr,
 		}
 
 		return nil
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return false, nil, fmt.Errorf("unable to get addresses for "+
 			"node(%x): %w", nodePub.SerializeCompressed(), err)
@@ -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),
@@ -321,7 +294,7 @@ func (s *SQLStore) DeleteLightningNode(pubKey route.Vertex) error {
 		}
 
 		return err
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return fmt.Errorf("unable to delete node: %w", err)
 	}
@@ -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),
@@ -370,7 +342,7 @@ func (s *SQLStore) LookupAlias(pub *btcec.PublicKey) (string, error) {
 		alias = dbNode.Alias.String
 
 		return nil
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return "", fmt.Errorf("unable to look up alias: %w", err)
 	}
@@ -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",
@@ -401,7 +370,7 @@ func (s *SQLStore) SourceNode() (*models.LightningNode, error) {
 		_, node, err = getNodeByPubKey(ctx, db, nodePub)
 
 		return err
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return nil, fmt.Errorf("unable to fetch source node: %w", err)
 	}
@@ -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",
@@ -442,7 +410,7 @@ func (s *SQLStore) SetSourceNode(node *models.LightningNode) error {
 		}
 
 		return db.AddSourceNode(ctx, id)
-	}, func() {})
+	}, sqldb.NoOpReset)
 }
 
 // NodeUpdatesInHorizon returns all the known lightning node which have an
@@ -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()),
@@ -482,7 +447,7 @@ func (s *SQLStore) NodeUpdatesInHorizon(startTime,
 		}
 
 		return nil
-	}, func() {})
+	}, sqldb.NoOpReset)
 	if err != nil {
 		return nil, fmt.Errorf("unable to fetch nodes: %w", err)
 	}

invoices/kv_sql_migration_test.go

@@ -70,14 +70,13 @@ 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,
 				)
-			}, func() {},
+			}, sqldb.NoOpReset,
 		)
 		require.NoError(t, err)
 

invoices/sql_migration_test.go

@@ -317,15 +317,15 @@ 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)
 		}
 
 		return nil
-	}, func() {})
+	}, sqldb.NoOpReset)
 	require.NoError(t, err)
 
 	// Fetch and compare each migrated invoice from the store with the