Merge pull request #789 from bhandras/sweep-batcher-db-parent-batch-fixup

sweepbatcher: fix spawning condition "unattached" spend notifiers

loopdb/sqlc/batch.sql.go

@@ -102,10 +102,6 @@ JOIN
         sweeps ON sweep_batches.id = sweeps.batch_id
 WHERE
         sweeps.swap_hash = $1
-AND
-        sweeps.completed = TRUE
-AND
-        sweep_batches.confirmed = TRUE
 `
 
 func (q *Queries) GetParentBatch(ctx context.Context, swapHash []byte) (SweepBatch, error) {

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
 	}