sweepbatcher: notify caller about confirmations

Add fields ConfChan and ConfErrChan to SpendNotifier type which is a part of
SweepRequest passed to AddSweep method.

This is needed to reuse confirmation notifications on the calling side the same
way it is done for spending notifications.

Author: starius

sweepbatcher/sweep_batch.go

@@ -881,8 +881,8 @@ func (b *batch) Run(ctx context.Context) error {
 				return fmt.Errorf("handleSpend error: %w", err)
 			}
 
-		case <-b.confChan:
-			if err := b.handleConf(runCtx); err != nil {
+		case conf := <-b.confChan:
+			if err := b.handleConf(runCtx, conf); err != nil {
 				return fmt.Errorf("handleConf error: %w", err)
 			}
 
@@ -1736,6 +1736,8 @@ func (b *batch) monitorConfirmations(ctx context.Context) error {
 			}
 
 		case err := <-errChan:
+			b.writeToConfErrChan(ctx, err)
+
 			b.writeToErrChan(fmt.Errorf("confirmations "+
 				"monitoring error: %w", err))
 
@@ -1908,11 +1910,40 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error {
 
 // handleConf handles a confirmation notification. This is the final step of the
 // batch. Here we signal to the batcher that this batch was completed.
-func (b *batch) handleConf(ctx context.Context) error {
+func (b *batch) handleConf(ctx context.Context,
+	conf *chainntnfs.TxConfirmation) error {
+
 	b.Infof("confirmed in txid %s", b.batchTxid)
 	b.state = Confirmed
 
-	return b.store.ConfirmBatch(ctx, b.id)
+	if err := b.store.ConfirmBatch(ctx, b.id); err != nil {
+		return fmt.Errorf("failed to store confirmed state: %w", err)
+	}
+
+	// Send the confirmation to all the notifiers.
+	for _, s := range b.sweeps {
+		// If the sweep's notifier is empty then this means that a swap
+		// is not waiting to read an update from it, so we can skip
+		// the notification part.
+		if s.notifier == nil || s.notifier.ConfChan == nil {
+			continue
+		}
+
+		select {
+		// Try to write the confirmation to the notification
+		// channel.
+		case s.notifier.ConfChan <- conf:
+
+		// If a quit signal was provided by the swap,
+		// continue.
+		case <-s.notifier.QuitChan:
+
+		// If the context was canceled, stop.
+		case <-ctx.Done():
+		}
+	}
+
+	return nil
 }
 
 // isComplete returns true if the batch is completed. This method is used by the
@@ -2053,6 +2084,44 @@ func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) {
 	}
 }
 
+// writeToConfErrChan sends an error to confirmation error channels of all the
+// sweeps.
+func (b *batch) writeToConfErrChan(ctx context.Context, confErr 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
+		// is not waiting to read an update from it, so we can skip
+		// the notification part.
+		if s.notifier == nil || s.notifier.ConfErrChan == nil {
+			continue
+		}
+
+		notifiers = append(notifiers, s.notifier)
+	}
+	done()
+
+	for _, notifier := range notifiers {
+		select {
+		// Try to write the error to the notification
+		// channel.
+		case notifier.ConfErrChan <- confErr:
+
+		// If a quit signal was provided by the swap,
+		// continue.
+		case <-notifier.QuitChan:
+
+		// If the context was canceled, stop.
+		case <-ctx.Done():
+		}
+	}
+}
+
 func (b *batch) persistSweep(ctx context.Context, sweep sweep,
 	completed bool) error {
 

sweepbatcher/sweep_batcher.go

@@ -20,6 +20,7 @@ import (
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/swap"
 	"github.com/lightninglabs/loop/utils"
+	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/clock"
 	"github.com/lightningnetwork/lnd/input"
 	"github.com/lightningnetwork/lnd/lntypes"
@@ -230,6 +231,14 @@ type SpendNotifier struct {
 	// SpendErrChan is a channel where spend errors are received.
 	SpendErrChan chan<- error
 
+	// ConfChan is a channel where the confirmation details are received.
+	// This channel is optional.
+	ConfChan chan<- *chainntnfs.TxConfirmation
+
+	// ConfErrChan is a channel where confirmation errors are received.
+	// This channel is optional.
+	ConfErrChan chan<- error
+
 	// QuitChan is a channel that can be closed to stop the notifier.
 	QuitChan <-chan bool
 }
@@ -937,7 +946,9 @@ 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.
+// the response back to the response channel. It is called if the batch is fully
+// confirmed and we just need to deliver the data back to the caller though
+// SpendNotifier.
 func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 	parentBatchID int32, notifier *SpendNotifier) error {
 
@@ -995,6 +1006,15 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 			select {
 			// Try to write the update to the notification channel.
 			case notifier.SpendChan <- spendDetail:
+				err := b.monitorConfAndNotify(
+					ctx, sweep, notifier, spendTx,
+				)
+				if err != nil {
+					b.writeToErrChan(
+						ctx, fmt.Errorf("monitor conf "+
+							"failed: %w", err),
+					)
+				}
 
 			// If a quit signal was provided by the swap, continue.
 			case <-notifier.QuitChan:
@@ -1038,6 +1058,78 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
 	return nil
 }
 
+// monitorConfAndNotify monitors the confirmation of a specific transaction and
+// writes the response back to the response channel. It is called if the batch
+// is fully confirmed and we just need to deliver the data back to the caller
+// though SpendNotifier.
+func (b *Batcher) monitorConfAndNotify(ctx context.Context, sweep *sweep,
+	notifier *SpendNotifier, spendTx *wire.MsgTx) error {
+
+	// If confirmation notifications were not requested, stop.
+	if notifier.ConfChan == nil && notifier.ConfErrChan == nil {
+		return nil
+	}
+
+	batchTxid := spendTx.TxHash()
+
+	if len(spendTx.TxOut) != 1 {
+		return fmt.Errorf("unexpected number of outputs in batch: %d, "+
+			"want %d", len(spendTx.TxOut), 1)
+	}
+	batchPkScript := spendTx.TxOut[0].PkScript
+
+	reorgChan := make(chan struct{})
+
+	confCtx, cancel := context.WithCancel(ctx)
+
+	confChan, errChan, err := b.chainNotifier.RegisterConfirmationsNtfn(
+		confCtx, &batchTxid, batchPkScript, batchConfHeight,
+		sweep.initiationHeight, lndclient.WithReOrgChan(reorgChan),
+	)
+	if err != nil {
+		cancel()
+		return err
+	}
+
+	b.wg.Add(1)
+	go func() {
+		defer cancel()
+		defer b.wg.Done()
+
+		select {
+		case conf := <-confChan:
+			if notifier.ConfChan != nil {
+				select {
+				case notifier.ConfChan <- conf:
+				case <-notifier.QuitChan:
+				case <-ctx.Done():
+				}
+			}
+
+		case err := <-errChan:
+			if notifier.ConfErrChan != nil {
+				select {
+				case notifier.ConfErrChan <- err:
+				case <-notifier.QuitChan:
+				case <-ctx.Done():
+				}
+			}
+
+			b.writeToErrChan(ctx, fmt.Errorf("confirmations "+
+				"monitoring error: %w", err))
+
+		case <-reorgChan:
+			// A re-org has been detected, but the batch is fully
+			// confirmed and this is unexpected. Crash the batcher.
+			b.writeToErrChan(ctx, fmt.Errorf("unexpected reorg"))
+
+		case <-ctx.Done():
+		}
+	}()
+
+	return nil
+}
+
 func (b *Batcher) writeToErrChan(ctx context.Context, err error) {
 	select {
 	case b.errChan <- err:

sweepbatcher/sweep_batcher_test.go

@@ -887,9 +887,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// Deliver sweep request to batcher.
 	spendChan := make(chan *SpendDetail, 1)
+	confErrChan := make(chan error)
 	notifier = &SpendNotifier{
 		SpendChan:    spendChan,
 		SpendErrChan: make(chan error, 1),
+		ConfErrChan:  confErrChan,
 		QuitChan:     make(chan bool, 1),
 	}
 	sweepReq1.Notifier = notifier
@@ -968,6 +970,10 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Emulate a confirmation error.
 	confReg.ErrChan <- testError
 
+	// Make sure the notifier gets the confirmation error.
+	confErr := <-confErrChan
+	require.ErrorIs(t, confErr, testError)
+
 	// Wait for the batcher to crash because of the confirmation error.
 	runErr = <-runErrChan
 	require.ErrorIs(t, runErr, testError)
@@ -986,9 +992,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// Deliver sweep request to batcher.
 	spendChan = make(chan *SpendDetail, 1)
+	confChan := make(chan *chainntnfs.TxConfirmation)
 	notifier = &SpendNotifier{
 		SpendChan:    spendChan,
 		SpendErrChan: make(chan error, 1),
+		ConfChan:     confChan,
 		QuitChan:     make(chan bool, 1),
 	}
 	sweepReq1.Notifier = notifier
@@ -1043,9 +1051,15 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 
 	// We mock the tx confirmation notification.
 	lnd.ConfChannel <- &chainntnfs.TxConfirmation{
-		Tx: spendingTx,
+		BlockHeight: 604,
+		Tx:          spendingTx,
 	}
 
+	// Make sure the notifier gets a confirmation notification.
+	conf := <-confChan
+	require.Equal(t, uint32(604), conf.BlockHeight)
+	require.Equal(t, spendingTx.TxHash(), conf.Tx.TxHash())
+
 	// Eventually the batch receives the confirmation notification and
 	// confirms itself.
 	require.Eventually(t, func() bool {
@@ -1060,9 +1074,11 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Now emulate adding the sweep again after it was fully confirmed.
 	// This triggers another code path (monitorSpendAndNotify).
 	spendChan = make(chan *SpendDetail, 1)
+	confChan = make(chan *chainntnfs.TxConfirmation)
 	notifier = &SpendNotifier{
 		SpendChan:    spendChan,
 		SpendErrChan: make(chan error, 1),
+		ConfChan:     confChan,
 		QuitChan:     make(chan bool, 1),
 	}
 	sweepReq1.Notifier = notifier
@@ -1079,6 +1095,18 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	require.Equal(t, spendingTxHash, spending.Tx.TxHash())
 	require.Equal(t, btcutil.Amount(fee), spending.OnChainFeePortion)
 
+	// We mock the tx confirmation notification.
+	<-lnd.RegisterConfChannel
+	lnd.ConfChannel <- &chainntnfs.TxConfirmation{
+		BlockHeight: 604,
+		Tx:          spendingTx,
+	}
+
+	// Make sure the notifier gets a confirmation notification.
+	conf = <-confChan
+	require.Equal(t, uint32(604), conf.BlockHeight)
+	require.Equal(t, spendingTx.TxHash(), conf.Tx.TxHash())
+
 	// Now check what happens in case of a spending error.
 	spendErrChan = make(chan error, 1)
 	notifier = &SpendNotifier{
@@ -1103,6 +1131,43 @@ func testSweepBatcherSimpleLifecycle(t *testing.T, store testStore,
 	// Wait for the batcher to crash because of the spending error.
 	runErr = <-runErrChan
 	require.ErrorIs(t, runErr, testError)
+
+	// Now launch the batcher again.
+	batcher = NewBatcher(lnd.WalletKit, lnd.ChainNotifier, lnd.Signer,
+		testMuSig2SignSweep, testVerifySchnorrSig, lnd.ChainParams,
+		batcherStore, sweepStore)
+	go func() {
+		runErrChan <- batcher.Run(ctx)
+	}()
+
+	// Now check what happens in case of a confirmation error.
+	confErrChan = make(chan error, 1)
+	notifier = &SpendNotifier{
+		SpendChan:    make(chan *SpendDetail, 1),
+		SpendErrChan: make(chan error, 1),
+		ConfErrChan:  confErrChan,
+		QuitChan:     make(chan bool, 1),
+	}
+	sweepReq1.Notifier = notifier
+	require.NoError(t, batcher.AddSweep(&sweepReq1))
+
+	// Expect a spending registration.
+	<-lnd.RegisterSpendChannel
+
+	// We notify the spend.
+	lnd.SpendChannel <- spendDetail
+
+	// We mock the tx confirmation error notification.
+	confReg = <-lnd.RegisterConfChannel
+	confReg.ErrChan <- testError
+
+	// Make sure the notifier gets the confirmation error.
+	confErr = <-confErrChan
+	require.ErrorIs(t, confErr, testError)
+
+	// Wait for the batcher to crash because of the confirmation error.
+	runErr = <-runErrChan
+	require.ErrorIs(t, runErr, testError)
 }
 
 // wrappedLogger implements btclog.Logger, recording last debug message format.