diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go index 8816cd16..74fa9d25 100644 --- a/sweepbatcher/sweep_batcher.go +++ b/sweepbatcher/sweep_batcher.go @@ -153,6 +153,10 @@ type Batcher struct { // quit signals that the batch must stop. quit chan struct{} + // initDone is a channel that is closed when the batcher has been + // initialized. + initDone chan struct{} + // wallet is the wallet kit client that is used by batches. wallet lndclient.WalletKitClient @@ -200,6 +204,7 @@ func NewBatcher(wallet lndclient.WalletKitClient, sweepReqs: make(chan SweepRequest), errChan: make(chan error, 1), quit: make(chan struct{}), + initDone: make(chan struct{}), wallet: wallet, chainNotifier: chainNotifier, signerClient: signerClient, @@ -239,6 +244,9 @@ func (b *Batcher) Run(ctx context.Context) error { } } + // Signal that the batcher has been initialized. + close(b.initDone) + for { select { case sweepReq := <-b.sweepReqs: diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index 4ddbb656..92067b03 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/sweepbatcher/sweep_batcher_test.go @@ -3,6 +3,7 @@ package sweepbatcher import ( "context" "errors" + "sync" "testing" "time" @@ -1028,3 +1029,85 @@ func TestGetFeePortionForSweep(t *testing.T) { }) } } + +// TestRestoringEmptyBatch tests that the batcher can be restored with an empty +// batch. +func TestRestoringEmptyBatch(t *testing.T) { + defer test.Guard(t)() + + lnd := test.NewMockLnd() + ctx, cancel := context.WithCancel(context.Background()) + + store := loopdb.NewStoreMock(t) + + batcherStore := NewStoreMock() + _, err := batcherStore.InsertSweepBatch(ctx, &dbBatch{}) + require.NoError(t, err) + + 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, + } + + err = store.CreateLoopOut(ctx, sweepReq.SwapHash, swap) + require.NoError(t, err) + store.AssertLoopOutStored() + + // Deliver sweep request to batcher. + batcher.sweepReqs <- 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. + return batcherStore.AssertSweepStored(sweepReq.SwapHash) && + len(batcher.batches) == 1 + }, test.Timeout, eventuallyCheckFrequency) + + // Make sure we have only one batch stored (as we dropped the dormant + // one). + batches, err := batcherStore.FetchUnconfirmedSweepBatches(ctx) + require.NoError(t, err) + require.Len(t, batches, 1) + + // Now make it quit by canceling the context. + cancel() + wg.Wait() + + checkBatcherError(t, runErr) +}