sweepbatcher: fix spawning condition "unattached" spend notifiers

Previously, when handling a sweep we assumed that if a sweep status
was completed, the parent batch was also finished. However, since the
batch confirmation status depends on three on-chain confirmations, it
is possible that a spend notifier was started for a sweep of an active
batch. The notifier would fetch the parent batch from the database, but
because we incorrectly assumed that  the parent was confirmed (when it
was not), the DB call would fail with a 'no rows returned' error.
This failure would cause the sweep to fail and the sweep batcher to
stop, resulting in a permanent failure state.

loopdb/sqlc/queries/batch.sql

@@ -73,11 +73,7 @@ FROM
 JOIN
         sweeps ON sweep_batches.id = sweeps.batch_id
 WHERE
-        sweeps.swap_hash = $1
-AND
-        sweeps.completed = TRUE
-AND
-        sweep_batches.confirmed = TRUE;
+        sweeps.swap_hash = $1;
 
 -- name: GetBatchSweptAmount :one
 SELECT

sweepbatcher/store.go

@@ -196,10 +196,6 @@ func (s *SQLStore) GetParentBatch(ctx context.Context, swapHash lntypes.Hash) (
 		return nil, err
 	}
 
-	if err != nil {
-		return nil, err
-	}
-
 	return convertBatchRow(batch), nil
 }
 

sweepbatcher/sweep_batcher.go

@@ -418,7 +418,24 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
 	// can't attach its notifier to the batch as that is no longer running.
 	// Instead we directly detect and return the spend here.
 	if completed && *notifier != (SpendNotifier{}) {
-		return b.monitorSpendAndNotify(ctx, sweep, notifier)
+		// Verify that the parent batch is confirmed. Note that a batch
+		// is only considered confirmed after it has received three
+		// on-chain confirmations to prevent issues caused by reorgs.
+		parentBatch, err := b.store.GetParentBatch(ctx, sweep.swapHash)
+		if err != nil {
+			log.Errorf("unable to get parent batch for sweep %x: "+
+				"%v", sweep.swapHash[:6], err)
+
+			return err
+		}
+
+		// The parent batch is indeed confirmed, meaning it is complete
+		// and we won't be able to attach this sweep to it.
+		if parentBatch.State == batchConfirmed {
+			return b.monitorSpendAndNotify(
+				ctx, sweep, parentBatch.ID, notifier,
+			)
+		}
 	}
 
 	sweep.notifier = notifier
@@ -688,19 +705,13 @@ func (b *Batcher) FetchUnconfirmedBatches(ctx context.Context) ([]*batch,
 // monitorSpendAndNotify monitors the spend of a specific outpoint and writes
 // the response back to the response channel.
 func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
-	notifier *SpendNotifier) error {
+	parentBatchID int32, notifier *SpendNotifier) error {
 
 	spendCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	// First get the batch that completed the sweep.
-	parentBatch, err := b.store.GetParentBatch(ctx, sweep.swapHash)
-	if err != nil {
-		return err
-	}
-
 	// Then we get the total amount that was swept by the batch.
-	totalSwept, err := b.store.TotalSweptAmount(ctx, parentBatch.ID)
+	totalSwept, err := b.store.TotalSweptAmount(ctx, parentBatchID)
 	if err != nil {
 		return err
 	}