sweepbatcher: threshold for immediate batch publishing

sweepbatcher/sweep_batch.go

@@ -200,6 +200,11 @@ type batchConfig struct {
 	// chainParams are the chain parameters of the chain that is used by
 	// batches.
 	chainParams *chaincfg.Params
+
+	// immediatePublishThreshold is the cumulative value threshold above
+	// which a batch should be published immediately without waiting for
+	// the initial delay grouping period. If 0, the feature is disabled.
+	immediatePublishThreshold btcutil.Amount
 }
 
 // rbfCache stores data related to our last fee bump.
@@ -807,6 +812,24 @@ func (b *batch) sweepExists(outpoint wire.OutPoint) bool {
 	return ok
 }
 
+// totalValue returns the sum of all sweep values in the batch.
+func (b *batch) totalValue() btcutil.Amount {
+	var total btcutil.Amount
+	for _, sweep := range b.sweeps {
+		total += sweep.value
+	}
+	return total
+}
+
+// exceedsThreshold checks if the batch value exceeds the immediate publish
+// threshold. Returns false if the threshold is not configured (0).
+func (b *batch) exceedsThreshold() bool {
+	if b.cfg.immediatePublishThreshold == 0 {
+		return false
+	}
+	return b.totalValue() >= b.cfg.immediatePublishThreshold
+}
+
 // Wait waits for the batch to gracefully stop.
 func (b *batch) Wait() {
 	b.Infof("Stopping")
@@ -920,6 +943,25 @@ func (b *batch) Run(ctx context.Context) error {
 				skipBefore = &delayStop
 				skipBeforeUpdated = true
 			}
+
+			// Check if batch exceeds threshold and should publish
+			// immediately. Only override skipBefore if threshold
+			// feature is enabled and exceeded.
+			if b.exceedsThreshold() {
+				b.Infof("Batch value (%v) exceeds threshold "+
+					"(%v), publishing immediately",
+					b.totalValue(),
+					b.cfg.immediatePublishThreshold)
+
+				// Set skipBefore to now to bypass initial
+				// delay.
+				now := clock.Now()
+				skipBefore = &now
+
+				// Trigger immediate publish by setting timer to
+				// fire after batchPublishDelay.
+				timerChan = clock.TickAfter(b.cfg.batchPublishDelay)
+			}
 		}
 
 		// Create new timer only if the value of skipBefore was updated.

sweepbatcher/sweep_batcher.go

@@ -443,6 +443,11 @@ type Batcher struct {
 	// 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{}
+
+	// immediatePublishThreshold is the cumulative value threshold above
+	// which a batch should be published immediately without waiting for
+	// the initial delay grouping period. If 0, the feature is disabled.
+	immediatePublishThreshold btcutil.Amount
 }
 
 // BatcherConfig holds batcher configuration.
@@ -491,6 +496,11 @@ type BatcherConfig struct {
 	// 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{}
+
+	// immediatePublishThreshold is the cumulative value threshold above
+	// which a batch should be published immediately without waiting for
+	// the initial delay grouping period. If 0, the feature is disabled.
+	immediatePublishThreshold btcutil.Amount
 }
 
 // BatcherOption configures batcher behaviour.
@@ -581,6 +591,14 @@ func WithSkippedTxns(skippedTxns map[chainhash.Hash]struct{}) BatcherOption {
 	}
 }
 
+// WithImmediatePublishThreshold sets the threshold above which batches are
+// published immediately without waiting for the initial delay grouping period.
+func WithImmediatePublishThreshold(threshold btcutil.Amount) BatcherOption {
+	return func(cfg *BatcherConfig) {
+		cfg.immediatePublishThreshold = threshold
+	}
+}
+
 // NewBatcher creates a new Batcher instance.
 func NewBatcher(wallet lndclient.WalletKitClient,
 	chainNotifier lndclient.ChainNotifierClient,
@@ -626,29 +644,30 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 	}
 
 	return &Batcher{
-		batches:              make(map[int32]*batch),
-		addSweepsChan:        make(chan *addSweepsRequest),
-		testReqs:             make(chan *testRequest),
-		errChan:              make(chan error, 1),
-		quit:                 make(chan struct{}),
-		initDone:             make(chan struct{}),
-		wallet:               wallet,
-		chainNotifier:        chainNotifier,
-		signerClient:         signerClient,
-		musig2ServerSign:     musig2ServerSigner,
-		VerifySchnorrSig:     verifySchnorrSig,
-		chainParams:          chainparams,
-		store:                store,
-		sweepStore:           sweepStore,
-		clock:                cfg.clock,
-		initialDelayProvider: cfg.initialDelayProvider,
-		publishDelay:         cfg.publishDelay,
-		customFeeRate:        cfg.customFeeRate,
-		txLabeler:            cfg.txLabeler,
-		customMuSig2Signer:   cfg.customMuSig2Signer,
-		publishErrorHandler:  cfg.publishErrorHandler,
-		presignedHelper:      cfg.presignedHelper,
-		skippedTxns:          cfg.skippedTxns,
+		batches:                   make(map[int32]*batch),
+		addSweepsChan:             make(chan *addSweepsRequest),
+		testReqs:                  make(chan *testRequest),
+		errChan:                   make(chan error, 1),
+		quit:                      make(chan struct{}),
+		initDone:                  make(chan struct{}),
+		wallet:                    wallet,
+		chainNotifier:             chainNotifier,
+		signerClient:              signerClient,
+		musig2ServerSign:          musig2ServerSigner,
+		VerifySchnorrSig:          verifySchnorrSig,
+		chainParams:               chainparams,
+		store:                     store,
+		sweepStore:                sweepStore,
+		clock:                     cfg.clock,
+		initialDelayProvider:      cfg.initialDelayProvider,
+		publishDelay:              cfg.publishDelay,
+		customFeeRate:             cfg.customFeeRate,
+		txLabeler:                 cfg.txLabeler,
+		customMuSig2Signer:        cfg.customMuSig2Signer,
+		publishErrorHandler:       cfg.publishErrorHandler,
+		presignedHelper:           cfg.presignedHelper,
+		skippedTxns:               cfg.skippedTxns,
+		immediatePublishThreshold: cfg.immediatePublishThreshold,
 	}
 }
 
@@ -1683,14 +1702,15 @@ func minimumSweepFeeRate(ctx context.Context, customFeeRate FeeRateProvider,
 // newBatchConfig creates new batch config.
 func (b *Batcher) newBatchConfig(maxTimeoutDistance int32) batchConfig {
 	return batchConfig{
-		maxTimeoutDistance: maxTimeoutDistance,
-		customFeeRate:      b.customFeeRate,
-		txLabeler:          b.txLabeler,
-		customMuSig2Signer: b.customMuSig2Signer,
-		presignedHelper:    b.presignedHelper,
-		skippedTxns:        b.skippedTxns,
-		clock:              b.clock,
-		chainParams:        b.chainParams,
+		maxTimeoutDistance:        maxTimeoutDistance,
+		customFeeRate:             b.customFeeRate,
+		txLabeler:                 b.txLabeler,
+		customMuSig2Signer:        b.customMuSig2Signer,
+		presignedHelper:           b.presignedHelper,
+		skippedTxns:               b.skippedTxns,
+		clock:                     b.clock,
+		chainParams:               b.chainParams,
+		immediatePublishThreshold: b.immediatePublishThreshold,
 	}
 }