GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-13 12:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

sweepbatcher: fix reorg detection

Browse source 
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.
This commit is contained in:
Boris Nagaev 2025-07-08 00:43:41 -03:00
parent 22d345a534
commit 957ca21e61
No known key found for this signature in database
1 changed files with 34 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

98
sweepbatcher/sweep_batch.go
View file

@ -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 {