diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index 49045781..1cf8e573 100644 --- a/sweepbatcher/sweep_batch.go +++ b/sweepbatcher/sweep_batch.go @@ -151,11 +151,13 @@ type batchConfig struct { // clock provides methods to work with time and timers. clock clock.Clock - // initialDelay is the delay of first batch publishing after creation. - // It only affects newly created batches, not batches loaded from DB, - // so publishing does happen in case of a daemon restart (especially - // important in case of a crashloop). - initialDelay time.Duration + // initialDelayProvider provides the delay of first batch publishing + // after creation. It only affects newly created batches, not batches + // loaded from DB, so publishing does happen in case of a daemon restart + // (especially important in case of a crashloop). If a sweep is about to + // expire (time until timeout is less that 2x initialDelay), then + // waiting is skipped. + initialDelayProvider InitialDelayProvider // batchPublishDelay is the delay between receiving a new block or // initial delay completion and publishing the batch transaction. @@ -650,6 +652,7 @@ func (b *batch) Run(ctx context.Context) error { // Cache clock variable. clock := b.cfg.clock + startTime := clock.Now() blockChan, blockErrChan, err := b.chainNotifier.RegisterBlockEpochNtfn(runCtx) @@ -679,17 +682,15 @@ func (b *batch) Run(ctx context.Context) error { // skipBefore is the time before which we skip batch publishing. // This is needed to facilitate better grouping of sweeps. + // The value is set only if the batch has at least one sweep. // For batches loaded from DB initialDelay should be 0. - skipBefore := clock.Now().Add(b.cfg.initialDelay) + var skipBefore *time.Time // initialDelayChan is a timer which fires upon initial delay end. // If initialDelay is set to 0, it will not trigger to avoid setting up // timerChan twice, which could lead to double publishing if // batchPublishDelay is also 0. var initialDelayChan <-chan time.Time - if b.cfg.initialDelay > 0 { - initialDelayChan = clock.TickAfter(b.cfg.initialDelay) - } // We use a timer in order to not publish new transactions at the same // time as the block epoch notification. This is done to prevent @@ -703,6 +704,45 @@ func (b *batch) Run(ctx context.Context) error { b.primarySweepID, len(b.sweeps)) for { + // If the batch is not empty, find earliest initialDelay. + var totalSweptAmt btcutil.Amount + for _, sweep := range b.sweeps { + totalSweptAmt += sweep.value + } + + skipBeforeUpdated := false + if totalSweptAmt != 0 { + initialDelay, err := b.cfg.initialDelayProvider( + ctx, len(b.sweeps), totalSweptAmt, + ) + if err != nil { + b.Warnf("InitialDelayProvider failed: %v. We "+ + "publish this batch without a delay.", + err) + initialDelay = 0 + } + if initialDelay < 0 { + b.Warnf("Negative delay: %v. We publish this "+ + "batch without a delay.", initialDelay) + initialDelay = 0 + } + delayStop := startTime.Add(initialDelay) + if skipBefore == nil || delayStop.Before(*skipBefore) { + skipBefore = &delayStop + skipBeforeUpdated = true + } + } + + // Create new timer only if the value of skipBefore was updated. + // Don't create the timer if the delay is <= 0 to avoid double + // publishing if batchPublishDelay is also 0. + if skipBeforeUpdated { + delay := skipBefore.Sub(clock.Now()) + if delay > 0 { + initialDelayChan = clock.TickAfter(delay) + } + } + select { case <-b.callEnter: <-b.callLeave @@ -718,7 +758,7 @@ func (b *batch) Run(ctx context.Context) error { case <-initialDelayChan: b.Debugf("initial delay of duration %v has ended", - b.cfg.initialDelay) + clock.Now().Sub(startTime)) // Set the timer to publish the batch transaction after // the configured delay. @@ -732,9 +772,15 @@ func (b *batch) Run(ctx context.Context) error { continue } + if skipBefore == nil { + b.Debugf("Skipping publishing, because " + + "the batch is empty.") + continue + } + // If the batch became urgent, skipBefore is set to now. - if b.isUrgent(skipBefore) { - skipBefore = clock.Now() + if b.isUrgent(*skipBefore) { + *skipBefore = clock.Now() } // Check that the initial delay has ended. We have also @@ -742,7 +788,7 @@ func (b *batch) Run(ctx context.Context) error { // initialDelayChan has just fired, this check passes. now := clock.Now() if skipBefore.After(now) { - b.Debugf(stillWaitingMsg, skipBefore, now) + b.Debugf(stillWaitingMsg, *skipBefore, now) continue } diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go index d71387dc..3c253497 100644 --- a/sweepbatcher/sweep_batcher.go +++ b/sweepbatcher/sweep_batcher.go @@ -165,6 +165,21 @@ type VerifySchnorrSig func(pubKey *btcec.PublicKey, hash, sig []byte) error type FeeRateProvider func(ctx context.Context, swapHash lntypes.Hash) (chainfee.SatPerKWeight, error) +// InitialDelayProvider returns the duration after which a newly created batch +// is first published. It allows to customize the duration based on total value +// of the batch. There is a trade-off between better grouping and getting funds +// 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) + +// zeroInitialDelay returns no delay for any sweeps. +func zeroInitialDelay(_ context.Context, _ int, + _ btcutil.Amount) (time.Duration, error) { + + return 0, nil +} + // PublishErrorHandler is a function that handles transaction publishing error. type PublishErrorHandler func(err error, errMsg string, log btclog.Logger) @@ -299,13 +314,13 @@ type Batcher struct { // clock provides methods to work with time and timers. clock clock.Clock - // initialDelay is the delay of first batch publishing after creation. - // It only affects newly created batches, not batches loaded from DB, - // so publishing does happen in case of a daemon restart (especially - // important in case of a crashloop). If a sweep is about to expire - // (time until timeout is less that 2x initialDelay), then waiting is - // skipped. - initialDelay time.Duration + // initialDelayProvider provides the delay of first batch publishing + // after creation. It only affects newly created batches, not batches + // loaded from DB, so publishing does happen in case of a daemon restart + // (especially important in case of a crashloop). If a sweep is about to + // expire (time until timeout is less that 2x initialDelay), then + // waiting is skipped. + initialDelayProvider InitialDelayProvider // publishDelay is the delay of batch publishing that is applied in the // beginning, after the appearance of a new block in the network or @@ -339,13 +354,13 @@ type BatcherConfig struct { // clock provides methods to work with time and timers. clock clock.Clock - // initialDelay is the delay of first batch publishing after creation. - // It only affects newly created batches, not batches loaded from DB, - // so publishing does happen in case of a daemon restart (especially - // important in case of a crashloop). If a sweep is about to expire - // (time until timeout is less that 2x initialDelay), then waiting is - // skipped. - initialDelay time.Duration + // initialDelayProvider provides the delay of first batch publishing + // after creation. It only affects newly created batches, not batches + // loaded from DB, so publishing does happen in case of a daemon restart + // (especially important in case of a crashloop). If a sweep is about to + // expire (time until timeout is less that 2x initialDelay), then + // waiting is skipped. + initialDelayProvider InitialDelayProvider // publishDelay is the delay of batch publishing that is applied in the // beginning, after the appearance of a new block in the network or @@ -390,9 +405,9 @@ func WithClock(clock clock.Clock) BatcherOption { // better grouping. Defaults to 0s (no initial delay). If a sweep is about // to expire (time until timeout is less that 2x initialDelay), then waiting // is skipped. -func WithInitialDelay(initialDelay time.Duration) BatcherOption { +func WithInitialDelay(provider InitialDelayProvider) BatcherOption { return func(cfg *BatcherConfig) { - cfg.initialDelay = initialDelay + cfg.initialDelayProvider = provider } } @@ -478,27 +493,27 @@ func NewBatcher(wallet lndclient.WalletKitClient, } return &Batcher{ - batches: make(map[int32]*batch), - sweepReqs: make(chan SweepRequest), - 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, - initialDelay: cfg.initialDelay, - publishDelay: cfg.publishDelay, - customFeeRate: cfg.customFeeRate, - txLabeler: cfg.txLabeler, - customMuSig2Signer: cfg.customMuSig2Signer, - publishErrorHandler: cfg.publishErrorHandler, + batches: make(map[int32]*batch), + sweepReqs: make(chan SweepRequest), + 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, } } @@ -749,11 +764,10 @@ func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) { cfg.batchPublishDelay = b.publishDelay } - if b.initialDelay < 0 { - return nil, fmt.Errorf("negative initialDelay: %v", - b.initialDelay) + cfg.initialDelayProvider = b.initialDelayProvider + if cfg.initialDelayProvider == nil { + cfg.initialDelayProvider = zeroInitialDelay } - cfg.initialDelay = b.initialDelay batchKit := b.newBatchKit() @@ -847,6 +861,8 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error { // Note that initialDelay and batchPublishDelay are 0 for batches // recovered from DB so publishing happen in case of a daemon restart // (especially important in case of a crashloop). + cfg.initialDelayProvider = zeroInitialDelay + newBatch, err := NewBatchFromDB(cfg, batchKit) if err != nil { return fmt.Errorf("failed in NewBatchFromDB: %w", err) diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index 1cf6cd70..dfef3395 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/sweepbatcher/sweep_batcher_test.go @@ -920,6 +920,12 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { publishDelay = 3 * time.Second ) + initialDelayProvider := func(_ context.Context, _ int, + _ btcutil.Amount) (time.Duration, error) { + + return initialDelay, nil + } + defer test.Guard(t)() lnd := test.NewMockLnd() @@ -935,7 +941,8 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { batcher := NewBatcher( lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams, - batcherStore, sweepStore, WithInitialDelay(initialDelay), + batcherStore, sweepStore, + WithInitialDelay(initialDelayProvider), WithPublishDelay(publishDelay), WithClock(testClock), ) @@ -1010,7 +1017,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { wg2.Wait() // Expect timer for initialDelay and publishDelay to be registered. - wantDelays := []time.Duration{initialDelay, publishDelay} + wantDelays := []time.Duration{publishDelay, initialDelay} require.Equal(t, wantDelays, delays) // Eventually the batch is launched. @@ -1090,7 +1097,8 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { batcher = NewBatcher( lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams, - batcherStore, sweepStore, WithInitialDelay(initialDelay), + batcherStore, sweepStore, + WithInitialDelay(initialDelayProvider), WithPublishDelay(publishDelay), WithClock(testClock), ) ctx, cancel = context.WithCancel(context.Background()) @@ -1187,10 +1195,17 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { // for an urgent sweep. const largeInitialDelay = 6 * time.Hour + largeInitialDelayProvider := func(_ context.Context, _ int, + _ btcutil.Amount) (time.Duration, error) { + + return largeInitialDelay, nil + } + batcher = NewBatcher( lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams, - batcherStore, sweepStore, WithInitialDelay(largeInitialDelay), + batcherStore, sweepStore, + WithInitialDelay(largeInitialDelayProvider), WithPublishDelay(publishDelay), WithClock(testClock), ) ctx, cancel = context.WithCancel(context.Background()) @@ -1302,7 +1317,7 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { wg5.Wait() // Expect two timers: largeInitialDelay, publishDelay. - wantDelays = []time.Duration{largeInitialDelay, publishDelay} + wantDelays = []time.Duration{publishDelay, largeInitialDelay} require.Equal(t, wantDelays, delays) // Replace the logger in the batch with wrappedLogger to watch messages. @@ -1377,6 +1392,228 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { checkBatcherError(t, runErr) } +// testCustomDelays tests per-sweep customization in WithInitialDelay. +func testCustomDelays(t *testing.T, store testStore, + batcherStore testBatcherStore) { + + defer test.Guard(t)() + + // Set initial delay and publish delay. + const ( + initialDelay1 = 100 * time.Second + initialDelay2 = 4 * time.Second + publishDelay = 3 * time.Second + ) + + swapHash1 := lntypes.Hash{1, 1, 1} + swapHash2 := lntypes.Hash{2, 2, 2} + + const ( + swapSize1 = 111 + swapSize2 = 222 + ) + + // initialDelay returns initialDelay depending of batch size (sats). + initialDelayProvider := func(_ context.Context, numSweeps int, + value btcutil.Amount) (time.Duration, error) { + + if value <= swapSize1 { + // Verify the number of sweeps. + if numSweeps != 1 { + return 0, fmt.Errorf("got unexpected number "+ + "of sweeps: %d, want %d", numSweeps, 1) + } + + return initialDelay1, nil + } else { + // Verify the number of sweeps. + if numSweeps != 2 { + return 0, fmt.Errorf("got unexpected number "+ + "of sweeps: %d, want %d", numSweeps, 2) + } + + return initialDelay2, nil + } + } + + lnd := test.NewMockLnd() + ctx, cancel := context.WithCancel(context.Background()) + + sweepStore, err := NewSweepFetcherFromSwapStore(store, lnd.ChainParams) + require.NoError(t, err) + + startTime := time.Date(2018, 11, 1, 0, 0, 0, 0, time.UTC) + now := startTime + tickSignal := make(chan time.Duration) + testClock := clock.NewTestClockWithTickSignal(startTime, tickSignal) + + batcher := NewBatcher( + lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, + testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams, + batcherStore, sweepStore, + WithInitialDelay(initialDelayProvider), + WithPublishDelay(publishDelay), WithClock(testClock), + ) + + var wg sync.WaitGroup + wg.Add(1) + + var runErr error + go func() { + defer wg.Done() + runErr = batcher.Run(ctx) + }() + + // Wait for the batcher to be initialized. + <-batcher.initDone + + // Create a sweep request. + sweepReq1 := SweepRequest{ + SwapHash: swapHash1, + Value: swapSize1, + Outpoint: wire.OutPoint{ + Hash: chainhash.Hash{1, 1}, + Index: 1, + }, + Notifier: &dummyNotifier, + } + + swap1 := &loopdb.LoopOutContract{ + SwapContract: loopdb.SwapContract{ + CltvExpiry: 1000, + AmountRequested: 111, + ProtocolVersion: loopdb.ProtocolVersionMuSig2, + HtlcKeys: htlcKeys, + }, + + DestAddr: destAddr, + SwapInvoice: swapInvoice, + SweepConfTarget: confTarget, + } + err = store.CreateLoopOut(ctx, swapHash1, swap1) + require.NoError(t, err) + store.AssertLoopOutStored() + + // Deliver sweep request to batcher. + require.NoError(t, batcher.AddSweep(&sweepReq1)) + + // Expect two timers to be set: initialDelay and publishDelay, + // and RegisterSpend to be called. The order is not determined, + // so catch these actions from two separate goroutines. + var wg2 sync.WaitGroup + + wg2.Add(1) + go func() { + defer wg2.Done() + + // Since a batch was created we check that it registered for its + // primary sweep's spend. + <-lnd.RegisterSpendChannel + }() + + wg2.Add(1) + var delays []time.Duration + go func() { + defer wg2.Done() + + // Expect two timers: initialDelay and publishDelay. + delays = append(delays, <-tickSignal) + delays = append(delays, <-tickSignal) + }() + + // Wait for RegisterSpend and for timer registrations. + wg2.Wait() + + // Expect timer for initialDelay1 and publishDelay to be registered. + wantDelays := []time.Duration{publishDelay, initialDelay1} + require.Equal(t, wantDelays, delays) + + // Eventually the batch is launched. + require.Eventually(t, func() bool { + return batcher.numBatches(ctx) == 1 + }, test.Timeout, eventuallyCheckFrequency) + + // Now add swap 2, which has lower initialDelay. + sweepReq2 := SweepRequest{ + SwapHash: swapHash2, + Value: swapSize2, + Outpoint: wire.OutPoint{ + Hash: chainhash.Hash{2, 2}, + Index: 2, + }, + Notifier: &dummyNotifier, + } + + swap2 := &loopdb.LoopOutContract{ + SwapContract: loopdb.SwapContract{ + CltvExpiry: 1000, + AmountRequested: 111, + ProtocolVersion: loopdb.ProtocolVersionMuSig2, + HtlcKeys: htlcKeys, + + // Make preimage unique to pass SQL constraints. + Preimage: lntypes.Preimage{2}, + }, + + DestAddr: destAddr, + SwapInvoice: swapInvoice, + SweepConfTarget: confTarget, + } + err = store.CreateLoopOut(ctx, swapHash2, swap2) + require.NoError(t, err) + store.AssertLoopOutStored() + + // Deliver sweep request to batcher. + require.NoError(t, batcher.AddSweep(&sweepReq2)) + + // Expect timer for initialDelay2 to be registered, because + // initialDelay2 is lower than initialDelay1, meaning that swap2 + // has higher priority than swap1. + require.Equal(t, initialDelay2, <-tickSignal) + + // Replace the logger in the batch with wrappedLogger to watch messages. + batch1 := getOnlyBatch(t, ctx, batcher) + testLogger := &wrappedLogger{ + Logger: batch1.log(), + } + batch1.setLog(testLogger) + + // Wait for publishDelay. + now = now.Add(publishDelay) + testClock.SetTime(now) + + // Wait for batch publishing to be skipped, because initialDelay2 + // has not ended. + require.EventuallyWithT(t, func(c *assert.CollectT) { + testLogger.mu.Lock() + defer testLogger.mu.Unlock() + + assert.Contains(c, testLogger.debugMessages, stillWaitingMsg) + }, test.Timeout, eventuallyCheckFrequency) + + // Wait for initialDelay2. + now = now.Add(initialDelay2 - publishDelay) + testClock.SetTime(now) + + // It should subscribe for publishDelay now. + require.Equal(t, publishDelay, <-tickSignal) + + // Wait for publishDelay. + now = now.Add(publishDelay) + testClock.SetTime(now) + + // Wait for tx to be published. + tx := <-lnd.TxPublishChannel + require.Len(t, tx.TxIn, 2) + + // Now make the batcher quit by canceling the context. + cancel() + wg.Wait() + + // Make sure the batcher exited without an error. + checkBatcherError(t, runErr) +} + // testMaxSweepsPerBatch tests the limit on max number of sweeps per batch. func testMaxSweepsPerBatch(t *testing.T, store testStore, batcherStore testBatcherStore) { @@ -4049,6 +4286,11 @@ func TestDelays(t *testing.T) { runTests(t, testDelays) } +// TestCustomDelays tests per-sweep customization in WithInitialDelay. +func TestCustomDelays(t *testing.T) { + runTests(t, testCustomDelays) +} + // TestMaxSweepsPerBatch tests the limit on max number of sweeps per batch. func TestMaxSweepsPerBatch(t *testing.T) { runTests(t, testMaxSweepsPerBatch)