sweepbatcher: fast-flag enables zero delay publishing

Author: hieblmi

sweepbatcher/greedy_batch_selection.go

@@ -75,9 +75,10 @@ func (b *Batcher) greedyAddSweeps(ctx context.Context, sweeps []*sweep) error {
 
 	// Try batches, starting with the best.
 	for _, batchId := range batchesIds {
-		// If the best option is to start new batch, do it.
+		// If the best option is to start a new batch, do it.
 		if batchId == newBatchSignal {
-			return b.spinUpNewBatch(ctx, sweeps)
+			fast := false
+			return b.spinUpNewBatch(ctx, sweeps, fast)
 		}
 
 		// Locate the batch to add the sweeps to.

sweepbatcher/sweep_batch.go

@@ -900,8 +900,9 @@ func (b *batch) Run(ctx context.Context) error {
 
 		skipBeforeUpdated := false
 		if totalSweptAmt != 0 {
+			fast := false
 			initialDelay, err := b.cfg.initialDelayProvider(
-				ctx, len(b.sweeps), totalSweptAmt,
+				ctx, len(b.sweeps), totalSweptAmt, fast,
 			)
 			if err != nil {
 				b.Warnf("InitialDelayProvider failed: %v. We "+

sweepbatcher/sweep_batcher.go

@@ -210,11 +210,11 @@ type FeeRateProvider func(ctx context.Context, swapHash lntypes.Hash,
 // faster. If the function returns an error, no delay is used and the error is
 // logged as a warning.
 type InitialDelayProvider func(ctx context.Context, numSweeps int,
-	value btcutil.Amount) (time.Duration, error)
+	value btcutil.Amount, fast bool) (time.Duration, error)
 
 // zeroInitialDelay returns no delay for any sweeps.
 func zeroInitialDelay(_ context.Context, _ int,
-	_ btcutil.Amount) (time.Duration, error) {
+	_ btcutil.Amount, _ bool) (time.Duration, error) {
 
 	return 0, nil
 }
@@ -259,6 +259,10 @@ type SweepRequest struct {
 	// Notifier is a notifier that is used to notify the requester of this
 	// sweep that the sweep was successful.
 	Notifier *SpendNotifier
+
+	// Fast is set by the client if the sweep should be published
+	// immediately.
+	Fast bool
 }
 
 // addSweepsRequest is a request to sweep an outpoint or a group of outpoints
@@ -271,6 +275,9 @@ type addSweepsRequest struct {
 	// Notifier is a notifier that is used to notify the requester of this
 	// sweep that the sweep was successful.
 	notifier *SpendNotifier
+
+	// fast indicates sweeps that are part of a fast swap.
+	fast bool
 }
 
 // SpendDetail is a notification that is send to the user of sweepbatcher when
@@ -679,7 +686,9 @@ func (b *Batcher) Run(ctx context.Context) error {
 	for {
 		select {
 		case req := <-b.addSweepsChan:
-			err = b.handleSweeps(runCtx, req.sweeps, req.notifier)
+			err = b.handleSweeps(
+				runCtx, req.sweeps, req.notifier, req.fast,
+			)
 			if err != nil {
 				warnf("handleSweeps failed: %v.", err)
 
@@ -792,12 +801,15 @@ func (b *Batcher) AddSweep(ctx context.Context, sweepReq *SweepRequest) error {
 		return fmt.Errorf("failed to get the status of sweep %v: %w",
 			sweep.outpoint, err)
 	}
-	var fullyConfirmed bool
+	var (
+		parentBatch    *dbBatch
+		fullyConfirmed bool
+	)
 	if completed {
 		// Verify that the parent batch is confirmed. Note that a batch
 		// is only considered confirmed after it has received three
 		// on-chain confirmations to prevent issues caused by reorgs.
-		parentBatch, err := b.store.GetParentBatch(ctx, sweep.outpoint)
+		parentBatch, err = b.store.GetParentBatch(ctx, sweep.outpoint)
 		if err != nil {
 			return fmt.Errorf("unable to get parent batch for "+
 				"sweep %x: %w", sweep.swapHash[:6], err)
@@ -835,6 +847,7 @@ func (b *Batcher) AddSweep(ctx context.Context, sweepReq *SweepRequest) error {
 	req := &addSweepsRequest{
 		sweeps:   sweeps,
 		notifier: sweepReq.Notifier,
+		fast:     sweepReq.Fast,
 	}
 
 	select {
@@ -879,7 +892,7 @@ func (b *Batcher) testRunInEventLoop(ctx context.Context, handler func()) {
 // handleSweeps handles a sweep request by either placing the group of sweeps in
 // an existing batch, or by spinning up a new batch for it.
 func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
-	notifier *SpendNotifier) error {
+	notifier *SpendNotifier, fast bool) error {
 
 	// Since the whole group is added to the same batch and belongs to
 	// the same transaction, we use sweeps[0] below where we need any sweep.
@@ -968,6 +981,12 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
 		}
 	}
 
+	// If fast is set, we spin up a new batch which is published
+	// immediately.
+	if fast {
+		return b.spinUpNewBatch(ctx, sweeps, true)
+	}
+
 	// Try to run the greedy algorithm of batch selection to minimize costs.
 	err = b.greedyAddSweeps(ctx, sweeps)
 	if err == nil {
@@ -994,15 +1013,17 @@ func (b *Batcher) handleSweeps(ctx context.Context, sweeps []*sweep,
 
 	// If no batch is capable of accepting the sweep, we spin up a fresh
 	// batch and hand the sweep over to it.
-	return b.spinUpNewBatch(ctx, sweeps)
+	return b.spinUpNewBatch(ctx, sweeps, false)
 }
 
 // spinUpNewBatch creates new batch, starts it and adds the sweeps to it. If
 // presigned mode is enabled, the result also depends on outcome of
 // presignedHelper.SignTx.
-func (b *Batcher) spinUpNewBatch(ctx context.Context, sweeps []*sweep) error {
+func (b *Batcher) spinUpNewBatch(ctx context.Context, sweeps []*sweep,
+	fast bool) error {
+
 	// Spin up a fresh batch.
-	newBatch, err := b.spinUpBatch(ctx)
+	newBatch, err := b.spinUpBatch(ctx, fast)
 	if err != nil {
 		return err
 	}
@@ -1024,7 +1045,7 @@ func (b *Batcher) spinUpNewBatch(ctx context.Context, sweeps []*sweep) error {
 }
 
 // spinUpBatch spins up a new batch and returns it.
-func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
+func (b *Batcher) spinUpBatch(ctx context.Context, fast bool) (*batch, error) {
 	cfg := b.newBatchConfig(defaultMaxTimeoutDistance)
 
 	switch b.chainParams {
@@ -1044,7 +1065,7 @@ func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
 	}
 
 	cfg.initialDelayProvider = b.initialDelayProvider
-	if cfg.initialDelayProvider == nil {
+	if cfg.initialDelayProvider == nil || fast {
 		cfg.initialDelayProvider = zeroInitialDelay
 	}
 

sweepbatcher/sweep_batcher_test.go

@@ -1374,7 +1374,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) {
 	)
 
 	initialDelayProvider := func(_ context.Context, _ int,
-		_ btcutil.Amount) (time.Duration, error) {
+		_ btcutil.Amount, fast bool) (time.Duration, error) {
 
 		return initialDelay, nil
 	}
@@ -1652,7 +1652,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) {
 	const largeInitialDelay = 6 * time.Hour
 
 	largeInitialDelayProvider := func(_ context.Context, _ int,
-		_ btcutil.Amount) (time.Duration, error) {
+		_ btcutil.Amount, fast bool) (time.Duration, error) {
 
 		return largeInitialDelay, nil
 	}
@@ -1877,7 +1877,7 @@ func testCustomDelays(t *testing.T, store testStore,
 
 	// initialDelay returns initialDelay depending of batch size (sats).
 	initialDelayProvider := func(_ context.Context, numSweeps int,
-		value btcutil.Amount) (time.Duration, error) {
+		value btcutil.Amount, fast bool) (time.Duration, error) {
 
 		if value <= swapSize1 {
 			// Verify the number of sweeps.