sweepbatcher: add an option to ignore HTLC txids

Added option WithSkippedTxns, which has one historical problematic tx by default.
Sweeps originating from these transactions are omitted when reading from DB.

loopdb: add column sweep_batches.cancelled and replaced DropBatch with
CancelBatch. It is needed, because sweep.batch_id is a foreign key to batch.

Changed StoreMock.InsertSweepBatch not to reuse batch_id. This is needed by
the test, which checks that new batch has fresh ID.

sweepbatcher/presigned.go

@@ -126,6 +126,8 @@ func (b *batch) getOrderedSweeps(ctx context.Context) ([]sweep, error) {
 		return nil, fmt.Errorf("FetchBatchSweeps(%d) failed: %w", b.id,
 			err)
 	}
+	dbSweeps = filterDbSweeps(b.cfg.skippedTxns, dbSweeps)
+
 	if len(dbSweeps) != len(utxo2sweep) {
 		return nil, fmt.Errorf("FetchBatchSweeps(%d) returned %d "+
 			"sweeps, len(b.sweeps) is %d", b.id, len(dbSweeps),

sweepbatcher/presigned_test.go

@@ -33,8 +33,9 @@ func TestOrderedSweeps(t *testing.T) {
 	ctx := context.Background()
 
 	cases := []struct {
-		name   string
-		sweeps []sweep
+		name        string
+		sweeps      []sweep
+		skippedTxns map[chainhash.Hash]struct{}
 
 		// Testing errors.
 		skipStore     bool
@@ -69,6 +70,20 @@ func TestOrderedSweeps(t *testing.T) {
 			},
 		},
 
+		{
+			name: "one sweep, skipped",
+			sweeps: []sweep{
+				{
+					outpoint: op1,
+					swapHash: swapHash1,
+				},
+			},
+			skippedTxns: map[chainhash.Hash]struct{}{
+				op1.Hash: {},
+			},
+			wantGroups: [][]sweep{},
+		},
+
 		{
 			name: "two sweeps, one swap",
 			sweeps: []sweep{
@@ -95,6 +110,31 @@ func TestOrderedSweeps(t *testing.T) {
 			},
 		},
 
+		{
+			name: "two sweeps, one swap, one skipped",
+			sweeps: []sweep{
+				{
+					outpoint: op2,
+					swapHash: swapHash1,
+				},
+				{
+					outpoint: op1,
+					swapHash: swapHash1,
+				},
+			},
+			skippedTxns: map[chainhash.Hash]struct{}{
+				op1.Hash: {},
+			},
+			wantGroups: [][]sweep{
+				{
+					{
+						outpoint: op2,
+						swapHash: swapHash1,
+					},
+				},
+			},
+		},
+
 		{
 			name: "two sweeps, two swap",
 			sweeps: []sweep{
@@ -266,6 +306,9 @@ func TestOrderedSweeps(t *testing.T) {
 			b := &batch{
 				sweeps: m,
 				store:  NewStoreMock(),
+				cfg: &batchConfig{
+					skippedTxns: tc.skippedTxns,
+				},
 			}
 
 			// Store the sweeps in mock store.
@@ -299,6 +342,14 @@ func TestOrderedSweeps(t *testing.T) {
 				m[added.outpoint] = added
 			}
 
+			// Remove skipped sweeps from the batch to make it
+			// match with what is read from DB after filtering.
+			for op := range m {
+				if _, has := tc.skippedTxns[op.Hash]; has {
+					delete(m, op)
+				}
+			}
+
 			// Now run the tested functions.
 			orderedSweeps, err := b.getOrderedSweeps(ctx)
 			if tc.wantErr1 != "" {
@@ -313,7 +364,15 @@ func TestOrderedSweeps(t *testing.T) {
 			}
 
 			// The wanted list of sweeps matches the input order.
-			require.Equal(t, tc.sweeps, orderedSweeps)
+			notSkipped := make([]sweep, 0, len(tc.sweeps))
+			for _, s := range tc.sweeps {
+				_, has := tc.skippedTxns[s.outpoint.Hash]
+				if has {
+					continue
+				}
+				notSkipped = append(notSkipped, s)
+			}
+			require.Equal(t, notSkipped, orderedSweeps)
 
 			groups, err := b.getSweepsGroups(ctx)
 			if tc.wantErr2 != "" {

sweepbatcher/store.go

@@ -3,7 +3,6 @@ package sweepbatcher
 import (
 	"context"
 	"database/sql"
-	"fmt"
 
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg"
@@ -44,8 +43,8 @@ type Querier interface {
 	InsertBatch(ctx context.Context, arg sqlc.InsertBatchParams) (
 		int32, error)
 
-	// DropBatch drops a batch from the database.
-	DropBatch(ctx context.Context, id int32) error
+	// CancelBatch marks the batch as cancelled.
+	CancelBatch(ctx context.Context, id int32) error
 
 	// UpdateBatch updates a batch in the database.
 	UpdateBatch(ctx context.Context, arg sqlc.UpdateBatchParams) error
@@ -113,22 +112,11 @@ func (s *SQLStore) InsertSweepBatch(ctx context.Context, batch *dbBatch) (int32,
 	return s.baseDb.InsertBatch(ctx, batchToInsertArgs(*batch))
 }
 
-// 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.NewSqlWriteOpts()
-	return s.baseDb.ExecTx(ctx, readOpts, func(tx Querier) error {
-		dbSweeps, err := tx.GetBatchSweeps(ctx, id)
-		if err != nil {
-			return err
-		}
-
-		if len(dbSweeps) != 0 {
-			return fmt.Errorf("cannot drop a non-empty batch")
-		}
-
-		return tx.DropBatch(ctx, id)
-	})
+// CancelBatch marks a batch as cancelled in the database. Note that we only use
+// this call for batches that have no sweeps or all the sweeps are in skipped
+// transaction and so we'd not be able to resume.
+func (s *SQLStore) CancelBatch(ctx context.Context, id int32) error {
+	return s.baseDb.CancelBatch(ctx, id)
 }
 
 // UpdateSweepBatch updates a batch in the database.

sweepbatcher/store_mock.go

@@ -16,6 +16,7 @@ type StoreMock struct {
 	sweeps  map[wire.OutPoint]dbSweep
 	mu      sync.Mutex
 	sweepID int32
+	batchID int32
 }
 
 // NewStoreMock instantiates a new mock store.
@@ -52,20 +53,15 @@ func (s *StoreMock) InsertSweepBatch(ctx context.Context,
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	var id int32
-
-	if len(s.batches) == 0 {
-		id = 0
-	} else {
-		id = int32(len(s.batches))
-	}
+	id := s.batchID
+	s.batchID++
 
 	s.batches[id] = *batch
 	return id, nil
 }
 
-// DropBatch drops a batch from the database.
-func (s *StoreMock) DropBatch(ctx context.Context, id int32) error {
+// CancelBatch drops a batch from the database.
+func (s *StoreMock) CancelBatch(ctx context.Context, id int32) error {
 	delete(s.batches, id)
 	return nil
 }

sweepbatcher/sweep_batch.go

@@ -187,6 +187,10 @@ type batchConfig struct {
 	// enabled.
 	presignedHelper PresignedHelper
 
+	// skippedTxns is the list of previous transactions to ignore when
+	// loading the sweeps from DB. This is needed to fix a historical bug.
+	skippedTxns map[chainhash.Hash]struct{}
+
 	// chainParams are the chain parameters of the chain that is used by
 	// batches.
 	chainParams *chaincfg.Params

sweepbatcher/sweep_batcher.go

@@ -50,9 +50,10 @@ type BatcherStore interface {
 	// of the inserted batch.
 	InsertSweepBatch(ctx context.Context, batch *dbBatch) (int32, error)
 
-	// DropBatch drops a batch from the database. This should only be used
-	// when a batch is empty.
-	DropBatch(ctx context.Context, id int32) error
+	// CancelBatch marks a batch as cancelled in the database. Note that we
+	// only use this call for batches that have no sweeps or all the sweeps
+	// are in skipped transaction and so we'd not be able to resume.
+	CancelBatch(ctx context.Context, id int32) error
 
 	// UpdateSweepBatch updates a batch in the database.
 	UpdateSweepBatch(ctx context.Context, batch *dbBatch) error
@@ -440,6 +441,10 @@ type Batcher struct {
 	// presignedHelper provides methods used when presigned batches are
 	// enabled.
 	presignedHelper PresignedHelper
+
+	// skippedTxns is the list of previous transactions to ignore when
+	// loading the sweeps from DB. This is needed to fix a historical bug.
+	skippedTxns map[chainhash.Hash]struct{}
 }
 
 // BatcherConfig holds batcher configuration.
@@ -484,6 +489,10 @@ type BatcherConfig struct {
 	// presignedHelper provides methods used when presigned batches are
 	// enabled.
 	presignedHelper PresignedHelper
+
+	// skippedTxns is the list of previous transactions to ignore when
+	// loading the sweeps from DB. This is needed to fix a historical bug.
+	skippedTxns map[chainhash.Hash]struct{}
 }
 
 // BatcherOption configures batcher behaviour.
@@ -566,6 +575,14 @@ func WithPresignedHelper(presignedHelper PresignedHelper) BatcherOption {
 	}
 }
 
+// WithSkippedTxns is the list of previous transactions to ignore when
+// loading the sweeps from DB. This is needed to fix a historical bug.
+func WithSkippedTxns(skippedTxns map[chainhash.Hash]struct{}) BatcherOption {
+	return func(cfg *BatcherConfig) {
+		cfg.skippedTxns = skippedTxns
+	}
+}
+
 // NewBatcher creates a new Batcher instance.
 func NewBatcher(wallet lndclient.WalletKitClient,
 	chainNotifier lndclient.ChainNotifierClient,
@@ -574,6 +591,14 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 	store BatcherStore, sweepStore SweepFetcher,
 	opts ...BatcherOption) *Batcher {
 
+	badTx1, err := chainhash.NewHashFromStr(
+		"7028bdac753a254785d29506f311abcda323706b531345105f38999" +
+			"aecd6f3d1",
+	)
+	if err != nil {
+		panic(err)
+	}
+
 	cfg := BatcherConfig{
 		// By default, loop/labels.LoopOutBatchSweepSuccess is used
 		// to label sweep transactions.
@@ -583,6 +608,10 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 		// publishing error. By default, it logs all errors as warnings,
 		// but "insufficient fee" as Info.
 		publishErrorHandler: defaultPublishErrorLogger,
+
+		skippedTxns: map[chainhash.Hash]struct{}{
+			*badTx1: {},
+		},
 	}
 	for _, opt := range opts {
 		opt(&cfg)
@@ -621,6 +650,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 		customMuSig2Signer:   cfg.customMuSig2Signer,
 		publishErrorHandler:  cfg.publishErrorHandler,
 		presignedHelper:      cfg.presignedHelper,
+		skippedTxns:          cfg.skippedTxns,
 	}
 }
 
@@ -1000,6 +1030,22 @@ func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
 	return batch, nil
 }
 
+// filterDbSweeps copies dbSweeps, skipping the sweeps from skipped txs.
+func filterDbSweeps(skippedTxns map[chainhash.Hash]struct{},
+	dbSweeps []*dbSweep) []*dbSweep {
+
+	result := make([]*dbSweep, 0, len(dbSweeps))
+	for _, dbSweep := range dbSweeps {
+		if _, has := skippedTxns[dbSweep.Outpoint.Hash]; has {
+			continue
+		}
+
+		result = append(result, dbSweep)
+	}
+
+	return result
+}
+
 // spinUpBatchFromDB spins up a batch that already existed in storage, then
 // returns it.
 func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
@@ -1007,13 +1053,15 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
 	if err != nil {
 		return err
 	}
+	dbSweeps = filterDbSweeps(b.skippedTxns, dbSweeps)
 
 	if len(dbSweeps) == 0 {
 		infof("skipping restored batch %d as it has no sweeps",
 			batch.id)
 
-		// It is safe to drop this empty batch as it has no sweeps.
-		err := b.store.DropBatch(ctx, batch.id)
+		// It is safe to cancel this empty batch as it has no sweeps
+		// that are not skipped.
+		err := b.store.CancelBatch(ctx, batch.id)
 		if err != nil {
 			warnf("unable to drop empty batch %d: %v",
 				batch.id, err)
@@ -1502,6 +1550,7 @@ func (b *Batcher) newBatchConfig(maxTimeoutDistance int32) batchConfig {
 		txLabeler:          b.txLabeler,
 		customMuSig2Signer: b.customMuSig2Signer,
 		presignedHelper:    b.presignedHelper,
+		skippedTxns:        b.skippedTxns,
 		clock:              b.clock,
 		chainParams:        b.chainParams,
 	}

sweepbatcher/sweep_batcher_test.go

@@ -1172,6 +1172,159 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	require.ErrorIs(t, runErr, testError)
 }
 
+// testSweepBatcherSkippedTxns tests that option WithSkippedTxns
+// works as expected.
+func testSweepBatcherSkippedTxns(t *testing.T, store testStore,
+	batcherStore testBatcherStore) {
+
+	defer test.Guard(t)()
+
+	lnd := test.NewMockLnd()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	sweepStore, err := NewSweepFetcherFromSwapStore(store, lnd.ChainParams)
+	require.NoError(t, err)
+
+	batcher := NewBatcher(
+		lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore,
+	)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	var runErr error
+	go func() {
+		defer wg.Done()
+		runErr = batcher.Run(ctx)
+	}()
+	// Wait for the batcher to be initialized.
+	<-batcher.initDone
+
+	// Create a sweep request.
+	op1 := wire.OutPoint{
+		Hash:  chainhash.Hash{1, 1},
+		Index: 1,
+	}
+	swapHash := lntypes.Hash{1, 1, 1}
+	const (
+		inputValue       = 111
+		initiationHeight = 550
+	)
+
+	swap1 := &loopdb.LoopOutContract{
+		SwapContract: loopdb.SwapContract{
+			CltvExpiry:       111,
+			AmountRequested:  inputValue,
+			ProtocolVersion:  loopdb.ProtocolVersionMuSig2,
+			HtlcKeys:         htlcKeys,
+			InitiationHeight: initiationHeight,
+		},
+
+		DestAddr:        destAddr,
+		SwapInvoice:     swapInvoice,
+		SweepConfTarget: 111,
+	}
+
+	err = store.CreateLoopOut(ctx, swapHash, swap1)
+	require.NoError(t, err)
+	store.AssertLoopOutStored()
+
+	// Deliver sweep request to batcher.
+	require.NoError(t, batcher.AddSweep(ctx, &SweepRequest{
+		SwapHash: swapHash,
+		Inputs: []Input{{
+			Value:    inputValue,
+			Outpoint: op1,
+		}},
+		Notifier: &dummyNotifier,
+	}))
+
+	// When batch is successfully created it will execute it's first step,
+	// which leads to a spend monitor of the primary sweep.
+	<-lnd.RegisterSpendChannel
+
+	// Wait for tx to be published.
+	<-lnd.TxPublishChannel
+
+	// Record batch ID.
+	var oldBatchID int32
+	require.Eventually(t, func() bool {
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		oldBatchID = batch.id
+
+		return true
+	}, test.Timeout, eventuallyCheckFrequency)
+
+	// Restart the batcher, adding the oldBatchID to skipped batches.
+	cancel()
+	wg.Wait()
+	checkBatcherError(t, runErr)
+
+	ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+
+	batcher = NewBatcher(
+		lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore,
+		WithSkippedTxns(map[chainhash.Hash]struct{}{
+			op1.Hash: {},
+		}),
+	)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		runErr = batcher.Run(ctx)
+	}()
+	// Wait for the batcher to be initialized.
+	<-batcher.initDone
+
+	// Add the same swap with another outpoint.
+	op2 := wire.OutPoint{
+		Hash:  chainhash.Hash{2, 2},
+		Index: 2,
+	}
+	require.NoError(t, batcher.AddSweep(ctx, &SweepRequest{
+		SwapHash: swapHash,
+		Inputs: []Input{{
+			Value:    inputValue,
+			Outpoint: op2,
+		}},
+		Notifier: &dummyNotifier,
+	}))
+
+	// Make sure it is launched in a new batch.
+	<-lnd.RegisterSpendChannel
+
+	// Wait for tx to be published.
+	tx := <-lnd.TxPublishChannel
+	require.Len(t, tx.TxIn, 1)
+
+	// Record new batch ID.
+	var newBatchID int32
+	require.Eventually(t, func() bool {
+		batch := tryGetOnlyBatch(ctx, batcher)
+		if batch == nil {
+			return false
+		}
+		newBatchID = batch.id
+
+		return true
+	}, test.Timeout, eventuallyCheckFrequency)
+
+	// Make sure it is another batch.
+	require.NotEqual(t, oldBatchID, newBatchID)
+
+	// Stop the batcher.
+	cancel()
+	wg.Wait()
+	checkBatcherError(t, runErr)
+}
+
 // wrappedLogger implements btclog.Logger, recording last debug message format.
 // It is needed to watch for messages in tests.
 type wrappedLogger struct {
@@ -4745,6 +4898,12 @@ func TestSweepBatcherSimpleLifecycle(t *testing.T) {
 	runTests(t, testSweepBatcherSimpleLifecycle)
 }
 
+// TestSweepBatcherSkippedTxns tests that option WithSkippedTxns
+// works as expected.
+func TestSweepBatcherSkippedTxns(t *testing.T) {
+	runTests(t, testSweepBatcherSkippedTxns)
+}
+
 // TestDelays tests that WithInitialDelay and WithPublishDelay work.
 func TestDelays(t *testing.T) {
 	runTests(t, testDelays)