From ab59bdafd29593532e1b1fa820d2098922e498a9 Mon Sep 17 00:00:00 2001 From: Boris Nagaev Date: Tue, 25 Feb 2025 23:29:25 -0300 Subject: [PATCH] sweepbatcher: replace batch logger atomically This is needed to fix crashes in unit tests under -race. --- sweepbatcher/sweep_batch.go | 129 +++++++++++++++++++---------- sweepbatcher/sweep_batcher_test.go | 8 +- 2 files changed, 87 insertions(+), 50 deletions(-) diff --git a/sweepbatcher/sweep_batch.go b/sweepbatcher/sweep_batch.go index 6a4989cd..1206b066 100644 --- a/sweepbatcher/sweep_batch.go +++ b/sweepbatcher/sweep_batch.go @@ -9,6 +9,7 @@ import ( "math" "strings" "sync" + "sync/atomic" "time" "github.com/btcsuite/btcd/blockchain" @@ -284,8 +285,8 @@ type batch struct { // cfg is the configuration for this batch. cfg *batchConfig - // log is the logger for this batch. - log btclog.Logger + // log_ is the logger for this batch. + log_ atomic.Pointer[btclog.Logger] wg sync.WaitGroup } @@ -387,7 +388,7 @@ func NewBatchFromDB(cfg batchConfig, bk batchKit) (*batch, error) { } } - return &batch{ + b := &batch{ id: bk.id, state: bk.state, primarySweepID: bk.primaryID, @@ -412,9 +413,42 @@ func NewBatchFromDB(cfg batchConfig, bk batchKit) (*batch, error) { publishErrorHandler: bk.publishErrorHandler, purger: bk.purger, store: bk.store, - log: bk.log, cfg: &cfg, - }, nil + } + + b.setLog(bk.log) + + return b, nil +} + +// log returns current logger. +func (b *batch) log() btclog.Logger { + return *b.log_.Load() +} + +// setLog atomically replaces the logger. +func (b *batch) setLog(logger btclog.Logger) { + b.log_.Store(&logger) +} + +// Debugf logs a message with level DEBUG. +func (b *batch) Debugf(format string, params ...interface{}) { + b.log().Debugf(format, params...) +} + +// Infof logs a message with level INFO. +func (b *batch) Infof(format string, params ...interface{}) { + b.log().Infof(format, params...) +} + +// Warnf logs a message with level WARN. +func (b *batch) Warnf(format string, params ...interface{}) { + b.log().Warnf(format, params...) +} + +// Errorf logs a message with level ERROR. +func (b *batch) Errorf(format string, params ...interface{}) { + b.log().Errorf(format, params...) } // addSweep tries to add a sweep to the batch. If this is the first sweep being @@ -430,7 +464,7 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) { // If the provided sweep is nil, we can't proceed with any checks, so // we just return early. if sweep == nil { - b.log.Infof("the sweep is nil") + b.Infof("the sweep is nil") return false, nil } @@ -473,7 +507,7 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) { // the batch, do not add another sweep to prevent the tx from becoming // non-standard. if len(b.sweeps) >= MaxSweepsPerBatch { - b.log.Infof("the batch has already too many sweeps (%d >= %d)", + b.Infof("the batch has already too many sweeps %d >= %d", len(b.sweeps), MaxSweepsPerBatch) return false, nil @@ -483,7 +517,7 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) { // arrive here after the batch got closed because of a spend. In this // case we cannot add the sweep to this batch. if b.state != Open { - b.log.Infof("the batch state (%v) is not open", b.state) + b.Infof("the batch state (%v) is not open", b.state) return false, nil } @@ -493,15 +527,15 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) { // we cannot add this sweep to the batch. for _, s := range b.sweeps { if s.isExternalAddr { - b.log.Infof("the batch already has a sweep (%x) with "+ + b.Infof("the batch already has a sweep %x with "+ "an external address", s.swapHash[:6]) return false, nil } if sweep.isExternalAddr { - b.log.Infof("the batch is not empty and new sweep (%x)"+ - " has an external address", sweep.swapHash[:6]) + b.Infof("the batch is not empty and new sweep %x "+ + "has an external address", sweep.swapHash[:6]) return false, nil } @@ -515,7 +549,7 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) { int32(math.Abs(float64(sweep.timeout - s.timeout))) if timeoutDistance > b.cfg.maxTimeoutDistance { - b.log.Infof("too long timeout distance between the "+ + b.Infof("too long timeout distance between the "+ "batch and sweep %x: %d > %d", sweep.swapHash[:6], timeoutDistance, b.cfg.maxTimeoutDistance) @@ -544,7 +578,7 @@ func (b *batch) addSweep(ctx context.Context, sweep *sweep) (bool, error) { } // Add the sweep to the batch's sweeps. - b.log.Infof("adding sweep %x", sweep.swapHash[:6]) + b.Infof("adding sweep %x", sweep.swapHash[:6]) b.sweeps[sweep.swapHash] = *sweep // Update FeeRate. Max(sweep.minFeeRate) for all the sweeps of @@ -572,7 +606,7 @@ func (b *batch) sweepExists(hash lntypes.Hash) bool { // Wait waits for the batch to gracefully stop. func (b *batch) Wait() { - b.log.Infof("Stopping") + b.Infof("Stopping") <-b.finished } @@ -613,8 +647,7 @@ func (b *batch) Run(ctx context.Context) error { // Set currentHeight here, because it may be needed in monitorSpend. select { case b.currentHeight = <-blockChan: - b.log.Debugf("initial height for the batch is %v", - b.currentHeight) + b.Debugf("initial height for the batch is %v", b.currentHeight) case <-runCtx.Done(): return runCtx.Err() @@ -652,7 +685,7 @@ func (b *batch) Run(ctx context.Context) error { // completes. timerChan := clock.TickAfter(b.cfg.batchPublishDelay) - b.log.Infof("started, primary %x, total sweeps %v", + b.Infof("started, primary %x, total sweeps %v", b.primarySweepID[0:6], len(b.sweeps)) for { @@ -662,7 +695,7 @@ func (b *batch) Run(ctx context.Context) error { // blockChan provides immediately the current tip. case height := <-blockChan: - b.log.Debugf("received block %v", height) + b.Debugf("received block %v", height) // Set the timer to publish the batch transaction after // the configured delay. @@ -670,7 +703,7 @@ func (b *batch) Run(ctx context.Context) error { b.currentHeight = height case <-initialDelayChan: - b.log.Debugf("initial delay of duration %v has ended", + b.Debugf("initial delay of duration %v has ended", b.cfg.initialDelay) // Set the timer to publish the batch transaction after @@ -680,8 +713,8 @@ func (b *batch) Run(ctx context.Context) error { case <-timerChan: // Check that batch is still open. if b.state != Open { - b.log.Debugf("Skipping publishing, because the"+ - " batch is not open (%v).", b.state) + b.Debugf("Skipping publishing, because "+ + "the batch is not open (%v).", b.state) continue } @@ -695,7 +728,7 @@ func (b *batch) Run(ctx context.Context) error { // initialDelayChan has just fired, this check passes. now := clock.Now() if skipBefore.After(now) { - b.log.Debugf(stillWaitingMsg, skipBefore, now) + b.Debugf(stillWaitingMsg, skipBefore, now) continue } @@ -715,8 +748,8 @@ func (b *batch) Run(ctx context.Context) error { case <-b.reorgChan: b.state = Open - b.log.Warnf("reorg detected, batch is able to accept " + - "new sweeps") + b.Warnf("reorg detected, batch is able to " + + "accept new sweeps") err := b.monitorSpend(ctx, b.sweeps[b.primarySweepID]) if err != nil { @@ -755,8 +788,10 @@ func (b *batch) timeout() int32 { func (b *batch) isUrgent(skipBefore time.Time) bool { timeout := b.timeout() if timeout <= 0 { - b.log.Warnf("Method timeout() returned %v. Number of"+ - " sweeps: %d. It may be an empty batch.", + // This may happen if the batch is empty or if SweepInfo.Timeout + // is not set, may be possible in tests or if there is a bug. + b.Warnf("Method timeout() returned %v. Number of "+ + "sweeps: %d. It may be an empty batch.", timeout, len(b.sweeps)) return false } @@ -779,7 +814,7 @@ func (b *batch) isUrgent(skipBefore time.Time) bool { return false } - b.log.Debugf("cancelling waiting for urgent sweep (timeBank is %v, "+ + b.Debugf("cancelling waiting for urgent sweep (timeBank is %v, "+ "remainingWaiting is %v)", timeBank, remainingWaiting) // Signal to the caller to cancel initialDelay. @@ -795,7 +830,7 @@ func (b *batch) publish(ctx context.Context) error { ) if len(b.sweeps) == 0 { - b.log.Debugf("skipping publish: no sweeps in the batch") + b.Debugf("skipping publish: no sweeps in the batch") return nil } @@ -808,7 +843,7 @@ func (b *batch) publish(ctx context.Context) error { // logPublishError is a function which logs publish errors. logPublishError := func(errMsg string, err error) { - b.publishErrorHandler(err, errMsg, b.log) + b.publishErrorHandler(err, errMsg, b.log()) } fee, err, signSuccess = b.publishMixedBatch(ctx) @@ -830,9 +865,9 @@ func (b *batch) publish(ctx context.Context) error { } } - b.log.Infof("published, total sweeps: %v, fees: %v", len(b.sweeps), fee) + b.Infof("published, total sweeps: %v, fees: %v", len(b.sweeps), fee) for _, sweep := range b.sweeps { - b.log.Infof("published sweep %x, value: %v", + b.Infof("published sweep %x, value: %v", sweep.swapHash[:6], sweep.value) } @@ -1026,7 +1061,7 @@ func (b *batch) publishMixedBatch(ctx context.Context) (btcutil.Amount, error, coopInputs int ) for attempt := 1; ; attempt++ { - b.log.Infof("Attempt %d of collecting cooperative signatures.", + b.Infof("Attempt %d of collecting cooperative signatures.", attempt) // Construct unsigned batch transaction. @@ -1062,7 +1097,7 @@ func (b *batch) publishMixedBatch(ctx context.Context) (btcutil.Amount, error, ctx, i, sweep, tx, prevOutsMap, psbtBytes, ) if err != nil { - b.log.Infof("cooperative signing failed for "+ + b.Infof("cooperative signing failed for "+ "sweep %x: %v", sweep.swapHash[:6], err) // Set coopFailed flag for this sweep in all the @@ -1201,7 +1236,7 @@ func (b *batch) publishMixedBatch(ctx context.Context) (btcutil.Amount, error, } } txHash := tx.TxHash() - b.log.Infof("attempting to publish batch tx=%v with feerate=%v, "+ + b.Infof("attempting to publish batch tx=%v with feerate=%v, "+ "weight=%v, feeForWeight=%v, fee=%v, sweeps=%d, "+ "%d cooperative: (%s) and %d non-cooperative (%s), destAddr=%s", txHash, b.rbfCache.FeeRate, weight, feeForWeight, fee, @@ -1215,7 +1250,7 @@ func (b *batch) publishMixedBatch(ctx context.Context) (btcutil.Amount, error, blockchain.GetTransactionWeight(btcutil.NewTx(tx)), ) if realWeight != weight { - b.log.Warnf("actual weight of tx %v is %v, estimated as %d", + b.Warnf("actual weight of tx %v is %v, estimated as %d", txHash, realWeight, weight) } @@ -1239,11 +1274,11 @@ func (b *batch) debugLogTx(msg string, tx *wire.MsgTx) { // Serialize the transaction and convert to hex string. buf := bytes.NewBuffer(make([]byte, 0, tx.SerializeSize())) if err := tx.Serialize(buf); err != nil { - b.log.Errorf("failed to serialize tx for debug log: %v", err) + b.Errorf("failed to serialize tx for debug log: %v", err) return } - b.log.Debugf("%s: %s", msg, hex.EncodeToString(buf.Bytes())) + b.Debugf("%s: %s", msg, hex.EncodeToString(buf.Bytes())) } // musig2sign signs one sweep using musig2. @@ -1405,15 +1440,16 @@ func (b *batch) updateRbfRate(ctx context.Context) error { if b.rbfCache.FeeRate == 0 { // We set minFeeRate in each sweep, so fee rate is expected to // be initiated here. - b.log.Warnf("rbfCache.FeeRate is 0, which must not happen.") + b.Warnf("rbfCache.FeeRate is 0, which must not happen.") if b.cfg.batchConfTarget == 0 { - b.log.Warnf("updateRbfRate called with zero " + + b.Warnf("updateRbfRate called with zero " + "batchConfTarget") } - b.log.Infof("initializing rbf fee rate for conf target=%v", + b.Infof("initializing rbf fee rate for conf target=%v", b.cfg.batchConfTarget) + rate, err := b.wallet.EstimateFeeRate( ctx, b.cfg.batchConfTarget, ) @@ -1453,6 +1489,7 @@ func (b *batch) monitorSpend(ctx context.Context, primarySweep sweep) error { ) if err != nil { cancel() + return err } @@ -1461,7 +1498,7 @@ func (b *batch) monitorSpend(ctx context.Context, primarySweep sweep) error { defer cancel() defer b.wg.Done() - b.log.Infof("monitoring spend for outpoint %s", + b.Infof("monitoring spend for outpoint %s", primarySweep.outpoint.String()) for { @@ -1584,7 +1621,7 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error { if len(spendTx.TxOut) > 0 { b.batchPkScript = spendTx.TxOut[0].PkScript } else { - b.log.Warnf("transaction %v has no outputs", txHash) + b.Warnf("transaction %v has no outputs", txHash) } // As a previous version of the batch transaction may get confirmed, @@ -1666,13 +1703,13 @@ func (b *batch) handleSpend(ctx context.Context, spendTx *wire.MsgTx) error { err := b.purger(&sweep) if err != nil { - b.log.Errorf("unable to purge sweep %x: %v", + b.Errorf("unable to purge sweep %x: %v", sweep.SwapHash[:6], err) } } }() - b.log.Infof("spent, total sweeps: %v, purged sweeps: %v", + b.Infof("spent, total sweeps: %v, purged sweeps: %v", len(notifyList), len(purgeList)) err := b.monitorConfirmations(ctx) @@ -1690,7 +1727,7 @@ 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 { - b.log.Infof("confirmed in txid %s", b.batchTxid) + b.Infof("confirmed in txid %s", b.batchTxid) b.state = Confirmed return b.store.ConfirmBatch(ctx, b.id) @@ -1769,7 +1806,7 @@ func (b *batch) insertAndAcquireID(ctx context.Context) (int32, error) { } b.id = id - b.log = batchPrefixLogger(fmt.Sprintf("%d", b.id)) + b.setLog(batchPrefixLogger(fmt.Sprintf("%d", b.id))) return id, nil } diff --git a/sweepbatcher/sweep_batcher_test.go b/sweepbatcher/sweep_batcher_test.go index d861a5b2..a09d4744 100644 --- a/sweepbatcher/sweep_batcher_test.go +++ b/sweepbatcher/sweep_batcher_test.go @@ -939,8 +939,8 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { batch1 = batch } require.NotNil(t, batch1) - testLogger := &wrappedLogger{Logger: batch1.log} - batch1.log = testLogger + testLogger := &wrappedLogger{Logger: batch1.log()} + batch1.setLog(testLogger) // Advance the clock to publishDelay. It will trigger the publishDelay // timer, but won't result in publishing, because of initialDelay. @@ -1233,8 +1233,8 @@ func testDelays(t *testing.T, store testStore, batcherStore testBatcherStore) { } } require.NotNil(t, batch2) - testLogger2 := &wrappedLogger{Logger: batch2.log} - batch2.log = testLogger2 + testLogger2 := &wrappedLogger{Logger: batch2.log()} + batch2.setLog(testLogger2) // Add another sweep which is urgent. It will go to the same batch // to make sure minimum timeout is calculated properly.