Merge pull request #783 from starius/sweepbatcher-WithNoBumping

sweepbatcher: add option WithNoBumping

sweepbatcher/sweep_batch.go

@@ -127,6 +127,12 @@ type batchConfig struct {
 	// batchPublishDelay is the delay between receiving a new block and
 	// publishing the batch transaction.
 	batchPublishDelay time.Duration
+
+	// noBumping instructs sweepbatcher not to fee bump itself and rely on
+	// external source of fee rates (MinFeeRate). To change the fee rate,
+	// the caller has to update it in the source of SweepInfo (interface
+	// SweepFetcher) and re-add the sweep by calling AddSweep.
+	noBumping bool
 }
 
 // rbfCache stores data related to our last fee bump.
@@ -1079,7 +1085,7 @@ func (b *batch) updateRbfRate(ctx context.Context) error {
 
 		// Set the initial value for our fee rate.
 		b.rbfCache.FeeRate = rate
-	} else {
+	} else if !b.cfg.noBumping {
 		// Bump the fee rate by the configured step.
 		b.rbfCache.FeeRate += defaultFeeRateStep
 	}

sweepbatcher/sweep_batcher.go

@@ -235,6 +235,34 @@ type Batcher struct {
 	// wg is a waitgroup that is used to wait for all the goroutines to
 	// exit.
 	wg sync.WaitGroup
+
+	// noBumping instructs sweepbatcher not to fee bump itself and rely on
+	// external source of fee rates (MinFeeRate). To change the fee rate,
+	// the caller has to update it in the source of SweepInfo (interface
+	// SweepFetcher) and re-add the sweep by calling AddSweep.
+	noBumping bool
+}
+
+// BatcherConfig holds batcher configuration.
+type BatcherConfig struct {
+	// noBumping instructs sweepbatcher not to fee bump itself and rely on
+	// external source of fee rates (MinFeeRate). To change the fee rate,
+	// the caller has to update it in the source of SweepInfo (interface
+	// SweepFetcher) and re-add the sweep by calling AddSweep.
+	noBumping bool
+}
+
+// BatcherOption configures batcher behaviour.
+type BatcherOption func(*BatcherConfig)
+
+// WithNoBumping instructs sweepbatcher not to fee bump itself and
+// rely on external source of fee rates (MinFeeRate). To change the
+// fee rate, the caller has to update it in the source of SweepInfo
+// (interface SweepFetcher) and re-add the sweep by calling AddSweep.
+func WithNoBumping() BatcherOption {
+	return func(cfg *BatcherConfig) {
+		cfg.noBumping = true
+	}
 }
 
 // NewBatcher creates a new Batcher instance.
@@ -242,7 +270,13 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 	chainNotifier lndclient.ChainNotifierClient,
 	signerClient lndclient.SignerClient, musig2ServerSigner MuSig2SignSweep,
 	verifySchnorrSig VerifySchnorrSig, chainparams *chaincfg.Params,
-	store BatcherStore, sweepStore SweepFetcher) *Batcher {
+	store BatcherStore, sweepStore SweepFetcher,
+	opts ...BatcherOption) *Batcher {
+
+	var cfg BatcherConfig
+	for _, opt := range opts {
+		opt(&cfg)
+	}
 
 	return &Batcher{
 		batches:          make(map[int32]*batch),
@@ -258,6 +292,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 		chainParams:      chainparams,
 		store:            store,
 		sweepStore:       sweepStore,
+		noBumping:        cfg.noBumping,
 	}
 }
 
@@ -416,6 +451,7 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
 func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
 	cfg := batchConfig{
 		maxTimeoutDistance: defaultMaxTimeoutDistance,
+		noBumping:          b.noBumping,
 	}
 
 	switch b.chainParams {
@@ -525,6 +561,7 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
 
 	cfg := batchConfig{
 		maxTimeoutDistance: batch.cfg.maxTimeoutDistance,
+		noBumping:          b.noBumping,
 	}
 
 	newBatch, err := NewBatchFromDB(cfg, batchKit)
@@ -587,6 +624,7 @@ func (b *Batcher) FetchUnconfirmedBatches(ctx context.Context) ([]*batch,
 
 		bchCfg := batchConfig{
 			maxTimeoutDistance: bch.MaxTimeoutDistance,
+			noBumping:          b.noBumping,
 		}
 		batch.cfg = &bchCfg
 

sweepbatcher/sweep_batcher_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
@@ -15,6 +16,7 @@ import (
 	"github.com/lightninglabs/loop/test"
 	"github.com/lightninglabs/loop/utils"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/keychain"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/stretchr/testify/require"
 )
@@ -42,6 +44,20 @@ var destAddr = func() btcutil.Address {
 	return addr
 }()
 
+var senderKey, receiverKey [33]byte
+
+func init() {
+	// Generate keys.
+	_, senderPubKey := test.CreateKey(1)
+	copy(senderKey[:], senderPubKey.SerializeCompressed())
+	_, receiverPubKey := test.CreateKey(2)
+	copy(receiverKey[:], receiverPubKey.SerializeCompressed())
+}
+
+func testVerifySchnorrSig(pubKey *btcec.PublicKey, hash, sig []byte) error {
+	return nil
+}
+
 func testMuSig2SignSweep(ctx context.Context,
 	protocolVersion loopdb.ProtocolVersion, swapHash lntypes.Hash,
 	paymentAddr [32]byte, nonce []byte, sweepTxPsbt []byte,
@@ -245,6 +261,103 @@ func testSweepBatcherBatchCreation(t *testing.T, store testStore,
 	require.True(t, batcherStore.AssertSweepStored(sweepReq3.SwapHash))
 }
 
+// testFeeBumping tests that sweep is RBFed with slightly higher fee rate after
+// each block unless WithNoBumping is passed.
+func testFeeBumping(t *testing.T, store testStore,
+	batcherStore testBatcherStore, noFeeBumping bool) {
+
+	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)
+
+	// Disable fee bumping, if requested.
+	var opts []BatcherOption
+	if noFeeBumping {
+		opts = append(opts, WithNoBumping())
+	}
+
+	batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore, opts...)
+	go func() {
+		err := batcher.Run(ctx)
+		checkBatcherError(t, err)
+	}()
+
+	// Create a sweep request.
+	sweepReq1 := SweepRequest{
+		SwapHash: lntypes.Hash{1, 1, 1},
+		Value:    1_000_000,
+		Outpoint: wire.OutPoint{
+			Hash:  chainhash.Hash{1, 1},
+			Index: 1,
+		},
+		Notifier: &SpendNotifier{
+			SpendChan:    make(chan *SpendDetail, ntfnBufferSize),
+			SpendErrChan: make(chan error, ntfnBufferSize),
+			QuitChan:     make(chan bool, ntfnBufferSize),
+		},
+	}
+
+	swap1 := &loopdb.LoopOutContract{
+		SwapContract: loopdb.SwapContract{
+			CltvExpiry:      111,
+			AmountRequested: 1_000_000,
+			ProtocolVersion: loopdb.ProtocolVersionMuSig2,
+			HtlcKeys: loopdb.HtlcKeys{
+				SenderScriptKey:        senderKey,
+				ReceiverScriptKey:      receiverKey,
+				SenderInternalPubKey:   senderKey,
+				ReceiverInternalPubKey: receiverKey,
+				ClientScriptKeyLocator: keychain.KeyLocator{
+					Family: 1,
+					Index:  2,
+				},
+			},
+		},
+
+		DestAddr:        destAddr,
+		SwapInvoice:     swapInvoice,
+		SweepConfTarget: 111,
+	}
+
+	err = store.CreateLoopOut(ctx, sweepReq1.SwapHash, swap1)
+	require.NoError(t, err)
+	store.AssertLoopOutStored()
+
+	// Deliver sweep request to batcher.
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
+
+	// Since a batch was created we check that it registered for its primary
+	// sweep's spend.
+	<-lnd.RegisterSpendChannel
+
+	// Wait for tx to be published.
+	tx1 := <-lnd.TxPublishChannel
+	out1 := tx1.TxOut[0].Value
+
+	// Tick tock next block.
+	err = lnd.NotifyHeight(601)
+	require.NoError(t, err)
+
+	// Wait for another sweep tx to be published.
+	tx2 := <-lnd.TxPublishChannel
+	out2 := tx2.TxOut[0].Value
+
+	if noFeeBumping {
+		// Expect output to stay the same.
+		require.Equal(t, out1, out2, "expected out to stay the same")
+	} else {
+		// Expect output to drop.
+		require.Greater(t, out1, out2, "expected out to drop")
+	}
+}
+
 // testSweepBatcherSimpleLifecycle tests the simple lifecycle of the batches
 // that are created and run by the batcher.
 func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
@@ -1837,6 +1950,26 @@ func TestSweepBatcherBatchCreation(t *testing.T) {
 	runTests(t, testSweepBatcherBatchCreation)
 }
 
+// TestFeeBumping tests that sweep is RBFed with slightly higher fee rate after
+// each block unless WithNoBumping is passed.
+func TestFeeBumping(t *testing.T) {
+	t.Run("regular", func(t *testing.T) {
+		runTests(t, func(t *testing.T, store testStore,
+			batcherStore testBatcherStore) {
+
+			testFeeBumping(t, store, batcherStore, false)
+		})
+	})
+
+	t.Run("WithNoBumping", func(t *testing.T) {
+		runTests(t, func(t *testing.T, store testStore,
+			batcherStore testBatcherStore) {
+
+			testFeeBumping(t, store, batcherStore, true)
+		})
+	})
+}
+
 // TestSweepBatcherSimpleLifecycle tests the simple lifecycle of the batches
 // that are created and run by the batcher.
 func TestSweepBatcherSimpleLifecycle(t *testing.T) {