sweepbatcher: fix reorg detection

The reorg channel is now passed to RegisterSpendNtfn, and waiting for spend
notifications remains active even after the transaction receives its first
confirmation. The dedicated goroutine previously used to wait for the spend
is no longer needed, as we now handle both spend and potential reorg events
in the main event loop while the batch is running.

Following this change, RegisterConfirmationsNtfn runs without a reorg channel,
as it would only detect deep reorgs that undo the final confirmation - something
we can't handle anyway. We can't track fully confirmed swaps indefinitely to
guard against such rare reorgs; instead, we can mitigate the risk by increasing
the required confirmation depth.

Author: starius

sweepbatcher/sweep_batch.go

@@ -232,6 +232,10 @@ type batch struct {
 	// spendChan is the channel over which spend notifications are received.
 	spendChan chan *chainntnfs.SpendDetail
 
+	// spendErrChan is the channel over which spend notifier errors are
+	// received.
+	spendErrChan chan error
+
 	// confChan is the channel over which confirmation notifications are
 	// received.
 	confChan chan *chainntnfs.TxConfirmation
@@ -378,9 +382,7 @@ func NewBatch(cfg batchConfig, bk batchKit) *batch {
 		id:                  -1,
 		state:               Open,
 		sweeps:              make(map[wire.OutPoint]sweep),
-		spendChan:           make(chan *chainntnfs.SpendDetail),
 		confChan:            make(chan *chainntnfs.TxConfirmation, 1),
-		reorgChan:           make(chan struct{}, 1),
 		testReqs:            make(chan *testRequest),
 		errChan:             make(chan error, 1),
 		callEnter:           make(chan struct{}),
@@ -423,9 +425,7 @@ func NewBatchFromDB(cfg batchConfig, bk batchKit) (*batch, error) {
 		state:               bk.state,
 		primarySweepID:      bk.primaryID,
 		sweeps:              bk.sweeps,
-		spendChan:           make(chan *chainntnfs.SpendDetail),
 		confChan:            make(chan *chainntnfs.TxConfirmation, 1),
-		reorgChan:           make(chan struct{}, 1),
 		testReqs:            make(chan *testRequest),
 		errChan:             make(chan error, 1),
 		callEnter:           make(chan struct{}),
@@ -979,23 +979,26 @@ func (b *batch) Run(ctx context.Context) error {
 				return fmt.Errorf("handleSpend error: %w", err)
 			}
 
+		case err := <-b.spendErrChan:
+			b.writeToSpendErrChan(ctx, err)
+
+			return fmt.Errorf("spend notifier failed: %w", err)
+
 		case conf := <-b.confChan:
 			if err := b.handleConf(runCtx, conf); err != nil {
 				return fmt.Errorf("handleConf error: %w", err)
 			}
 
 			return nil
 
+		// A re-org has been detected. We set the batch state back to
+		// open since our batch transaction is no longer present in any
+		// block. We can accept more sweeps and try to publish.
 		case <-b.reorgChan:
 			b.state = Open
 			b.Warnf("reorg detected, batch is able to " +
 				"accept new sweeps")
 
-			err := b.monitorSpend(ctx, b.sweeps[b.primarySweepID])
-			if err != nil {
-				return fmt.Errorf("monitorSpend error: %w", err)
-			}
-
 		case testReq := <-b.testReqs:
 			testReq.handler()
 			close(testReq.quit)
@@ -1812,44 +1815,31 @@ func (b *batch) updateRbfRate(ctx context.Context) error {
 // of the batch transaction gets confirmed, due to the uncertainty of RBF
 // replacements and network propagation, we can always detect the transaction.
 func (b *batch) monitorSpend(ctx context.Context, primarySweep sweep) error {
-	spendCtx, cancel := context.WithCancel(ctx)
+	if b.spendChan != nil || b.spendErrChan != nil || b.reorgChan != nil {
+		return fmt.Errorf("an attempt to run monitorSpend multiple " +
+			"times per batch")
+	}
 
-	spendChan, spendErr, err := b.chainNotifier.RegisterSpendNtfn(
-		spendCtx, &primarySweep.outpoint, primarySweep.htlc.PkScript,
+	reorgChan := make(chan struct{}, 1)
+
+	spendChan, spendErrChan, err := b.chainNotifier.RegisterSpendNtfn(
+		ctx, &primarySweep.outpoint, primarySweep.htlc.PkScript,
 		primarySweep.initiationHeight,
+		lndclient.WithReOrgChan(reorgChan),
 	)
 	if err != nil {
-		cancel()
-
-		return err
+		return fmt.Errorf("failed to register spend notifier for "+
+			"primary sweep %v, pkscript %x, height %d: %w",
+			primarySweep.outpoint, primarySweep.htlc.PkScript,
+			primarySweep.initiationHeight, err)
 	}
 
-	b.wg.Add(1)
-	go func() {
-		defer cancel()
-		defer b.wg.Done()
+	b.Infof("monitoring spend for outpoint %s",
+		primarySweep.outpoint.String())
 
-		b.Infof("monitoring spend for outpoint %s",
-			primarySweep.outpoint.String())
-
-		select {
-		case spend := <-spendChan:
-			select {
-			case b.spendChan <- spend:
-
-			case <-ctx.Done():
-			}
-
-		case err := <-spendErr:
-			b.writeToSpendErrChan(ctx, err)
-
-			b.writeToErrChan(
-				fmt.Errorf("spend error: %w", err),
-			)
-
-		case <-ctx.Done():
-		}
-	}()
+	b.spendChan = spendChan
+	b.spendErrChan = spendErrChan
+	b.reorgChan = reorgChan
 
 	return nil
 }
@@ -1862,14 +1852,11 @@ func (b *batch) monitorConfirmations(ctx context.Context) error {
 		return fmt.Errorf("can't find primarySweep")
 	}
 
-	reorgChan := make(chan struct{})
-
 	confCtx, cancel := context.WithCancel(ctx)
 
 	confChan, errChan, err := b.chainNotifier.RegisterConfirmationsNtfn(
 		confCtx, b.batchTxid, b.batchPkScript, batchConfHeight,
 		primarySweep.initiationHeight,
-		lndclient.WithReOrgChan(reorgChan),
 	)
 	if err != nil {
 		cancel()
@@ -1895,18 +1882,6 @@ func (b *batch) monitorConfirmations(ctx context.Context) error {
 			b.writeToErrChan(fmt.Errorf("confirmations "+
 				"monitoring error: %w", err))
 
-		case <-reorgChan:
-			// A re-org has been detected. We set the batch
-			// state back to open since our batch
-			// transaction is no longer present in any
-			// block. We can accept more sweeps and try to
-			// publish new transactions, at this point we
-			// need to monitor again for a new spend.
-			select {
-			case b.reorgChan <- struct{}{}:
-			case <-ctx.Done():
-			}
-
 		case <-ctx.Done():
 		}
 	}()
@@ -2395,12 +2370,6 @@ func (b *batch) writeToErrChan(err error) {
 
 // writeToSpendErrChan sends an error to spend error channels of all the sweeps.
 func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) {
-	done, err := b.scheduleNextCall()
-	if err != nil {
-		done()
-
-		return
-	}
 	notifiers := make([]*SpendNotifier, 0, len(b.sweeps))
 	for _, s := range b.sweeps {
 		// If the sweep's notifier is empty then this means that a swap
@@ -2412,7 +2381,6 @@ func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) {
 
 		notifiers = append(notifiers, s.notifier)
 	}
-	done()
 
 	for _, notifier := range notifiers {
 		select {