sweepbatcher: update feerate outside AddSweep

AddSweep may not be called after getting the first confirmation, but feerate
updates are still needed in case of reorg.

Update test TestFeeRateGrows not to call AddSweep again and make sure feerate
is updated itself.

Author: starius

sweepbatcher/sweep_batch.go

@@ -169,9 +169,10 @@ type batchConfig struct {
 	// initial delay completion and publishing the batch transaction.
 	batchPublishDelay time.Duration
 
-	// noBumping instructs sweepbatcher not to fee bump itself and rely on
-	// external source of fee rates (FeeRateProvider).
-	noBumping bool
+	// customFeeRate provides custom min fee rate per swap. The batch uses
+	// max of the fee rates of its swaps. In this mode confTarget is
+	// ignored and fee bumping by sweepbatcher is disabled.
+	customFeeRate FeeRateProvider
 
 	// txLabeler is a function generating a transaction label. It is called
 	// before publishing a batch transaction. Batch ID is passed to it.
@@ -723,6 +724,9 @@ func (b *batch) addSweeps(ctx context.Context, sweeps []*sweep) (bool, error) {
 			// lower that minFeeRate of other sweeps (so it is
 			// applied).
 			if b.rbfCache.FeeRate < s.minFeeRate {
+				b.Infof("Increasing feerate of the batch "+
+					"from %v to %v", b.rbfCache.FeeRate,
+					s.minFeeRate)
 				b.rbfCache.FeeRate = s.minFeeRate
 			}
 		}
@@ -769,6 +773,9 @@ func (b *batch) addSweeps(ctx context.Context, sweeps []*sweep) (bool, error) {
 		// Update FeeRate. Max(s.minFeeRate) for all the sweeps of
 		// the batch is the basis for fee bumps.
 		if b.rbfCache.FeeRate < s.minFeeRate {
+			b.Infof("Increasing feerate of the batch "+
+				"from %v to %v", b.rbfCache.FeeRate,
+				s.minFeeRate)
 			b.rbfCache.FeeRate = s.minFeeRate
 			b.rbfCache.SkipNextBump = true
 		}
@@ -968,6 +975,12 @@ func (b *batch) Run(ctx context.Context) error {
 				continue
 			}
 
+			// Update feerate of sweeps. This is normally done by
+			// AddSweep, but it may not be called after the sweep
+			// is confirmed, but fresh feerate is still needed to
+			// keep publishing in case of reorg.
+			b.updateFeeRate(ctx)
+
 			err := b.publish(ctx)
 			if err != nil {
 				return fmt.Errorf("publish error: %w", err)
@@ -1016,6 +1029,41 @@ func (b *batch) Run(ctx context.Context) error {
 	}
 }
 
+// updateFeeRate gets fresh values of minFeeRate for sweeps and updates the
+// feerate of the batch if needed. This method must be called from event loop.
+func (b *batch) updateFeeRate(ctx context.Context) {
+	for outpoint, s := range b.sweeps {
+		minFeeRate, err := minimumSweepFeeRate(
+			ctx, b.cfg.customFeeRate, b.wallet,
+			s.swapHash, s.outpoint, s.confTarget,
+		)
+		if err != nil {
+			b.Warnf("failed to determine feerate for sweep %v of "+
+				"swap %x, confTarget %d: %w", s.outpoint,
+				s.swapHash[:6], s.confTarget, err)
+			continue
+		}
+
+		if minFeeRate <= s.minFeeRate {
+			continue
+		}
+
+		b.Infof("Increasing feerate of sweep %v of swap %x from %v "+
+			"to %v", s.outpoint, s.swapHash[:6], s.minFeeRate,
+			minFeeRate)
+		s.minFeeRate = minFeeRate
+		b.sweeps[outpoint] = s
+
+		if s.minFeeRate <= b.rbfCache.FeeRate {
+			continue
+		}
+
+		b.Infof("Increasing feerate of the batch from %v to %v",
+			b.rbfCache.FeeRate, s.minFeeRate)
+		b.rbfCache.FeeRate = s.minFeeRate
+	}
+}
+
 // testRunInEventLoop runs a function in the event loop blocking until
 // the function returns. For unit tests only!
 func (b *batch) testRunInEventLoop(ctx context.Context, handler func()) {
@@ -1793,7 +1841,7 @@ func (b *batch) updateRbfRate(ctx context.Context) error {
 
 		// Set the initial value for our fee rate.
 		b.rbfCache.FeeRate = rate
-	} else if !b.cfg.noBumping {
+	} else if noBumping := b.cfg.customFeeRate != nil; !noBumping {
 		if b.rbfCache.SkipNextBump {
 			// Skip fee bumping, unset the flag, to bump next time.
 			b.rbfCache.SkipNextBump = false

sweepbatcher/sweep_batcher.go

@@ -1522,28 +1522,12 @@ func (b *Batcher) loadSweep(ctx context.Context, swapHash lntypes.Hash,
 
 	// Find minimum fee rate for the sweep. Use customFeeRate if it is
 	// provided, otherwise use wallet's EstimateFeeRate.
-	var minFeeRate chainfee.SatPerKWeight
-	if b.customFeeRate != nil {
-		minFeeRate, err = b.customFeeRate(ctx, swapHash, outpoint)
-		if err != nil {
-			return nil, fmt.Errorf("failed to fetch min fee rate "+
-				"for %x: %w", swapHash[:6], err)
-		}
-		if minFeeRate < chainfee.AbsoluteFeePerKwFloor {
-			return nil, fmt.Errorf("min fee rate too low (%v) for "+
-				"%x", minFeeRate, swapHash[:6])
-		}
-	} else {
-		if s.ConfTarget == 0 {
-			warnf("Fee estimation was requested for zero "+
-				"confTarget for sweep %x.", swapHash[:6])
-		}
-		minFeeRate, err = b.wallet.EstimateFeeRate(ctx, s.ConfTarget)
-		if err != nil {
-			return nil, fmt.Errorf("failed to estimate fee rate "+
-				"for %x, confTarget=%d: %w", swapHash[:6],
-				s.ConfTarget, err)
-		}
+	minFeeRate, err := minimumSweepFeeRate(
+		ctx, b.customFeeRate, b.wallet,
+		swapHash, outpoint, s.ConfTarget,
+	)
+	if err != nil {
+		return nil, err
 	}
 
 	return &sweep{
@@ -1567,11 +1551,53 @@ func (b *Batcher) loadSweep(ctx context.Context, swapHash lntypes.Hash,
 	}, nil
 }
 
+// feeRateEstimator determines feerate by confTarget.
+type feeRateEstimator interface {
+	// EstimateFeeRate returns feerate corresponding to the confTarget.
+	EstimateFeeRate(ctx context.Context,
+		confTarget int32) (chainfee.SatPerKWeight, error)
+}
+
+// minimumSweepFeeRate determines minimum feerate for a sweep.
+func minimumSweepFeeRate(ctx context.Context, customFeeRate FeeRateProvider,
+	wallet feeRateEstimator, swapHash lntypes.Hash, outpoint wire.OutPoint,
+	sweepConfTarget int32) (chainfee.SatPerKWeight, error) {
+
+	// Find minimum fee rate for the sweep. Use customFeeRate if it is
+	// provided, otherwise use wallet's EstimateFeeRate.
+	if customFeeRate != nil {
+		minFeeRate, err := customFeeRate(ctx, swapHash, outpoint)
+		if err != nil {
+			return 0, fmt.Errorf("failed to fetch min fee rate "+
+				"for %x: %w", swapHash[:6], err)
+		}
+		if minFeeRate < chainfee.AbsoluteFeePerKwFloor {
+			return 0, fmt.Errorf("min fee rate too low (%v) for "+
+				"%x", minFeeRate, swapHash[:6])
+		}
+
+		return minFeeRate, nil
+	}
+
+	if sweepConfTarget == 0 {
+		warnf("Fee estimation was requested for zero "+
+			"confTarget for sweep %x.", swapHash[:6])
+	}
+	minFeeRate, err := wallet.EstimateFeeRate(ctx, sweepConfTarget)
+	if err != nil {
+		return 0, fmt.Errorf("failed to estimate fee rate "+
+			"for %x, confTarget=%d: %w", swapHash[:6],
+			sweepConfTarget, err)
+	}
+
+	return minFeeRate, nil
+}
+
 // newBatchConfig creates new batch config.
 func (b *Batcher) newBatchConfig(maxTimeoutDistance int32) batchConfig {
 	return batchConfig{
 		maxTimeoutDistance: maxTimeoutDistance,
-		noBumping:          b.customFeeRate != nil,
+		customFeeRate:      b.customFeeRate,
 		txLabeler:          b.txLabeler,
 		customMuSig2Signer: b.customMuSig2Signer,
 		presignedHelper:    b.presignedHelper,

sweepbatcher/sweep_batcher_test.go

@@ -4838,8 +4838,6 @@ func testFeeRateGrows(t *testing.T, store testStore,
 	// Now update fee rate of second sweep (which is not primary) to
 	// feeRateHigh. Fee rate of sweep 1 is still feeRateLow.
 	setFeeRate(swapHash2, feeRateHigh)
-	require.NoError(t, batcher.AddSweep(ctx, &sweepReq1))
-	require.NoError(t, batcher.AddSweep(ctx, &sweepReq2))
 
 	// Tick tock next block.
 	err = lnd.NotifyHeight(603)