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

Merge pull request #910 from starius/sweepbatcher-initial-delay-func

Browse source 
sweepbatcher: customize initialDelay per sweep
This commit is contained in:
Boris Nagaev 2025-04-03 09:42:18 -03:00 • committed by GitHub
commit dc7261be4a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
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)