GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-13 12:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

sweepbatcher: customize initialDelay per sweep

Browse source 
Option WithInitialDelay now accepts a function returning initialDelay
depending on sweep's data.

This is needed to be able to wait longer for sweeps with low priority, but
still sweeping high priority sweeps soon.
This commit is contained in:
Boris Nagaev 2025-03-28 00:21:20 -03:00
parent 00063418bd
commit 472bec0be5
No known key found for this signature in database
3 changed files with 363 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

72
sweepbatcher/sweep_batch.go
View file

@ -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
}

98
sweepbatcher/sweep_batcher.go
View file

@ -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)

252
sweepbatcher/sweep_batcher_test.go
View file

@ -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)