From c70257a4a537d9b43469cb1312ed719e8da5783b Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 8 Apr 2025 23:15:23 -0300 Subject: [PATCH] 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. --- sweepbatcher/sweep_batch.go | 90 +++++++++++++++++++++- sweepbatcher/sweep_batcher.go | 116 ++++++++++++++++++++++++++++- sweepbatcher/sweep_batcher_test.go | 69 ++++++++++++++++- 3 files changed, 272 insertions(+), 3 deletions(-) diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index dc57b0f9..3084a26a 100644 --- a/sweepbatcher/sweep_batch.go +++ b/sweepbatcher/sweep_batch.go @@ -1882,6 +1882,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)) @@ -2158,7 +2160,55 @@ func (b *batch) handleConf(ctx context.Context, 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) + } + + // Calculate the fee portion that each sweep should pay for the batch. + // TODO: make sure spendTx matches b.sweeps. + var totalSweptAmt btcutil.Amount + for _, s := range b.sweeps { + totalSweptAmt += s.value + } + feePortionPaidPerSweep, roundingDifference := getFeePortionForSweep( + spendTx, len(b.sweeps), totalSweptAmt, + ) + + // 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 + } + + confDetail := &ConfDetail{ + TxConfirmation: conf, + OnChainFeePortion: getFeePortionPaidBySweep( + spendTx, feePortionPaidPerSweep, + roundingDifference, &s, + ), + } + + // Notify the caller in a goroutine to avoid possible dead-lock. + go func(notifier *SpendNotifier) { + // Note that we don't unblock on ctx, because it will + // expire soon, when batch.Run completes. The caller is + // responsible to consume ConfChan or close QuitChan. + select { + // Try to write the confirmation to the notification + // channel. + case notifier.ConfChan <- confDetail: + + // If a quit signal was provided by the swap, + // continue. + case <-notifier.QuitChan: + } + }(s.notifier) + } + + return nil } // isComplete returns true if the batch is completed. This method is used by the @@ -2314,6 +2364,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 { diff --git a/sweepbatcher/sweep_batcher.go b/sweepbatcher/sweep_batcher.go index 9e3a7ab6..b89ba5f6 100644 --- a/sweepbatcher/sweep_batcher.go +++ b/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" @@ -280,6 +281,8 @@ type addSweepsRequest struct { parentBatch *dbBatch } +// SpendDetail is a notification that is send to the user of sweepbatcher when +// a batch gets the first confirmation. type SpendDetail struct { // Tx is the transaction that spent the outpoint. Tx *wire.MsgTx @@ -291,6 +294,19 @@ type SpendDetail struct { OnChainFeePortion btcutil.Amount } +// ConfDetail is a notification that is send to the user of sweepbatcher when +// a batch is fully confirmed, i.e. gets batchConfHeight confirmations. +type ConfDetail struct { + // TxConfirmation has data about the confirmation of the transaction. + *chainntnfs.TxConfirmation + + // OnChainFeePortion is the fee portion that was paid to get this sweep + // confirmed on chain. This is the difference between the value of the + // outpoint and the value of all sweeps that were included in the batch + // divided by the number of sweeps. + OnChainFeePortion btcutil.Amount +} + // SpendNotifier is a notifier that is used to notify the requester of a sweep // that the sweep was successful. type SpendNotifier struct { @@ -300,6 +316,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<- *ConfDetail + + // 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 } @@ -1114,7 +1138,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 { @@ -1172,6 +1198,16 @@ 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, + onChainFeePortion, + ) + 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: @@ -1215,6 +1251,84 @@ 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, + onChainFeePortion btcutil.Amount) 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 { + confDetail := &ConfDetail{ + TxConfirmation: conf, + OnChainFeePortion: onChainFeePortion, + } + + select { + case notifier.ConfChan <- confDetail: + 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: diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index 13188ff5..fa871829 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/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 *ConfDetail) notifier = &SpendNotifier{ SpendChan: spendChan, SpendErrChan: make(chan error, 1), + ConfChan: confChan, QuitChan: make(chan bool, 1), } sweepReq1.Notifier = notifier @@ -1043,9 +1051,16 @@ 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()) + require.Equal(t, btcutil.Amount(fee), conf.OnChainFeePortion) + // Eventually the batch receives the confirmation notification and // confirms itself. require.Eventually(t, func() bool { @@ -1060,9 +1075,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 *ConfDetail) notifier = &SpendNotifier{ SpendChan: spendChan, SpendErrChan: make(chan error, 1), + ConfChan: confChan, QuitChan: make(chan bool, 1), } sweepReq1.Notifier = notifier @@ -1079,6 +1096,19 @@ 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()) + require.Equal(t, btcutil.Amount(fee), conf.OnChainFeePortion) + // Now check what happens in case of a spending error. spendErrChan = make(chan error, 1) notifier = &SpendNotifier{ @@ -1103,6 +1133,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(ctx, &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.