Merge pull request #755 from starius/sweepbatcher-changes

sweepbatcher: small refactorings

sweepbatcher/store.go

@@ -38,7 +38,7 @@ type BaseDB interface {
 	GetSwapUpdates(ctx context.Context, swapHash []byte) (
 		[]sqlc.SwapUpdate, error)
 
-	// FetchUnconfirmedSweepBatches fetches all the batches from the
+	// GetUnconfirmedBatches fetches all the batches from the
 	// database that are not in a confirmed state.
 	GetUnconfirmedBatches(ctx context.Context) ([]sqlc.SweepBatch, error)
 

sweepbatcher/store_mock.go

@@ -23,7 +23,7 @@ func NewStoreMock() *StoreMock {
 	}
 }
 
-// FetchUnconfirmedBatches fetches all the loop out sweep batches from the
+// FetchUnconfirmedSweepBatches fetches all the loop out sweep batches from the
 // database that are not in a confirmed state.
 func (s *StoreMock) FetchUnconfirmedSweepBatches(ctx context.Context) (
 	[]*dbBatch, error) {

sweepbatcher/sweep_batch.go

@@ -215,7 +215,7 @@ type batch struct {
 	// signerClient is the signer client used to sign the batch transaction.
 	signerClient lndclient.SignerClient
 
-	// muSig2Kit includes all the required functionality to collect
+	// muSig2SignSweep includes all the required functionality to collect
 	// and verify signatures by the swap server in order to cooperatively
 	// sweep funds.
 	muSig2SignSweep MuSig2SignSweep

sweepbatcher/sweep_batcher.go

@@ -82,6 +82,14 @@ type BatcherStore interface {
 	TotalSweptAmount(ctx context.Context, id int32) (btcutil.Amount, error)
 }
 
+// LoopOutFetcher is used to load LoopOut swaps from the database.
+// It is implemented by loopdb.SwapStore.
+type LoopOutFetcher interface {
+	// FetchLoopOutSwap returns the loop out swap with the given hash.
+	FetchLoopOutSwap(ctx context.Context,
+		hash lntypes.Hash) (*loopdb.LoopOut, error)
+}
+
 // MuSig2SignSweep is a function that can be used to sign a sweep transaction
 // cooperatively with the swap server.
 type MuSig2SignSweep func(ctx context.Context,
@@ -183,9 +191,8 @@ type Batcher struct {
 	// batcher and the batches.
 	store BatcherStore
 
-	// swapStore includes all the database interactions that are needed for
-	// interacting with swaps.
-	swapStore loopdb.SwapStore
+	// swapStore is used to load LoopOut swaps from the database.
+	swapStore LoopOutFetcher
 
 	// wg is a waitgroup that is used to wait for all the goroutines to
 	// exit.
@@ -197,7 +204,7 @@ func NewBatcher(wallet lndclient.WalletKitClient,
 	chainNotifier lndclient.ChainNotifierClient,
 	signerClient lndclient.SignerClient, musig2ServerSigner MuSig2SignSweep,
 	verifySchnorrSig VerifySchnorrSig, chainparams *chaincfg.Params,
-	store BatcherStore, swapStore loopdb.SwapStore) *Batcher {
+	store BatcherStore, swapStore LoopOutFetcher) *Batcher {
 
 	return &Batcher{
 		batches:          make(map[int32]*batch),

sweepbatcher/sweep_batcher_test.go

@@ -98,14 +98,14 @@ func TestSweepBatcherBatchCreation(t *testing.T) {
 	store.AssertLoopOutStored()
 
 	// Deliver sweep request to batcher.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// Since a batch was created we check that it registered for its primary
 	// sweep's spend.
 	<-lnd.RegisterSpendChannel
 
 	// Insert the same swap twice, this should be a noop.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// Once batcher receives sweep request it will eventually spin up a
 	// batch.
@@ -137,7 +137,7 @@ func TestSweepBatcherBatchCreation(t *testing.T) {
 	require.NoError(t, err)
 	store.AssertLoopOutStored()
 
-	batcher.sweepReqs <- sweepReq2
+	require.NoError(t, batcher.AddSweep(&sweepReq2))
 
 	// Batcher should not create a second batch as timeout distance is small
 	// enough.
@@ -169,7 +169,7 @@ func TestSweepBatcherBatchCreation(t *testing.T) {
 	require.NoError(t, err)
 	store.AssertLoopOutStored()
 
-	batcher.sweepReqs <- sweepReq3
+	require.NoError(t, batcher.AddSweep(&sweepReq3))
 
 	// Batcher should create a second batch as timeout distance is greater
 	// than the threshold
@@ -251,7 +251,7 @@ func TestSweepBatcherSimpleLifecycle(t *testing.T) {
 	store.AssertLoopOutStored()
 
 	// Deliver sweep request to batcher.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// Eventually request will be consumed and a new batch will spin up.
 	require.Eventually(t, func() bool {
@@ -435,15 +435,15 @@ func TestSweepBatcherSweepReentry(t *testing.T) {
 	store.AssertLoopOutStored()
 
 	// Feed the sweeps to the batcher.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// After inserting the primary (first) sweep, a spend monitor should be
 	// registered.
 	<-lnd.RegisterSpendChannel
 
-	batcher.sweepReqs <- sweepReq2
+	require.NoError(t, batcher.AddSweep(&sweepReq2))
 
-	batcher.sweepReqs <- sweepReq3
+	require.NoError(t, batcher.AddSweep(&sweepReq3))
 
 	// Batcher should create a batch for the sweeps.
 	require.Eventually(t, func() bool {
@@ -593,7 +593,7 @@ func TestSweepBatcherNonWalletAddr(t *testing.T) {
 	store.AssertLoopOutStored()
 
 	// Deliver sweep request to batcher.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// Once batcher receives sweep request it will eventually spin up a
 	// batch.
@@ -606,7 +606,7 @@ func TestSweepBatcherNonWalletAddr(t *testing.T) {
 	<-lnd.RegisterSpendChannel
 
 	// Insert the same swap twice, this should be a noop.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// Create a second sweep request that has a timeout distance less than
 	// our configured threshold.
@@ -633,7 +633,7 @@ func TestSweepBatcherNonWalletAddr(t *testing.T) {
 	require.NoError(t, err)
 	store.AssertLoopOutStored()
 
-	batcher.sweepReqs <- sweepReq2
+	require.NoError(t, batcher.AddSweep(&sweepReq2))
 
 	// Batcher should create a second batch as first batch is a non wallet
 	// addr batch.
@@ -670,7 +670,7 @@ func TestSweepBatcherNonWalletAddr(t *testing.T) {
 	require.NoError(t, err)
 	store.AssertLoopOutStored()
 
-	batcher.sweepReqs <- sweepReq3
+	require.NoError(t, batcher.AddSweep(&sweepReq3))
 
 	// Batcher should create a new batch as timeout distance is greater than
 	// the threshold
@@ -879,7 +879,7 @@ func TestSweepBatcherComposite(t *testing.T) {
 	store.AssertLoopOutStored()
 
 	// Deliver sweep request to batcher.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
 	// Once batcher receives sweep request it will eventually spin up a
 	// batch.
@@ -892,9 +892,9 @@ func TestSweepBatcherComposite(t *testing.T) {
 	<-lnd.RegisterSpendChannel
 
 	// Insert the same swap twice, this should be a noop.
-	batcher.sweepReqs <- sweepReq1
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
 
-	batcher.sweepReqs <- sweepReq2
+	require.NoError(t, batcher.AddSweep(&sweepReq2))
 
 	// Batcher should not create a second batch as timeout distance is small
 	// enough.
@@ -902,7 +902,7 @@ func TestSweepBatcherComposite(t *testing.T) {
 		return len(batcher.batches) == 1
 	}, test.Timeout, eventuallyCheckFrequency)
 
-	batcher.sweepReqs <- sweepReq3
+	require.NoError(t, batcher.AddSweep(&sweepReq3))
 
 	// Batcher should create a second batch as this sweep pays to a non
 	// wallet address.
@@ -914,7 +914,7 @@ func TestSweepBatcherComposite(t *testing.T) {
 	// sweep's spend.
 	<-lnd.RegisterSpendChannel
 
-	batcher.sweepReqs <- sweepReq4
+	require.NoError(t, batcher.AddSweep(&sweepReq4))
 
 	// Batcher should create a third batch as timeout distance is greater
 	// than the threshold.
@@ -926,7 +926,7 @@ func TestSweepBatcherComposite(t *testing.T) {
 	// sweep's spend.
 	<-lnd.RegisterSpendChannel
 
-	batcher.sweepReqs <- sweepReq5
+	require.NoError(t, batcher.AddSweep(&sweepReq5))
 
 	// Batcher should not create a fourth batch as timeout distance is small
 	// enough for it to join the last batch.
@@ -934,7 +934,7 @@ func TestSweepBatcherComposite(t *testing.T) {
 		return len(batcher.batches) == 3
 	}, test.Timeout, eventuallyCheckFrequency)
 
-	batcher.sweepReqs <- sweepReq6
+	require.NoError(t, batcher.AddSweep(&sweepReq6))
 
 	// Batcher should create a fourth batch as this sweep pays to a non
 	// wallet address.
@@ -1084,7 +1084,7 @@ func TestRestoringEmptyBatch(t *testing.T) {
 	store.AssertLoopOutStored()
 
 	// Deliver sweep request to batcher.
-	batcher.sweepReqs <- sweepReq
+	require.NoError(t, batcher.AddSweep(&sweepReq))
 
 	// Since a batch was created we check that it registered for its primary
 	// sweep's spend.