From d38b7c55a7bf2aa547c70d536e97e31bf3c25287 Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Wed, 29 May 2024 15:03:37 -0300 Subject: [PATCH 1/3] sweepbatcher/StoreMock: load LoopOut from loopdb Method sweepbatcher.Store.FetchBatchSweeps (implementation using real DB) runs JOIN query to load LoopOut from swaps table. Now the mock does the same. It is needed to test store and load scenarios in tests. --- loopout_test.go | 6 +++--- sweepbatcher/store_mock.go | 29 ++++++++++++++++++++++------- sweepbatcher/sweep_batcher_test.go | 14 +++++++------- testcontext_test.go | 2 +- 4 files changed, 33 insertions(+), 18 deletions(-) diff --git a/loopout_test.go b/loopout_test.go index dfbeb620..7e5045cb 100644 --- a/loopout_test.go +++ b/loopout_test.go @@ -296,7 +296,7 @@ func testCustomSweepConfTarget(t *testing.T) { errChan := make(chan error, 2) - batcherStore := sweepbatcher.NewStoreMock() + batcherStore := sweepbatcher.NewStoreMock(cfg.store) batcher := sweepbatcher.NewBatcher( lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, @@ -529,7 +529,7 @@ func testPreimagePush(t *testing.T) { errChan := make(chan error, 2) - batcherStore := sweepbatcher.NewStoreMock() + batcherStore := sweepbatcher.NewStoreMock(cfg.store) batcher := sweepbatcher.NewBatcher( lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, @@ -950,7 +950,7 @@ func TestLoopOutMuSig2Sweep(t *testing.T) { errChan := make(chan error, 2) - batcherStore := sweepbatcher.NewStoreMock() + batcherStore := sweepbatcher.NewStoreMock(cfg.store) batcher := sweepbatcher.NewBatcher( lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, diff --git a/sweepbatcher/store_mock.go b/sweepbatcher/store_mock.go index 57cdd34b..96d5dcf4 100644 --- a/sweepbatcher/store_mock.go +++ b/sweepbatcher/store_mock.go @@ -3,6 +3,7 @@ package sweepbatcher import ( "context" "errors" + "fmt" "sort" "github.com/btcsuite/btcd/btcutil" @@ -11,15 +12,17 @@ import ( // StoreMock implements a mock client swap store. type StoreMock struct { - batches map[int32]dbBatch - sweeps map[lntypes.Hash]dbSweep + batches map[int32]dbBatch + sweeps map[lntypes.Hash]dbSweep + swapStore LoopOutFetcher } // NewStoreMock instantiates a new mock store. -func NewStoreMock() *StoreMock { +func NewStoreMock(swapStore LoopOutFetcher) *StoreMock { return &StoreMock{ - batches: make(map[int32]dbBatch), - sweeps: make(map[lntypes.Hash]dbSweep), + batches: make(map[int32]dbBatch), + sweeps: make(map[lntypes.Hash]dbSweep), + swapStore: swapStore, } } @@ -90,9 +93,21 @@ func (s *StoreMock) FetchBatchSweeps(ctx context.Context, result := []*dbSweep{} for _, sweep := range s.sweeps { sweep := sweep - if sweep.BatchID == id { - result = append(result, &sweep) + if sweep.BatchID != id { + continue } + + // Load swap from loopdb. + swap, err := s.swapStore.FetchLoopOutSwap( + ctx, sweep.SwapHash, + ) + if err != nil { + return nil, fmt.Errorf("failed to fetch swap "+ + "for SwapHash=%v", sweep.SwapHash) + } + sweep.LoopOut = swap + + result = append(result, &sweep) } sort.Slice(result, func(i, j int) bool { diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index afa7ab97..67528c71 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/sweepbatcher/sweep_batcher_test.go @@ -64,7 +64,7 @@ func TestSweepBatcherBatchCreation(t *testing.T) { store := loopdb.NewStoreMock(t) - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) @@ -218,7 +218,7 @@ func TestSweepBatcherSimpleLifecycle(t *testing.T) { store := loopdb.NewStoreMock(t) - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) @@ -355,7 +355,7 @@ func TestSweepBatcherSweepReentry(t *testing.T) { store := loopdb.NewStoreMock(t) - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) @@ -562,7 +562,7 @@ func TestSweepBatcherNonWalletAddr(t *testing.T) { store := loopdb.NewStoreMock(t) - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) @@ -727,7 +727,7 @@ func TestSweepBatcherComposite(t *testing.T) { store := loopdb.NewStoreMock(t) - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) @@ -1044,7 +1044,7 @@ func TestRestoringEmptyBatch(t *testing.T) { store := loopdb.NewStoreMock(t) - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) _, err := batcherStore.InsertSweepBatch(ctx, &dbBatch{}) require.NoError(t, err) @@ -1158,7 +1158,7 @@ func TestHandleSweepTwice(t *testing.T) { store := newLoopStoreMock() - batcherStore := NewStoreMock() + batcherStore := NewStoreMock(store) batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) diff --git a/testcontext_test.go b/testcontext_test.go index 423eb31d..9f8f120d 100644 --- a/testcontext_test.go +++ b/testcontext_test.go @@ -77,7 +77,7 @@ func newSwapClient(config *clientConfig) *Client { lndServices := config.LndServices - batcherStore := sweepbatcher.NewStoreMock() + batcherStore := sweepbatcher.NewStoreMock(config.Store) batcher := sweepbatcher.NewBatcher( config.LndServices.WalletKit, config.LndServices.ChainNotifier, From 40ad1ce609fb710be4520c2d9a2221dc2ceadb4e Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Wed, 29 May 2024 15:07:32 -0300 Subject: [PATCH 2/3] sweepbatcher: load from DB preserves confTarget It used to be set to default (defaultBatchConfTarget = 12) which could in theory affect fee rate if updateRbfRate() and publish() were not called before the batch was saved. (Unlikely scenario.) --- sweepbatcher/sweep_batch.go | 19 ++++- sweepbatcher/sweep_batcher.go | 5 +- sweepbatcher/sweep_batcher_test.go | 120 ++++++++++++++++++++++++++++- 3 files changed, 139 insertions(+), 5 deletions(-) diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index 0aa5b4c3..da27e308 100644 --- a/sweepbatcher/sweep_batch.go +++ b/sweepbatcher/sweep_batch.go @@ -316,7 +316,22 @@ func NewBatch(cfg batchConfig, bk batchKit) *batch { } // NewBatchFromDB creates a new batch that already existed in storage. -func NewBatchFromDB(cfg batchConfig, bk batchKit) *batch { +func NewBatchFromDB(cfg batchConfig, bk batchKit) (*batch, error) { + // Make sure the batch is not empty. + if len(bk.sweeps) == 0 { + // This should never happen, as this precondition is already + // ensured in spinUpBatchFromDB. + return nil, fmt.Errorf("empty batch is not allowed") + } + + // Assign batchConfTarget to primary sweep's confTarget. + for _, sweep := range bk.sweeps { + if sweep.swapHash == bk.primaryID { + cfg.batchConfTarget = sweep.confTarget + break + } + } + return &batch{ id: bk.id, state: bk.state, @@ -343,7 +358,7 @@ func NewBatchFromDB(cfg batchConfig, bk batchKit) *batch { store: bk.store, log: bk.log, cfg: &cfg, - } + }, nil } // addSweep tries to add a sweep to the batch. If this is the first sweep being diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go index bc69d421..4c3a5330 100644 --- a/sweepbatcher/sweep_batcher.go +++ b/sweepbatcher/sweep_batcher.go @@ -491,7 +491,10 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error { batchConfTarget: defaultBatchConfTarget, } - newBatch := NewBatchFromDB(cfg, batchKit) + newBatch, err := NewBatchFromDB(cfg, batchKit) + if err != nil { + return fmt.Errorf("failed in NewBatchFromDB: %w", err) + } // We add the batch to our map of batches and start it. b.batches[batch.id] = newBatch diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index 67528c71..4b73cfaa 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/sweepbatcher/sweep_batcher_test.go @@ -1109,7 +1109,7 @@ func TestRestoringEmptyBatch(t *testing.T) { require.NoError(t, err) require.Len(t, batches, 1) - // Now make it quit by canceling the context. + // Now make the batcher quit by canceling the context. cancel() wg.Wait() @@ -1297,9 +1297,125 @@ func TestHandleSweepTwice(t *testing.T) { require.Equal(t, 1, len(batcher.batches[0].sweeps)) require.Equal(t, 1, len(batcher.batches[1].sweeps)) - // Now make it quit by canceling the context. + // Now make the batcher quit by canceling the context. cancel() wg.Wait() checkBatcherError(t, runErr) } + +// TestRestoringPreservesConfTarget tests that after the batch is written to DB +// and loaded back, its batchConfTarget value is preserved. +func TestRestoringPreservesConfTarget(t *testing.T) { + defer test.Guard(t)() + + lnd := test.NewMockLnd() + ctx, cancel := context.WithCancel(context.Background()) + + store := loopdb.NewStoreMock(t) + + batcherStore := NewStoreMock(store) + + batcher := NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, + testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) + + 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. + sweepReq := SweepRequest{ + SwapHash: lntypes.Hash{1, 1, 1}, + Value: 111, + Outpoint: wire.OutPoint{ + Hash: chainhash.Hash{1, 1}, + Index: 1, + }, + Notifier: &dummyNotifier, + } + + swap := &loopdb.LoopOutContract{ + SwapContract: loopdb.SwapContract{ + CltvExpiry: 111, + AmountRequested: 111, + }, + + SwapInvoice: swapInvoice, + SweepConfTarget: 123, + } + + err := store.CreateLoopOut(ctx, sweepReq.SwapHash, swap) + require.NoError(t, err) + store.AssertLoopOutStored() + + // Deliver sweep request to batcher. + require.NoError(t, batcher.AddSweep(&sweepReq)) + + // Since a batch was created we check that it registered for its primary + // sweep's spend. + <-lnd.RegisterSpendChannel + + // Once batcher receives sweep request it will eventually spin up a + // batch. + require.Eventually(t, func() bool { + // Make sure that the sweep was stored and we have exactly one + // active batch, with one sweep and proper batchConfTarget. + return batcherStore.AssertSweepStored(sweepReq.SwapHash) && + len(batcher.batches) == 1 && + len(batcher.batches[0].sweeps) == 1 && + batcher.batches[0].cfg.batchConfTarget == 123 + }, test.Timeout, eventuallyCheckFrequency) + + // Make sure we have stored the batch. + batches, err := batcherStore.FetchUnconfirmedSweepBatches(ctx) + require.NoError(t, err) + require.Len(t, batches, 1) + + // Now make the batcher quit by canceling the context. + cancel() + wg.Wait() + + // Make sure the batcher exited without an error. + checkBatcherError(t, runErr) + + // Now launch it again. + batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer, + testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store) + ctx, cancel = context.WithCancel(context.Background()) + wg.Add(1) + go func() { + defer wg.Done() + runErr = batcher.Run(ctx) + }() + + // Wait for the batcher to be initialized. + <-batcher.initDone + + // Wait for batch to load. + require.Eventually(t, func() bool { + return batcherStore.AssertSweepStored(sweepReq.SwapHash) && + len(batcher.batches) == 1 && + len(batcher.batches[0].sweeps) == 1 + }, test.Timeout, eventuallyCheckFrequency) + + // Make sure batchConfTarget was preserved. + require.Equal(t, 123, int(batcher.batches[0].cfg.batchConfTarget)) + + // Expect registration for spend notification. + <-lnd.RegisterSpendChannel + + // Now make the batcher quit by canceling the context. + cancel() + wg.Wait() + + // Make sure the batcher exited without an error. + checkBatcherError(t, runErr) +} From 87fb185a9b27910f63fe2ed9bfeb4f2ded64e64c Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Wed, 29 May 2024 15:13:48 -0300 Subject: [PATCH 3/3] sweepbatcher: remove const defaultBatchConfTarget It is always overwritten with primary sweep's confTarget. Print a warning if batchConfTarget is 0 in updateRbfRate. See https://github.com/lightninglabs/loop/pull/754#discussion_r1613514363 --- sweepbatcher/sweep_batch.go | 8 ++++---- sweepbatcher/sweep_batcher.go | 2 -- 2 files changed, 4 insertions(+), 6 deletions(-) diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index da27e308..b0d8c04b 100644 --- a/sweepbatcher/sweep_batch.go +++ b/sweepbatcher/sweep_batch.go @@ -35,10 +35,6 @@ const ( // fee rate is increased when an rbf is attempted. defaultFeeRateStep = chainfee.SatPerKWeight(100) - // defaultBatchConfTarget is the default confirmation target of the - // batch transaction. - defaultBatchConfTarget = 12 - // batchConfHeight is the default confirmation height of the batch // transaction. batchConfHeight = 3 @@ -1059,6 +1055,10 @@ func (b *batch) updateRbfRate(ctx context.Context) error { // If the feeRate is unset then we never published before, so we // retrieve the fee estimate from our wallet. if b.rbfCache.FeeRate == 0 { + if b.cfg.batchConfTarget == 0 { + b.log.Warnf("updateRbfRate called with zero " + + "batchConfTarget") + } b.log.Infof("initializing rbf fee rate for conf target=%v", b.cfg.batchConfTarget) rate, err := b.wallet.EstimateFeeRate( diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go index 4c3a5330..749da451 100644 --- a/sweepbatcher/sweep_batcher.go +++ b/sweepbatcher/sweep_batcher.go @@ -378,7 +378,6 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep, func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) { cfg := batchConfig{ maxTimeoutDistance: defaultMaxTimeoutDistance, - batchConfTarget: defaultBatchConfTarget, } switch b.chainParams { @@ -488,7 +487,6 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error { cfg := batchConfig{ maxTimeoutDistance: batch.cfg.maxTimeoutDistance, - batchConfTarget: defaultBatchConfTarget, } newBatch, err := NewBatchFromDB(cfg, batchKit)