Merge pull request #754 from bhandras/sweepbatcher-empty-batch-fix

sweepbatcher: do not fail on restoring empty batches

loopdb/sqlc/batch.sql.go

@@ -25,6 +25,15 @@ func (q *Queries) ConfirmBatch(ctx context.Context, id int32) error {
 	return err
 }
 
+const dropBatch = `-- name: DropBatch :exec
+DELETE FROM sweep_batches WHERE id = $1
+`
+
+func (q *Queries) DropBatch(ctx context.Context, id int32) error {
+	_, err := q.db.ExecContext(ctx, dropBatch, id)
+	return err
+}
+
 const getBatchSweeps = `-- name: GetBatchSweeps :many
 SELECT
         sweeps.id, sweeps.swap_hash, sweeps.batch_id, sweeps.outpoint_txid, sweeps.outpoint_index, sweeps.amt, sweeps.completed,

loopdb/sqlc/querier.go

@@ -11,6 +11,7 @@ import (
 type Querier interface {
 	ConfirmBatch(ctx context.Context, id int32) error
 	CreateReservation(ctx context.Context, arg CreateReservationParams) error
+	DropBatch(ctx context.Context, id int32) error
 	FetchLiquidityParams(ctx context.Context) ([]byte, error)
 	GetBatchSweeps(ctx context.Context, batchID int32) ([]GetBatchSweepsRow, error)
 	GetBatchSweptAmount(ctx context.Context, batchID int32) (int64, error)

loopdb/sqlc/queries/batch.sql

@@ -23,6 +23,9 @@ INSERT INTO sweep_batches (
         $6
 ) RETURNING id;
 
+-- name: DropBatch :exec
+DELETE FROM sweep_batches WHERE id = $1;
+
 -- name: UpdateBatch :exec
 UPDATE sweep_batches SET
         confirmed = $2,

sweepbatcher/store.go

@@ -3,6 +3,7 @@ package sweepbatcher
 import (
 	"context"
 	"database/sql"
+	"fmt"
 
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg"
@@ -46,6 +47,9 @@ type BaseDB interface {
 	InsertBatch(ctx context.Context, arg sqlc.InsertBatchParams) (
 		int32, error)
 
+	// DropBatch drops a batch from the database.
+	DropBatch(ctx context.Context, id int32) error
+
 	// UpdateBatch updates a batch in the database.
 	UpdateBatch(ctx context.Context, arg sqlc.UpdateBatchParams) error
 
@@ -108,6 +112,24 @@ func (s *SQLStore) InsertSweepBatch(ctx context.Context, batch *dbBatch) (int32,
 	return s.baseDb.InsertBatch(ctx, batchToInsertArgs(*batch))
 }
 
+// DropBatch drops a batch from the database. Note that we only use this call
+// for batches that have no sweeps and so we'd not be able to resume.
+func (s *SQLStore) DropBatch(ctx context.Context, id int32) error {
+	readOpts := loopdb.NewSqlReadOpts()
+	return s.baseDb.ExecTx(ctx, readOpts, func(tx *sqlc.Queries) error {
+		dbSweeps, err := tx.GetBatchSweeps(ctx, id)
+		if err != nil {
+			return err
+		}
+
+		if len(dbSweeps) != 0 {
+			return fmt.Errorf("cannot drop a non-empty batch")
+		}
+
+		return tx.DropBatch(ctx, id)
+	})
+}
+
 // UpdateSweepBatch updates a batch in the database.
 func (s *SQLStore) UpdateSweepBatch(ctx context.Context, batch *dbBatch) error {
 	return s.baseDb.UpdateBatch(ctx, batchToUpdateArgs(*batch))

sweepbatcher/store_mock.go

@@ -56,6 +56,12 @@ func (s *StoreMock) InsertSweepBatch(ctx context.Context,
 	return id, nil
 }
 
+// DropBatch drops a batch from the database.
+func (s *StoreMock) DropBatch(ctx context.Context, id int32) error {
+	delete(s.batches, id)
+	return nil
+}
+
 // UpdateSweepBatch updates a batch in the database.
 func (s *StoreMock) UpdateSweepBatch(ctx context.Context,
 	batch *dbBatch) error {

sweepbatcher/sweep_batch.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"math"
 	"sync"
@@ -48,7 +49,7 @@ const (
 )
 
 var (
-	ErrBatchShuttingDown = fmt.Errorf("batch shutting down")
+	ErrBatchShuttingDown = errors.New("batch shutting down")
 )
 
 // sweep stores any data related to sweeping a specific outpoint.
@@ -196,7 +197,11 @@ type batch struct {
 	// main event loop.
 	callLeave chan struct{}
 
-	// quit signals that the batch must stop.
+	// stopped signals that the batch has stopped.
+	stopped chan struct{}
+
+	// quit is owned by the parent batcher and signals that the batch must
+	// stop.
 	quit chan struct{}
 
 	// wallet is the wallet client used to create and publish the batch
@@ -260,6 +265,7 @@ type batchKit struct {
 	purger           Purger
 	store            BatcherStore
 	log              btclog.Logger
+	quit             chan struct{}
 }
 
 // scheduleNextCall schedules the next call to the batch handler's main event
@@ -269,6 +275,9 @@ func (b *batch) scheduleNextCall() (func(), error) {
 	case b.callEnter <- struct{}{}:
 
 	case <-b.quit:
+		return func() {}, ErrBatcherShuttingDown
+
+	case <-b.stopped:
 		return func() {}, ErrBatchShuttingDown
 	}
 
@@ -292,7 +301,8 @@ func NewBatch(cfg batchConfig, bk batchKit) *batch {
 		errChan:          make(chan error, 1),
 		callEnter:        make(chan struct{}),
 		callLeave:        make(chan struct{}),
-		quit:             make(chan struct{}),
+		stopped:          make(chan struct{}),
+		quit:             bk.quit,
 		batchTxid:        bk.batchTxid,
 		wallet:           bk.wallet,
 		chainNotifier:    bk.chainNotifier,
@@ -319,7 +329,8 @@ func NewBatchFromDB(cfg batchConfig, bk batchKit) *batch {
 		errChan:          make(chan error, 1),
 		callEnter:        make(chan struct{}),
 		callLeave:        make(chan struct{}),
-		quit:             make(chan struct{}),
+		stopped:          make(chan struct{}),
+		quit:             bk.quit,
 		batchTxid:        bk.batchTxid,
 		batchPkScript:    bk.batchPkScript,
 		rbfCache:         bk.rbfCache,
@@ -446,7 +457,7 @@ func (b *batch) Run(ctx context.Context) error {
 	runCtx, cancel := context.WithCancel(ctx)
 	defer func() {
 		cancel()
-		close(b.quit)
+		close(b.stopped)
 		b.wg.Wait()
 	}()
 
@@ -539,6 +550,12 @@ func (b *batch) publish(ctx context.Context) error {
 		coopSuccess bool
 	)
 
+	if len(b.sweeps) == 0 {
+		b.log.Debugf("skipping publish: no sweeps in the batch")
+
+		return nil
+	}
+
 	// Run the RBF rate update.
 	err = b.updateRbfRate(ctx)
 	if err != nil {

sweepbatcher/sweep_batcher.go

@@ -2,6 +2,7 @@ package sweepbatcher
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -46,36 +47,35 @@ const (
 type BatcherStore interface {
 	// FetchUnconfirmedSweepBatches fetches all the batches from the
 	// database that are not in a confirmed state.
-	FetchUnconfirmedSweepBatches(ctx context.Context) ([]*dbBatch,
-		error)
+	FetchUnconfirmedSweepBatches(ctx context.Context) ([]*dbBatch, error)
 
 	// InsertSweepBatch inserts a batch into the database, returning the id
 	// of the inserted batch.
-	InsertSweepBatch(ctx context.Context,
-		batch *dbBatch) (int32, error)
+	InsertSweepBatch(ctx context.Context, batch *dbBatch) (int32, error)
+
+	// DropBatch drops a batch from the database. This should only be used
+	// when a batch is empty.
+	DropBatch(ctx context.Context, id int32) error
 
 	// UpdateSweepBatch updates a batch in the database.
-	UpdateSweepBatch(ctx context.Context,
-		batch *dbBatch) error
+	UpdateSweepBatch(ctx context.Context, batch *dbBatch) error
 
 	// ConfirmBatch confirms a batch by setting its state to confirmed.
 	ConfirmBatch(ctx context.Context, id int32) error
 
 	// FetchBatchSweeps fetches all the sweeps that belong to a batch.
-	FetchBatchSweeps(ctx context.Context,
-		id int32) ([]*dbSweep, error)
+	FetchBatchSweeps(ctx context.Context, id int32) ([]*dbSweep, error)
 
 	// UpsertSweep inserts a sweep into the database, or updates an existing
 	// sweep if it already exists.
 	UpsertSweep(ctx context.Context, sweep *dbSweep) error
 
 	// GetSweepStatus returns the completed status of the sweep.
-	GetSweepStatus(ctx context.Context, swapHash lntypes.Hash) (
-		bool, error)
+	GetSweepStatus(ctx context.Context, swapHash lntypes.Hash) (bool, error)
 
 	// GetParentBatch returns the parent batch of a (completed) sweep.
-	GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (
-		*dbBatch, error)
+	GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (*dbBatch,
+		error)
 
 	// TotalSweptAmount returns the total amount swept by a (confirmed)
 	// batch.
@@ -135,7 +135,7 @@ type SpendNotifier struct {
 }
 
 var (
-	ErrBatcherShuttingDown = fmt.Errorf("batcher shutting down")
+	ErrBatcherShuttingDown = errors.New("batcher shutting down")
 )
 
 // Batcher is a system that is responsible for accepting sweep requests and
@@ -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,
@@ -216,6 +221,7 @@ func (b *Batcher) Run(ctx context.Context) error {
 	runCtx, cancel := context.WithCancel(ctx)
 	defer func() {
 		cancel()
+		close(b.quit)
 
 		for _, batch := range b.batches {
 			batch.Wait()
@@ -238,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:
@@ -306,7 +315,7 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
 
 		if batch.sweepExists(sweep.swapHash) {
 			accepted, err := batch.addSweep(ctx, sweep)
-			if err != nil {
+			if err != nil && !errors.Is(err, ErrBatchShuttingDown) {
 				return err
 			}
 
@@ -321,7 +330,7 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
 	// If one of the batches accepts the sweep, we provide it to that batch.
 	for _, batch := range b.batches {
 		accepted, err := batch.addSweep(ctx, sweep)
-		if err != nil && err != ErrBatchShuttingDown {
+		if err != nil && !errors.Is(err, ErrBatchShuttingDown) {
 			return err
 		}
 
@@ -379,6 +388,7 @@ func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
 		verifySchnorrSig: b.VerifySchnorrSig,
 		purger:           b.AddSweep,
 		store:            b.store,
+		quit:             b.quit,
 	}
 
 	batch := NewBatch(cfg, batchKit)
@@ -407,23 +417,23 @@ func (b *Batcher) spinUpBatch(ctx context.Context) (*batch, error) {
 // spinUpBatchDB spins up a batch that already existed in storage, then
 // returns it.
 func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
-	cfg := batchConfig{
-		maxTimeoutDistance: batch.cfg.maxTimeoutDistance,
-		batchConfTarget:    defaultBatchConfTarget,
-	}
-
-	rbfCache := rbfCache{
-		LastHeight: batch.rbfCache.LastHeight,
-		FeeRate:    batch.rbfCache.FeeRate,
-	}
-
 	dbSweeps, err := b.store.FetchBatchSweeps(ctx, batch.id)
 	if err != nil {
 		return err
 	}
 
 	if len(dbSweeps) == 0 {
-		return fmt.Errorf("batch %d has no sweeps", batch.id)
+		log.Infof("skipping restored batch %d as it has no sweeps",
+			batch.id)
+
+		// It is safe to drop this empty batch as it has no sweeps.
+		err := b.store.DropBatch(ctx, batch.id)
+		if err != nil {
+			log.Warnf("unable to drop empty batch %d: %v",
+				batch.id, err)
+		}
+
+		return nil
 	}
 
 	primarySweep := dbSweeps[0]
@@ -439,6 +449,11 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
 		sweeps[sweep.swapHash] = *sweep
 	}
 
+	rbfCache := rbfCache{
+		LastHeight: batch.rbfCache.LastHeight,
+		FeeRate:    batch.rbfCache.FeeRate,
+	}
+
 	batchKit := batchKit{
 		id:               batch.id,
 		batchTxid:        batch.batchTxid,
@@ -456,6 +471,12 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
 		purger:           b.AddSweep,
 		store:            b.store,
 		log:              batchPrefixLogger(fmt.Sprintf("%d", batch.id)),
+		quit:             b.quit,
+	}
+
+	cfg := batchConfig{
+		maxTimeoutDistance: batch.cfg.maxTimeoutDistance,
+		batchConfTarget:    defaultBatchConfTarget,
 	}
 
 	newBatch := NewBatchFromDB(cfg, batchKit)

sweepbatcher/sweep_batcher_test.go

@@ -2,7 +2,8 @@ package sweepbatcher
 
 import (
 	"context"
-	"strings"
+	"errors"
+	"sync"
 	"testing"
 	"time"
 
@@ -43,6 +44,15 @@ var dummyNotifier = SpendNotifier{
 	QuitChan:     make(chan bool, ntfnBufferSize),
 }
 
+func checkBatcherError(t *testing.T, err error) {
+	if !errors.Is(err, context.Canceled) &&
+		!errors.Is(err, ErrBatcherShuttingDown) &&
+		!errors.Is(err, ErrBatchShuttingDown) {
+
+		require.NoError(t, err)
+	}
+}
+
 // TestSweepBatcherBatchCreation tests that sweep requests enter the expected
 // batch based on their timeout distance.
 func TestSweepBatcherBatchCreation(t *testing.T) {
@@ -60,9 +70,7 @@ func TestSweepBatcherBatchCreation(t *testing.T) {
 		testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store)
 	go func() {
 		err := batcher.Run(ctx)
-		if !strings.Contains(err.Error(), "context canceled") {
-			require.NoError(t, err)
-		}
+		checkBatcherError(t, err)
 	}()
 
 	// Create a sweep request.
@@ -215,9 +223,7 @@ func TestSweepBatcherSimpleLifecycle(t *testing.T) {
 		testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store)
 	go func() {
 		err := batcher.Run(ctx)
-		if !strings.Contains(err.Error(), "context canceled") {
-			require.NoError(t, err)
-		}
+		checkBatcherError(t, err)
 	}()
 
 	// Create a sweep request.
@@ -354,9 +360,7 @@ func TestSweepBatcherSweepReentry(t *testing.T) {
 		testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store)
 	go func() {
 		err := batcher.Run(ctx)
-		if !strings.Contains(err.Error(), "context canceled") {
-			require.NoError(t, err)
-		}
+		checkBatcherError(t, err)
 	}()
 
 	// Create some sweep requests with timeouts not too far away, in order
@@ -561,9 +565,7 @@ func TestSweepBatcherNonWalletAddr(t *testing.T) {
 		testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store)
 	go func() {
 		err := batcher.Run(ctx)
-		if !strings.Contains(err.Error(), "context canceled") {
-			require.NoError(t, err)
-		}
+		checkBatcherError(t, err)
 	}()
 
 	// Create a sweep request.
@@ -727,9 +729,7 @@ func TestSweepBatcherComposite(t *testing.T) {
 		testMuSig2SignSweep, nil, lnd.ChainParams, batcherStore, store)
 	go func() {
 		err := batcher.Run(ctx)
-		if !strings.Contains(err.Error(), "context canceled") {
-			require.NoError(t, err)
-		}
+		checkBatcherError(t, err)
 	}()
 
 	// Create a sweep request.
@@ -1029,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)
+}