From 957ca21e61484d401688f4c965934ac5faca7e95 Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 8 Jul 2025 00:43:41 -0300 Subject: [PATCH] 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. --- sweepbatcher/sweep_batch.go | 98 +++++++++++++------------------------ 1 file changed, 34 insertions(+), 64 deletions(-) diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index 933077fc..b739243b 100644 --- a/sweepbatcher/sweep_batch.go +++ b/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,6 +979,11 @@ 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) @@ -986,16 +991,14 @@ func (b *batch) Run(ctx context.Context) error { 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,33 @@ 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) - - spendChan, spendErr, err := b.chainNotifier.RegisterSpendNtfn( - spendCtx, &primarySweep.outpoint, primarySweep.htlc.PkScript, - primarySweep.initiationHeight, - ) - if err != nil { - cancel() - - return err + if b.spendChan != nil || b.spendErrChan != nil || b.reorgChan != nil { + return fmt.Errorf("an attempt to run monitorSpend multiple " + + "times per batch") } - b.wg.Add(1) - go func() { - defer cancel() - defer b.wg.Done() + reorgChan := make(chan struct{}, 1) - b.Infof("monitoring spend for outpoint %s", - primarySweep.outpoint.String()) + spendChan, spendErrChan, err := b.chainNotifier.RegisterSpendNtfn( + ctx, &primarySweep.outpoint, primarySweep.htlc.PkScript, + primarySweep.initiationHeight, + lndclient.WithReOrgChan(reorgChan), + ) + if err != nil { + 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) + } - select { - case spend := <-spendChan: - select { - case b.spendChan <- spend: + b.Infof("monitoring spend for outpoint %s", + primarySweep.outpoint.String()) - case <-ctx.Done(): - } - - case err := <-spendErr: - b.writeToSpendErrChan(ctx, err) - - b.writeToErrChan( - fmt.Errorf("spend error: %w", err), - ) - - case <-ctx.Done(): - } - }() + // This is safe to do as we always call monitorSpend from the event + // loop's goroutine. + b.spendChan = spendChan + b.spendErrChan = spendErrChan + b.reorgChan = reorgChan return nil } @@ -1862,14 +1854,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 +1884,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 +2372,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 +2383,6 @@ func (b *batch) writeToSpendErrChan(ctx context.Context, spendErr error) { notifiers = append(notifiers, s.notifier) } - done() for _, notifier := range notifiers { select {