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 race conditions in UseLogger

Browse source
This commit is contained in:
Boris Nagaev 2025-02-26 00:13:22 -03:00
parent a2cee86783
commit a333031bf9
No known key found for this signature in database
4 changed files with 51 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

4
sweepbatcher/greedy_batch_selection.go
View file

@ -92,8 +92,8 @@ func (b *Batcher) greedyAddSweep(ctx context.Context, sweep *sweep) error {
return nil
}
log.Debugf("Batch selection algorithm returned batch id %d for"+
" sweep %x, but acceptance failed.", batchId,
debugf("Batch selection algorithm returned batch id %d "+
"for sweep %x, but acceptance failed.", batchId,
sweep.swapHash[:6])
}

34
sweepbatcher/log.go
View file

@ -2,15 +2,21 @@ package sweepbatcher
import (
"fmt"
"sync/atomic"
"github.com/btcsuite/btclog"
"github.com/lightningnetwork/lnd/build"
)
// log is a logger that is initialized with no output filters. This
// log_ is a logger that is initialized with no output filters. This
// means the package will not perform any logging by default until the
// caller requests it.
var log btclog.Logger
var log_ atomic.Pointer[btclog.Logger]
// log returns active logger.
func log() btclog.Logger {
return *log_.Load()
}
// The default amount of logging is none.
func init() {
@ -20,12 +26,32 @@ func init() {
// batchPrefixLogger returns a logger that prefixes all log messages with
// the ID.
func batchPrefixLogger(batchID string) btclog.Logger {
return build.NewPrefixLog(fmt.Sprintf("[Batch %s]", batchID), log)
return build.NewPrefixLog(fmt.Sprintf("[Batch %s]", batchID), log())
}
// UseLogger uses a specified Logger to output package logging info.
// This should be used in preference to SetLogWriter if the caller is also
// using btclog.
func UseLogger(logger btclog.Logger) {
log = logger
log_.Store(&logger)
}
// debugf logs a message with level DEBUG.
func debugf(format string, params ...interface{}) {
log().Debugf(format, params...)
}
// infof logs a message with level INFO.
func infof(format string, params ...interface{}) {
log().Infof(format, params...)
}
// warnf logs a message with level WARN.
func warnf(format string, params ...interface{}) {
log().Warnf(format, params...)
}
// errorf logs a message with level ERROR.
func errorf(format string, params ...interface{}) {
log().Errorf(format, params...)
}

32
sweepbatcher/sweep_batcher.go
View file

@ -535,13 +535,15 @@ func (b *Batcher) Run(ctx context.Context) error {
case sweepReq := <-b.sweepReqs:
sweep, err := b.fetchSweep(runCtx, sweepReq)
if err != nil {
log.Warnf("fetchSweep failed: %v.", err)
warnf("fetchSweep failed: %v.", err)
return err
}
err = b.handleSweep(runCtx, sweep, sweepReq.Notifier)
if err != nil {
log.Warnf("handleSweep failed: %v.", err)
warnf("handleSweep failed: %v.", err)
return err
}
@ -550,11 +552,13 @@ func (b *Batcher) Run(ctx context.Context) error {
close(testReq.quit)
case err := <-b.errChan:
log.Warnf("Batcher received an error: %v.", err)
warnf("Batcher received an error: %v.", err)
return err
case <-runCtx.Done():
log.Infof("Stopping Batcher: run context cancelled.")
infof("Stopping Batcher: run context cancelled.")
return runCtx.Err()
}
}
@ -612,8 +616,8 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
return err
}
log.Infof("Batcher handling sweep %x, completed=%v", sweep.swapHash[:6],
completed)
infof("Batcher handling sweep %x, completed=%v",
sweep.swapHash[:6], completed)
// If the sweep has already been completed in a confirmed batch then we
// can't attach its notifier to the batch as that is no longer running.
@ -624,8 +628,8 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
// on-chain confirmations to prevent issues caused by reorgs.
parentBatch, err := b.store.GetParentBatch(ctx, sweep.swapHash)
if err != nil {
log.Errorf("unable to get parent batch for sweep %x: "+
"%v", sweep.swapHash[:6], err)
errorf("unable to get parent batch for sweep %x:"+
" %v", sweep.swapHash[:6], err)
return err
}
@ -676,8 +680,8 @@ func (b *Batcher) handleSweep(ctx context.Context, sweep *sweep,
return nil
}
log.Warnf("Greedy batch selection algorithm failed for sweep %x: %v. "+
"Falling back to old approach.", sweep.swapHash[:6], err)
warnf("Greedy batch selection algorithm failed for sweep %x: %v."+
" Falling back to old approach.", sweep.swapHash[:6], err)
// If one of the batches accepts the sweep, we provide it to that batch.
for _, batch := range b.batches {
@ -782,13 +786,13 @@ func (b *Batcher) spinUpBatchFromDB(ctx context.Context, batch *batch) error {
}
if len(dbSweeps) == 0 {
log.Infof("skipping restored batch %d as it has no sweeps",
infof("skipping restored batch %d as it has no sweeps",
batch.id)
// It is safe to drop this empty batch as it has no sweeps.
err := b.store.DropBatch(ctx, batch.id)
if err != nil {
log.Warnf("unable to drop empty batch %d: %v",
warnf("unable to drop empty batch %d: %v",
batch.id, err)
}
@ -930,7 +934,7 @@ func (b *Batcher) monitorSpendAndNotify(ctx context.Context, sweep *sweep,
b.wg.Add(1)
go func() {
defer b.wg.Done()
log.Infof("Batcher monitoring spend for swap %x",
infof("Batcher monitoring spend for swap %x",
sweep.swapHash[:6])
for {
@ -1109,7 +1113,7 @@ func (b *Batcher) loadSweep(ctx context.Context, swapHash lntypes.Hash,
}
} else {
if s.ConfTarget == 0 {
log.Warnf("Fee estimation was requested for zero "+
warnf("Fee estimation was requested for zero "+
"confTarget for sweep %x.", swapHash[:6])
}
minFeeRate, err = b.wallet.EstimateFeeRate(ctx, s.ConfTarget)

2
sweepbatcher/sweep_batcher_test.go
View file

@ -1382,7 +1382,7 @@ func testMaxSweepsPerBatch(t *testing.T, store testStore,
batcherStore testBatcherStore) {
// Disable logging, because this test is very noisy.
oldLogger := log
oldLogger := log()
UseLogger(build.NewSubLogger("SWEEP", nil))
defer UseLogger(oldLogger)