diff --git a/sweepbatcher/greedy_batch_selection.go b/sweepbatcher/greedy_batch_selection.go index 1ecda9b0..e1056a18 100644 --- a/sweepbatcher/greedy_batch_selection.go +++ b/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. diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index 47deb2be..91a6f2d1 100644 --- a/sweepbatcher/sweep_batch.go +++ b/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 "+ diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go index 777d5c2c..ec5453a8 100644 --- a/sweepbatcher/sweep_batcher.go +++ b/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 } diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index 546a23e0..009e8da0 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/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.