Merge pull request #780 from starius/sweepbatcher-dropbatch-writetx

sweepbatcher: use write tx in DropBatch

Author: starius

instantout/reservation/store.go

@@ -81,7 +81,7 @@ func (r *SQLStore) CreateReservation(ctx context.Context,
 		UpdateState:     string(reservation.State),
 	}
 
-	return r.baseDb.ExecTx(ctx, &loopdb.SqliteTxOptions{},
+	return r.baseDb.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
 		func(q *sqlc.Queries) error {
 			err := q.CreateReservation(ctx, args)
 			if err != nil {
@@ -121,7 +121,7 @@ func (r *SQLStore) UpdateReservation(ctx context.Context,
 		),
 	}
 
-	return r.baseDb.ExecTx(ctx, &loopdb.SqliteTxOptions{},
+	return r.baseDb.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
 		func(q *sqlc.Queries) error {
 			err := q.UpdateReservation(ctx, updateArgs)
 			if err != nil {

instantout/store.go

@@ -132,7 +132,7 @@ func (s *SQLStore) CreateInstantLoopOut(ctx context.Context,
 		UpdateState:     string(instantOut.State),
 	}
 
-	return s.baseDb.ExecTx(ctx, &loopdb.SqliteTxOptions{},
+	return s.baseDb.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
 		func(q *sqlc.Queries) error {
 			err := q.InsertSwap(ctx, swapArgs)
 			if err != nil {
@@ -204,7 +204,7 @@ func (s *SQLStore) UpdateInstantLoopOut(ctx context.Context,
 		UpdateState:     string(instantOut.State),
 	}
 
-	return s.baseDb.ExecTx(ctx, &loopdb.SqliteTxOptions{},
+	return s.baseDb.ExecTx(ctx, loopdb.NewSqlWriteOpts(),
 		func(q *sqlc.Queries) error {
 			err := q.UpdateInstantOut(ctx, updateParams)
 			if err != nil {

loopdb/sql_store.go

@@ -96,7 +96,7 @@ func (s *BaseDB) FetchLoopOutSwap(ctx context.Context,
 func (s *BaseDB) CreateLoopOut(ctx context.Context, hash lntypes.Hash,
 	swap *LoopOutContract) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return s.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		insertArgs := loopToInsertArgs(
 			hash, &swap.SwapContract,
@@ -134,7 +134,7 @@ func (s *BaseDB) CreateLoopOut(ctx context.Context, hash lntypes.Hash,
 func (s *BaseDB) BatchCreateLoopOut(ctx context.Context,
 	swaps map[lntypes.Hash]*LoopOutContract) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return s.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		for swapHash, swap := range swaps {
 			swap := swap
@@ -223,7 +223,7 @@ func (s *BaseDB) FetchLoopInSwaps(ctx context.Context) (
 func (s *BaseDB) CreateLoopIn(ctx context.Context, hash lntypes.Hash,
 	swap *LoopInContract) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return s.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		insertArgs := loopToInsertArgs(
 			hash, &swap.SwapContract,
@@ -260,7 +260,7 @@ func (s *BaseDB) CreateLoopIn(ctx context.Context, hash lntypes.Hash,
 func (s *BaseDB) BatchCreateLoopIn(ctx context.Context,
 	swaps map[lntypes.Hash]*LoopInContract) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return s.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		for swapHash, swap := range swaps {
 			swap := swap
@@ -351,7 +351,7 @@ var _ SwapStore = (*BaseDB)(nil)
 func (s *BaseDB) updateLoop(ctx context.Context, hash lntypes.Hash,
 	time time.Time, state SwapStateData) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return s.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		updateParams := sqlc.InsertSwapUpdateParams{
 			SwapHash:        hash[:],
@@ -379,7 +379,7 @@ func (s *BaseDB) updateLoop(ctx context.Context, hash lntypes.Hash,
 func (s *BaseDB) BatchInsertUpdate(ctx context.Context,
 	updateData map[lntypes.Hash][]BatchInsertUpdateData) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return s.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		for swapHash, updates := range updateData {
 			for _, update := range updates {
@@ -412,7 +412,7 @@ func (s *BaseDB) BatchInsertUpdate(ctx context.Context,
 func (b *BaseDB) BatchUpdateLoopOutSwapCosts(ctx context.Context,
 	costs map[lntypes.Hash]SwapCost) error {
 
-	writeOpts := &SqliteTxOptions{}
+	writeOpts := NewSqlWriteOpts()
 	return b.ExecTx(ctx, writeOpts, func(tx *sqlc.Queries) error {
 		for swapHash, cost := range costs {
 			lastUpdateID, err := tx.GetLastUpdateID(

loopdb/sqlite.go

@@ -323,14 +323,22 @@ type SqliteTxOptions struct {
 	readOnly bool
 }
 
-// NewSqlReadOpts returns a new KeyStoreTxOptions instance triggers a read
+// NewSqlReadOpts returns a new KeyStoreTxOptions instance that triggers a read
 // transaction.
 func NewSqlReadOpts() *SqliteTxOptions {
 	return &SqliteTxOptions{
 		readOnly: true,
 	}
 }
 
+// NewSqlWriteOpts returns a new KeyStoreTxOptions instance that triggers a
+// write (regular) transaction.
+func NewSqlWriteOpts() *SqliteTxOptions {
+	return &SqliteTxOptions{
+		readOnly: false,
+	}
+}
+
 // ReadOnly returns true if the transaction should be read only.
 //
 // NOTE: This implements the TxOptions interface.

sweepbatcher/store.go

@@ -111,7 +111,7 @@ func (s *SQLStore) InsertSweepBatch(ctx context.Context, batch *dbBatch) (int32,
 // DropBatch drops a batch from the database. Note that we only use this call
 // for batches that have no sweeps and so we'd not be able to resume.
 func (s *SQLStore) DropBatch(ctx context.Context, id int32) error {
-	readOpts := loopdb.NewSqlReadOpts()
+	readOpts := loopdb.NewSqlWriteOpts()
 	return s.baseDb.ExecTx(ctx, readOpts, func(tx *sqlc.Queries) error {
 		dbSweeps, err := tx.GetBatchSweeps(ctx, id)
 		if err != nil {