sweepbatcher: add an option to ignore HTLC txids

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

Changed SQLStore.DropBatch not to check that the batch being removed is empty.
It is not the case when removing a batch having only skipped sweeps.

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

Author: starius

client.go

@@ -12,6 +12,7 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/btcec/v2/schnorr"
 	"github.com/btcsuite/btcd/btcutil"
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/lightninglabs/aperture/l402"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/assets"
@@ -146,6 +147,10 @@ type ClientConfig struct {
 	// be attempted.
 	LoopOutMaxParts uint32
 
+	// SkippedTxs is the list of existing HTLC txids to skip when starting
+	// Loop. This should only be used if affected by the historical bug.
+	SkippedTxs []string
+
 	// TotalPaymentTimeout is the total amount of time until we time out
 	// off-chain payments (used in loop out).
 	TotalPaymentTimeout time.Duration
@@ -246,11 +251,7 @@ func NewClient(dbDir string, loopDB loopdb.SwapStore,
 		sweeper, loopDB, cfg.Lnd.ChainParams, getHeight,
 	)
 
-	batcher := sweepbatcher.NewBatcher(
-		cfg.Lnd.WalletKit, cfg.Lnd.ChainNotifier, cfg.Lnd.Signer,
-		swapServerClient.MultiMuSig2SignSweep, verifySchnorrSig,
-		cfg.Lnd.ChainParams, sweeperDb, sweepStore,
-
+	batcherOpts := []sweepbatcher.BatcherOption{
 		// Disable 100 sats/kw fee bump every block and retarget feerate
 		// every block according to the current mempool condition.
 		sweepbatcher.WithCustomFeeRate(
@@ -265,8 +266,29 @@ func NewClient(dbDir string, loopDB loopdb.SwapStore,
 		// delay time to sweepbatcher's handling. The delay used in
 		// loopout.go is repushDelay.
 		sweepbatcher.WithPublishDelay(
-			repushDelay+additionalDelay,
+			repushDelay + additionalDelay,
 		),
+	}
+
+	if len(cfg.SkippedTxs) != 0 {
+		skippedTxs := make(map[chainhash.Hash]struct{})
+		for _, txid := range cfg.SkippedTxs {
+			txid, err := chainhash.NewHashFromStr(txid)
+			if err != nil {
+				return nil, nil, fmt.Errorf("failed to parse "+
+					"txid to skip %v: %w", txid, err)
+			}
+			skippedTxs[*txid] = struct{}{}
+		}
+		batcherOpts = append(batcherOpts, sweepbatcher.WithSkippedTxs(
+			skippedTxs,
+		))
+	}
+
+	batcher := sweepbatcher.NewBatcher(
+		cfg.Lnd.WalletKit, cfg.Lnd.ChainNotifier, cfg.Lnd.Signer,
+		swapServerClient.MultiMuSig2SignSweep, verifySchnorrSig,
+		cfg.Lnd.ChainParams, sweeperDb, sweepStore, batcherOpts...,
 	)
 
 	executor = newExecutor(&executorConfig{

loopd/config.go

@@ -184,6 +184,8 @@ type Config struct {
 
 	LoopOutMaxParts uint32 `long:"loopoutmaxparts" description:"The maximum number of payment parts that may be used for a loop out swap."`
 
+	SkippedTxs []string `long:"skippedtxs" description:"The list of existing HTLC txids to skip when starting Loop. This should only be used if affected by the historical bug."`
+
 	TotalPaymentTimeout time.Duration `long:"totalpaymenttimeout" description:"The timeout to use for off-chain payments."`
 	MaxPaymentRetries   int           `long:"maxpaymentretries" description:"The maximum number of times an off-chain payment may be retried."`
 

loopd/utils.go

@@ -50,6 +50,7 @@ func getClient(cfg *Config, swapDb loopdb.SwapStore,
 		MaxL402Cost:                          btcutil.Amount(cfg.MaxL402Cost),
 		MaxL402Fee:                           btcutil.Amount(cfg.MaxL402Fee),
 		LoopOutMaxParts:                      cfg.LoopOutMaxParts,
+		SkippedTxs:                           cfg.SkippedTxs,
 		TotalPaymentTimeout:                  cfg.TotalPaymentTimeout,
 		MaxPaymentRetries:                    cfg.MaxPaymentRetries,
 		MaxStaticAddrHtlcFeePercentage:       cfg.MaxStaticAddrHtlcFeePercentage,

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.skippedTxs, 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/store.go

@@ -3,7 +3,6 @@ package sweepbatcher
 import (
 	"context"
 	"database/sql"
-	"fmt"
 
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg"
@@ -116,19 +115,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.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)
-	})
+	return s.baseDb.DropBatch(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,13 +53,8 @@ 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

sweepbatcher/sweep_batch.go

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

sweepbatcher/sweep_batcher.go

@@ -440,6 +440,10 @@ type Batcher struct {
 	// presignedHelper provides methods used when presigned batches are
 	// enabled.
 	presignedHelper PresignedHelper
+
+	// skippedTxs is the list of previous transactions to ignore when
+	// loading the sweeps from DB. This is needed to fix a historical bug.
+	skippedTxs map[chainhash.Hash]struct{}
 }
 
 // BatcherConfig holds batcher configuration.
@@ -484,6 +488,10 @@ type BatcherConfig struct {
 	// presignedHelper provides methods used when presigned batches are
 	// enabled.
 	presignedHelper PresignedHelper
+
+	// skippedTxs is the list of previous transactions to ignore when
+	// loading the sweeps from DB. This is needed to fix a historical bug.
+	skippedTxs map[chainhash.Hash]struct{}
 }
 
 // BatcherOption configures batcher behaviour.
@@ -566,6 +574,14 @@ func WithPresignedHelper(presignedHelper PresignedHelper) BatcherOption {
 	}
 }
 
+// WithSkippedTxs is the list of previous transactions to ignore when
+// loading the sweeps from DB. This is needed to fix a historical bug.
+func WithSkippedTxs(skippedTxs map[chainhash.Hash]struct{}) BatcherOption {
+	return func(cfg *BatcherConfig) {
+		cfg.skippedTxs = skippedTxs
+	}
+}
+
 // NewBatcher creates a new Batcher instance.
 func NewBatcher(wallet lndclient.WalletKitClient,
 	chainNotifier lndclient.ChainNotifierClient,
@@ -574,6 +590,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 +607,10 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 		// publishing error. By default, it logs all errors as warnings,
 		// but "insufficient fee" as Info.
 		publishErrorHandler: defaultPublishErrorLogger,
+
+		skippedTxs: map[chainhash.Hash]struct{}{
+			*badTx1: {},
+		},
 	}
 	for _, opt := range opts {
 		opt(&cfg)
@@ -621,6 +649,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 		customMuSig2Signer:   cfg.customMuSig2Signer,
 		publishErrorHandler:  cfg.publishErrorHandler,
 		presignedHelper:      cfg.presignedHelper,
+		skippedTxs:           cfg.skippedTxs,
 	}
 }
 
@@ -1000,13 +1029,30 @@ func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
 	return batch, nil
 }
 
+// filterDbSweeps copies dbSweeps, skipping the sweeps from skipped txs.
+func filterDbSweeps(skippedTxs map[chainhash.Hash]struct{},
+	dbSweeps []*dbSweep) []*dbSweep {
+
+	result := make([]*dbSweep, 0, len(dbSweeps))
+	for _, dbSweep := range dbSweeps {
+		if _, has := skippedTxs[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 {
 	dbSweeps, err := b.store.FetchBatchSweeps(ctx, batch.id)
 	if err != nil {
 		return err
 	}
+	dbSweeps = filterDbSweeps(b.skippedTxs, dbSweeps)
 
 	if len(dbSweeps) == 0 {
 		infof("skipping restored batch %d as it has no sweeps",
@@ -1502,6 +1548,7 @@ func (b *Batcher) newBatchConfig(maxTimeoutDistance int32) batchConfig {
 		txLabeler:          b.txLabeler,
 		customMuSig2Signer: b.customMuSig2Signer,
 		presignedHelper:    b.presignedHelper,
+		skippedTxs:         b.skippedTxs,
 		clock:              b.clock,
 		chainParams:        b.chainParams,
 	}