sweepbatcher: do not fail on restoring empty batches

Previously storing an empty batch would make the batcher fail to start
as spinning up a restored batch assumes that there's a primary sweep
added already. As there's no point in spinning up such batch we can just
skip over it.
Furthermore we'll ensure that we won't try to ever publish an empty
batch to avoid setting the fee rate too early.

Author: bhandras

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.
@@ -539,6 +540,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"
@@ -135,7 +136,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
@@ -306,7 +307,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 +322,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 {
 			return err
 		}
 
@@ -407,23 +408,16 @@ 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)
+
+		return nil
 	}
 
 	primarySweep := dbSweeps[0]
@@ -439,6 +433,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,
@@ -458,6 +457,11 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
 		log:              batchPrefixLogger(fmt.Sprintf("%d", batch.id)),
 	}
 
+	cfg := batchConfig{
+		maxTimeoutDistance: batch.cfg.maxTimeoutDistance,
+		batchConfTarget:    defaultBatchConfTarget,
+	}
+
 	newBatch := NewBatchFromDB(cfg, batchKit)
 
 	// We add the batch to our map of batches and start it.