diff --git a/darepod/config.go b/darepod/config.go index 5af6dc5ab..041c02895 100644 --- a/darepod/config.go +++ b/darepod/config.go @@ -73,7 +73,13 @@ const ( // DefaultEsploraPollInterval is the default interval at which the // lwwallet polls the Esplora API for new blocks and transactions. - DefaultEsploraPollInterval = 5 * time.Second + // Mainnet blocks land roughly every 10 minutes, so a 30 s cadence + // stays comfortably within one block's worth of latency while + // keeping the request volume well under the public mempool.space + // rate limits. Tests / regtest environments that mine blocks on + // demand should override this to a sub-second value via the + // `wallet.pollinterval` config knob. + DefaultEsploraPollInterval = 30 * time.Second // DefaultRecoveryWindow is the default address look-ahead window // used during lwwallet recovery. diff --git a/darepod/server.go b/darepod/server.go index 0590dfa5d..a041e21aa 100644 --- a/darepod/server.go +++ b/darepod/server.go @@ -1198,6 +1198,18 @@ func (s *Server) startLwwallet(ctx context.Context, s.lwWallet = fn.Some(w) s.refreshProofKeyBackend() + // Wire up the chain backend reference if it was deferred at + // startup because the wallet was not yet available. The wallet's + // chain backend was already started inside w.Start() above + // (lwwallet.Wallet.Start calls chainBackend.Start as part of its + // startup sequence). Calling Start a second time here would + // subscribe to the shared TipPoller again and spawn a duplicate + // handleTipEvents goroutine, which would double block-epoch + // notifications and confirmation/spend re-checks. + if s.chainBackend == nil { + s.chainBackend = w.ChainBackend() + } + // Refresh the RPC clients once the wallet is available so the // indexer client picks up the wallet-backed identity key and signer // before any deferred wallet-dependent actors start. @@ -1607,23 +1619,25 @@ func (s *Server) initChainBackend(ctx context.Context) error { case WalletTypeLwwallet: // If the lwwallet is already started (auto-unlock - // succeeded), use its chain backend. Otherwise, we - // need a standalone Esplora chain backend that can - // serve the chain source actor before the wallet is - // ready. + // succeeded), use its chain backend. Otherwise defer + // chain backend creation to startLwwallet so that the + // wallet's TipPoller, EsploraClient, and ChainBackend + // are all owned by the wallet — running a standalone + // EsploraClient + TipPoller here in the interactive- + // unlock path would silently double the Esplora call + // rate and pin s.chainBackend to an orphan that the + // wallet never replaces. if s.lwWallet.IsSome() { w := s.lwWallet.UnsafeFromSome() s.chainBackend = w.ChainBackend() alreadyStarted = true } else { - s.chainBackend = lwwallet.NewChainBackend( - lwwallet.NewEsploraClient( - s.cfg.Wallet.EsploraURL, - s.subLogger(lwwallet.Subsystem), - ), - s.cfg.Wallet.PollInterval, - s.subLogger(lwwallet.Subsystem), - ) + // Defer chain backend start to startLwwallet. + // Skip the Start() call below; mirrors the + // btcwallet path. The chain source actor + // registration in Run is also deferred via + // the same chainBackend == nil check. + return nil } case WalletTypeBtcwallet: diff --git a/go.mod b/go.mod index 49ce5c7b3..8fd4689a0 100644 --- a/go.mod +++ b/go.mod @@ -25,6 +25,7 @@ require ( github.com/lightninglabs/lndclient v0.21.0-rc1 github.com/lightninglabs/loop v0.33.0-beta github.com/lightninglabs/neutrino v0.16.2 + github.com/lightninglabs/neutrino/cache v1.1.3 github.com/lightninglabs/taproot-assets v0.7.0 github.com/lightninglabs/taproot-assets/taprpc v1.0.11 github.com/lightningnetwork/lnd v0.21.0-beta.rc1 @@ -39,6 +40,7 @@ require ( github.com/stretchr/testify v1.11.1 golang.org/x/crypto v0.47.0 golang.org/x/exp v0.0.0-20251113190631-e25ba8c21ef6 + golang.org/x/sync v0.19.0 google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 google.golang.org/grpc v1.79.3 google.golang.org/protobuf v1.36.11 @@ -124,7 +126,6 @@ require ( github.com/klauspost/compress v1.17.9 // indirect github.com/lightninglabs/gozmq v0.0.0-20191113021534-d20a764486bf // indirect github.com/lightninglabs/lightning-node-connect/hashmailrpc v1.0.4-0.20250610182311-2f1d46ef18b7 // indirect - github.com/lightninglabs/neutrino/cache v1.1.3 // indirect github.com/lightningnetwork/lightning-onion v1.3.0 // indirect github.com/lightningnetwork/lnd/actor v0.0.6 // indirect github.com/lightningnetwork/lnd/cert v1.2.2 // indirect @@ -202,7 +203,6 @@ require ( golang.org/x/mod v0.32.0 // indirect golang.org/x/net v0.49.0 // indirect golang.org/x/oauth2 v0.34.0 // indirect - golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.40.0 // indirect golang.org/x/term v0.39.0 // indirect golang.org/x/text v0.33.0 // indirect diff --git a/lwwallet/chain_backend.go b/lwwallet/chain_backend.go index 5fa0588f0..29f8370ac 100644 --- a/lwwallet/chain_backend.go +++ b/lwwallet/chain_backend.go @@ -16,8 +16,32 @@ import ( "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btclog/v2" "github.com/lightninglabs/darepo-client/chainsource" + "golang.org/x/sync/singleflight" ) +// singleflight keys for the broad re-check paths driven by tip +// events. Naming is intentionally stable so concurrent calls +// coalesce; the key strings are not load-bearing for behavior, just +// for deduplication. +const ( + sfKeyCheckConfirmations = "checkConfirmations" + sfKeyCheckSpends = "checkSpends" +) + +// recheckHeartbeatInterval is the cadence at which handleTipEvents +// re-runs checkConfirmations / checkSpends in addition to its +// per-tip-event re-checks. Esplora can index a confirmed tx +// 1-3 seconds after the block lands, so the per-tip-event check +// may run before the status flip is visible — without a heartbeat +// the next re-check would not happen until the *following* block +// arrives, which on mainnet means up to ~10 minutes of latency for +// a status flip that would have been visible in seconds. 60s is +// the upper bound on perceived registration latency under the +// default 30s tip-poll cadence and is well below mainnet's ~10 +// minute typical inter-block gap, so it does not meaningfully +// raise Esplora load over the default poll cadence. +const recheckHeartbeatInterval = 60 * time.Second + // confRegistration tracks a pending confirmation registration within the // polling loop. type confRegistration struct { @@ -72,12 +96,21 @@ type blockRegistration struct { } // ChainBackend implements chainsource.ChainBackend using an Esplora HTTP -// client with polling-based chain monitoring. The backend periodically -// polls the Esplora API for new blocks and checks pending confirmation -// and spend registrations against the current chain state. +// client. New-block detection is delegated to a shared TipPoller so the +// chain backend does not race independently against other Esplora +// consumers (notably the EsploraChainService that feeds btcwallet). +// On each TipBlock event the backend dispatches block epochs and re- +// checks pending confirmation and spend registrations. type ChainBackend struct { - esplora *EsploraClient - pollInterval time.Duration + esplora *EsploraClient + + // tipPoller is the shared tip-event source. The backend may + // own this poller (when nobody else needs the tip stream) or + // just subscribe to one started by another component such as + // the wallet. ownsTipPoller distinguishes the two cases for + // Start/Stop lifecycle management. + tipPoller *TipPoller + ownsTipPoller bool // log is the structured logger for this chain backend instance. log btclog.Logger @@ -95,43 +128,88 @@ type ChainBackend struct { blockRegs map[uint64]*blockRegistration nextRegID uint64 + // sf coalesces concurrent broad re-check goroutines so that N + // rapid tip events do not produce N parallel scans of the + // registration maps. The per-registration one-shot path at + // register time is already O(1) by design; sf protects the + // O(N) broadcast-time path from accidental amplification. + sf singleflight.Group + stopCh chan struct{} stopOnce sync.Once wg sync.WaitGroup } -// NewChainBackend creates a new Esplora-backed chain backend. The -// pollInterval controls how frequently the backend checks for new blocks -// and updates to pending registrations. +// NewChainBackend creates a new Esplora-backed chain backend with a +// dedicated TipPoller it owns end-to-end. Use this constructor when +// no other component (e.g. the wallet's btcwallet chain adapter) +// already runs a TipPoller against the same Esplora client. The +// pollInterval controls how frequently the underlying TipPoller +// asks Esplora for the latest tip. func NewChainBackend(esplora *EsploraClient, pollInterval time.Duration, logger btclog.Logger) *ChainBackend { + tp := NewTipPoller(esplora, pollInterval, logger) + return &ChainBackend{ - esplora: esplora, - pollInterval: pollInterval, - log: logger, - confRegs: make(map[uint64]*confRegistration), - spendRegs: make(map[uint64]*spendRegistration), - blockRegs: make(map[uint64]*blockRegistration), - stopCh: make(chan struct{}), + esplora: esplora, + tipPoller: tp, + ownsTipPoller: true, + log: logger, + confRegs: make(map[uint64]*confRegistration), + spendRegs: make(map[uint64]*spendRegistration), + blockRegs: make(map[uint64]*blockRegistration), + stopCh: make(chan struct{}), + } +} + +// NewChainBackendWithPoller creates a chain backend that subscribes to +// an externally-managed TipPoller. The caller retains responsibility +// for starting and stopping the poller; the backend will not touch its +// lifecycle. Use this constructor when the same Esplora instance is +// shared across components (e.g. the lwwallet integrated wallet, where +// the wallet owns the TipPoller and both the chain backend and +// EsploraChainService subscribe to it). +// +// The tipPoller argument must be non-nil. Passing a nil poller would +// surface as a nil pointer dereference inside Start when the backend +// tries to subscribe; surfacing the misuse at construction time +// instead lets callers see the violation directly. +func NewChainBackendWithPoller(esplora *EsploraClient, + tipPoller *TipPoller, + logger btclog.Logger) (*ChainBackend, error) { + + if tipPoller == nil { + return nil, fmt.Errorf("tip poller must be non-nil") } + + return &ChainBackend{ + esplora: esplora, + tipPoller: tipPoller, + ownsTipPoller: false, + log: logger, + confRegs: make(map[uint64]*confRegistration), + spendRegs: make(map[uint64]*spendRegistration), + blockRegs: make(map[uint64]*blockRegistration), + stopCh: make(chan struct{}), + }, nil } -// Start initializes the chain backend by fetching the current chain tip -// and starting the polling loop. +// Start subscribes the chain backend to the configured TipPoller and +// begins dispatching tip events. When the backend owns its TipPoller, +// Start also starts the poller; otherwise the caller must have +// started it before calling Start. func (b *ChainBackend) Start() error { - // Fetch the initial chain tip. We get height first, then - // resolve the hash for that specific height to avoid drift - // if a new block arrives between the two HTTP calls. - height, err := b.esplora.GetTipHeight() - if err != nil { - return fmt.Errorf("get initial tip: %w", err) + if b.ownsTipPoller { + if err := b.tipPoller.Start(); err != nil { + return fmt.Errorf("start tip poller: %w", err) + } } - hash, err := b.esplora.GetBlockHashByHeight(height) + height, hash, _, sub, err := b.tipPoller.BestBlockAndSubscribe() if err != nil { - return fmt.Errorf("get initial hash: %w", err) + return fmt.Errorf("subscribe to tip poller: %w", err) } b.mu.Lock() @@ -139,9 +217,8 @@ func (b *ChainBackend) Start() error { b.bestHash = hash b.mu.Unlock() - // Start the polling loop. b.wg.Add(1) - go b.pollLoop() + go b.handleTipEvents(sub) b.log.InfoS(context.Background(), "Chain backend started", slog.Int("tip_height", int(height)), @@ -150,15 +227,21 @@ func (b *ChainBackend) Start() error { return nil } -// Stop shuts down the polling loop and cleans up resources. Stop is -// idempotent and safe to call multiple times; the stop channel is -// closed exactly once. +// Stop unsubscribes from the TipPoller, drains the event handler +// goroutine, and (if the backend owns the poller) stops the poller +// itself. Stop is idempotent: the second call returns immediately +// after the first has finished. func (b *ChainBackend) Stop() error { b.log.InfoS(context.Background(), "Stopping chain backend") b.stopOnce.Do(func() { close(b.stopCh) }) + + if b.ownsTipPoller { + b.tipPoller.Stop() + } + b.wg.Wait() b.log.InfoS(context.Background(), "Chain backend stopped") @@ -348,12 +431,72 @@ func (b *ChainBackend) RegisterConf(ctx context.Context, b.mu.Unlock() } + // Run an immediate single-shot check scoped to JUST this + // registration so a tx that is already buried beyond numConfs + // at registration time fires synchronously rather than waiting + // for the next tip event. The previous design re-iterated all + // pending registrations here; with N concurrent registrations + // (the boarding-on-restart flow) that produced an O(N²) HTTP + // burst against Esplora — the very rate-limit problem this + // PR set out to fix. The per-reg one-shot is O(1). + // + // The goroutine is tracked in b.wg so Stop() waits for any + // in-flight registration check to complete before returning; + // without that, a slow Esplora response could outlive the + // chain backend and write into a torn-down subscriber. + b.wg.Add(1) + go b.runConfOneShot(id, reg) + return &chainsource.ConfRegistration{ Confirmed: confChan, Cancel: cancelFn, }, nil } +// runConfOneShot performs the per-registration confirmation check +// triggered at RegisterConf time. It snapshots the current best +// height under the chain backend's lock, asks Esplora for this one +// registration's status, and delivers the result if confirmed. The +// goroutine exits on any of: cancellation, stopCh closure, a +// successful delivery, or a non-confirmed status. +func (b *ChainBackend) runConfOneShot(id uint64, reg *confRegistration) { + defer b.wg.Done() + + select { + case <-reg.cancelCh: + return + case <-b.stopCh: + return + default: + } + + b.mu.Lock() + currentHeight := b.bestHeight + b.mu.Unlock() + + conf := b.checkSingleConf(reg, currentHeight) + if conf == nil { + return + } + + select { + case reg.confChan <- conf: + case <-reg.cancelCh: + return + case <-b.stopCh: + return + } + + b.log.DebugS(context.Background(), + "Confirmation registration fulfilled (one-shot)", + slog.Uint64("reg_id", id), + slog.Int("block_height", int(conf.BlockHeight))) + + b.mu.Lock() + delete(b.confRegs, id) + b.mu.Unlock() +} + // RegisterSpend registers for spend notifications of a transaction output. func (b *ChainBackend) RegisterSpend(ctx context.Context, outpoint *wire.OutPoint, pkScript []byte, @@ -394,12 +537,65 @@ func (b *ChainBackend) RegisterSpend(ctx context.Context, b.mu.Unlock() } + // Per-registration one-shot to handle outpoints that are + // already spent at registration time. See RegisterConf for the + // O(N²)-vs-O(1) rationale and the b.wg lifecycle note. + b.wg.Add(1) + go b.runSpendOneShot(id, reg) + return &chainsource.SpendRegistration{ Spend: spendChan, Cancel: cancelFn, }, nil } +// runSpendOneShot performs the per-registration spend check +// triggered at RegisterSpend time. It exits on cancellation, stopCh +// closure, a successful delivery, or any non-spent / unconfirmed +// status; the broad checkSpends called from processTipEvent re-runs +// it on every tip advance. +func (b *ChainBackend) runSpendOneShot(id uint64, + reg *spendRegistration) { + + defer b.wg.Done() + + select { + case <-reg.cancelCh: + return + case <-b.stopCh: + return + default: + } + + if reg.outpoint == nil { + return + } + + detail := b.checkSingleSpend(reg) + if detail == nil { + return + } + + select { + case reg.spendChan <- detail: + case <-reg.cancelCh: + return + case <-b.stopCh: + return + } + + b.log.DebugS(context.Background(), + "Spend registration fulfilled (one-shot)", + slog.Uint64("reg_id", id), + slog.String("outpoint", reg.outpoint.String()), + slog.String("spender_txid", + detail.SpenderTxHash.String())) + + b.mu.Lock() + delete(b.spendRegs, id) + b.mu.Unlock() +} + // RegisterBlocks registers for new block notifications. func (b *ChainBackend) RegisterBlocks( _ context.Context) (*chainsource.BlockRegistration, error) { @@ -432,18 +628,46 @@ func (b *ChainBackend) RegisterBlocks( }, nil } -// pollLoop is the main polling goroutine. It periodically checks for new -// blocks and processes pending registrations. -func (b *ChainBackend) pollLoop() { +// handleTipEvents drains TipBlock events from the shared poller and +// translates them into chain backend work: emit a BlockEpoch to each +// block-registration subscriber, advance the cached tip, and re-check +// pending confirmation/spend registrations. +// +// On stopCh the loop exits and cancels its subscription so the +// poller does not waste effort fanning to a dead consumer. The +// subscription's Quit channel covers the inverse direction: if the +// poller is shut down externally, we exit promptly without waiting +// for stopCh. +func (b *ChainBackend) handleTipEvents(sub *TipSubscription) { defer b.wg.Done() + defer sub.Cancel() - ticker := time.NewTicker(b.pollInterval) - defer ticker.Stop() + heartbeat := time.NewTicker(recheckHeartbeatInterval) + defer heartbeat.Stop() for { select { - case <-ticker.C: - b.poll() + case event, ok := <-sub.Updates(): + if !ok { + return + } + + b.processTipEvent(event) + + case <-heartbeat.C: + // Re-run the broad checks even when the tip + // hasn't moved. Esplora's status indexer lags + // the block-found event by 1-3 seconds, so a + // processTipEvent that ran before the indexer + // caught up would otherwise not retry until + // the next block lands. Coalesced via the same + // singleflight keys used by processTipEvent so + // a tip event arriving on the same tick does + // not produce two parallel scans. + b.runRecheckHeartbeat() + + case <-sub.Quit(): + return case <-b.stopCh: return @@ -451,85 +675,91 @@ func (b *ChainBackend) pollLoop() { } } -// poll performs a single polling iteration: checks for new blocks and -// processes pending confirmation and spend registrations. -func (b *ChainBackend) poll() { - // Check for new blocks. - newHeight, err := b.esplora.GetTipHeight() - if err != nil { - b.log.WarnS(context.Background(), "Poll tip height failed", err) - return - } - - b.mu.Lock() - oldHeight := b.bestHeight - b.mu.Unlock() +// runRecheckHeartbeat is the heartbeat-tick path that re-runs +// checkConfirmations and checkSpends without a new tip event. It +// goes through the same singleflight keys as processTipEvent so a +// concurrent tip-driven scan and a heartbeat-driven scan share one +// in-flight call rather than running in parallel. +func (b *ChainBackend) runRecheckHeartbeat() { + _, _, _ = b.sf.Do(sfKeyCheckConfirmations, + func() (interface{}, error) { + b.checkConfirmations() + return nil, nil + }) + + _, _, _ = b.sf.Do(sfKeyCheckSpends, + func() (interface{}, error) { + b.checkSpends() + return nil, nil + }) +} - if newHeight <= oldHeight { - // No new blocks, but still check registrations in case - // of reorgs or newly broadcast transactions. - b.checkConfirmations() - b.checkSpends() +// processTipEvent applies one TipBlock to the registration maps: +// fans out a BlockEpoch to every block subscriber, advances the +// cached tip under the same lock, and re-checks confirmation + +// spend registrations. +func (b *ChainBackend) processTipEvent(event *TipBlock) { + if event == nil || event.Header == nil { return } - // Process each new block height. - for height := oldHeight + 1; height <= newHeight; height++ { - hash, err := b.esplora.GetBlockHashByHeight(height) - if err != nil { - b.log.WarnS( - context.Background(), - "Poll block hash failed", err, - ) - - return - } - - // Fetch block metadata for timestamp. - blockInfo, err := b.esplora.GetBlockHeader(hash) - if err != nil { - b.log.WarnS( - context.Background(), - "Poll block header failed", err, - ) - - return - } - - b.log.DebugS(context.Background(), "New block processed", - slog.Int("height", int(height)), - slog.String("hash", hash.String())) + b.log.DebugS(context.Background(), "New block processed", + slog.Int("height", int(event.Height)), + slog.String("hash", event.Hash.String())) - // Notify block subscribers and update the best known - // tip atomically under the same lock. - epoch := &chainsource.BlockEpoch{ - Hash: hash, - Height: height, - Timestamp: blockInfo.Timestamp, - } + epoch := &chainsource.BlockEpoch{ + Hash: event.Hash, + Height: event.Height, + Timestamp: event.Header.Timestamp, + } - b.mu.Lock() - b.bestHeight = height - b.bestHash = hash + // Snapshot the registration set under the lock and update the + // cached tip, then drop the lock before fanning out. Holding + // b.mu across N channel sends serializes every other lock + // acquirer (registrations, confirmation checks) behind the + // fan-out, and a future maintainer who removes the `default:` + // case below would silently introduce a deadlock — separating + // the snapshot from the send makes that mistake impossible. + b.mu.Lock() + b.bestHeight = event.Height + b.bestHash = event.Hash + regs := make([]*blockRegistration, 0, len(b.blockRegs)) + for _, reg := range b.blockRegs { + regs = append(regs, reg) + } + b.mu.Unlock() - for _, reg := range b.blockRegs { - select { - case reg.epochChan <- epoch: + for _, reg := range regs { + select { + case reg.epochChan <- epoch: - case <-reg.cancelCh: + case <-reg.cancelCh: - default: - // Channel full, skip this block for this - // subscriber. The subscriber will catch up - // on the next poll. - } + default: + // Channel full, skip this block for this + // subscriber. The subscriber will catch up on + // the next event. } - b.mu.Unlock() } - // Check registrations after processing new blocks. - b.checkConfirmations() - b.checkSpends() + // Coalesce concurrent broad re-checks via singleflight: if a + // previous tip event's checkConfirmations / checkSpends is + // still running when a new tip event arrives we share its + // result rather than starting a parallel scan. Each scan is + // already O(N) HTTP calls in the registration count; without + // this guard a fast burst of blocks would multiply the load + // against an already-rate-limited Esplora. + _, _, _ = b.sf.Do(sfKeyCheckConfirmations, + func() (interface{}, error) { + b.checkConfirmations() + return nil, nil + }) + + _, _, _ = b.sf.Do(sfKeyCheckSpends, + func() (interface{}, error) { + b.checkSpends() + return nil, nil + }) } // checkConfirmations iterates over all pending confirmation registrations @@ -724,41 +954,11 @@ func (b *ChainBackend) checkSpends() { continue } - outspend, err := b.esplora.GetOutspend( - reg.outpoint.Hash, reg.outpoint.Index, - ) - if err != nil { - continue - } - - if !outspend.Spent { - continue - } - - if !outspend.Status.Confirmed { - continue - } - - // Parse the spending transaction ID. - spenderHash, err := chainhash.NewHashFromStr(outspend.Txid) - if err != nil { - continue - } - - // Fetch the full spending transaction. - spendingTx, err := b.esplora.GetRawTx(*spenderHash) - if err != nil { + detail := b.checkSingleSpend(reg) + if detail == nil { continue } - detail := &chainsource.SpendDetail{ - SpentOutPoint: reg.outpoint, - SpenderTxHash: spenderHash, - SpendingTx: spendingTx, - SpenderInputIndex: outspend.Vin, - SpendingHeight: int32(outspend.Status.BlockHeight), - } - // Send the spend detail. select { case reg.spendChan <- detail: @@ -773,7 +973,7 @@ func (b *ChainBackend) checkSpends() { slog.String("outpoint", reg.outpoint.String()), slog.String("spender_txid", - spenderHash.String())) + detail.SpenderTxHash.String())) // Remove the fulfilled registration. b.mu.Lock() @@ -782,6 +982,49 @@ func (b *ChainBackend) checkSpends() { } } +// checkSingleSpend resolves the spend status of a single spend +// registration via Esplora. Returns nil when the outpoint is not yet +// confirmed-spent, when any HTTP / parse error occurs, or when the +// registration has no outpoint. The caller is responsible for +// delivery, logging, and removing the fulfilled registration; this +// helper only resolves the on-chain question. +func (b *ChainBackend) checkSingleSpend( + reg *spendRegistration) *chainsource.SpendDetail { + + if reg.outpoint == nil { + return nil + } + + outspend, err := b.esplora.GetOutspend( + reg.outpoint.Hash, reg.outpoint.Index, + ) + if err != nil { + return nil + } + + if !outspend.Spent || !outspend.Status.Confirmed { + return nil + } + + spenderHash, err := chainhash.NewHashFromStr(outspend.Txid) + if err != nil { + return nil + } + + spendingTx, err := b.esplora.GetRawTx(*spenderHash) + if err != nil { + return nil + } + + return &chainsource.SpendDetail{ + SpentOutPoint: reg.outpoint, + SpenderTxHash: spenderHash, + SpendingTx: spendingTx, + SpenderInputIndex: outspend.Vin, + SpendingHeight: int32(outspend.Status.BlockHeight), + } +} + // Compile-time check that ChainBackend implements // chainsource.ChainBackend. var _ chainsource.ChainBackend = (*ChainBackend)(nil) diff --git a/lwwallet/chain_backend_test.go b/lwwallet/chain_backend_test.go index 1d0717c34..8f9800a4f 100644 --- a/lwwallet/chain_backend_test.go +++ b/lwwallet/chain_backend_test.go @@ -1,6 +1,7 @@ package lwwallet import ( + "bytes" "encoding/json" "fmt" "net/http" @@ -309,7 +310,11 @@ func TestChainBackendSubmitPackage(t *testing.T) { func TestChainBackendConfRegistration(t *testing.T) { t.Parallel() - txid := chainhash.HashH([]byte("test-tx")) + // The cache fillers in esplora.go verify that the response + // body actually hashes to the requested key before insertion, + // so the txid we register must equal the TxHash of the bytes + // the mock will hand back. + txid := minimalRawTxID(t) srv := mockEsploraServer( t, func(w http.ResponseWriter, r *http.Request) { @@ -384,7 +389,11 @@ func TestChainBackendSpendRegistration(t *testing.T) { t.Parallel() txid := chainhash.HashH([]byte("funding-tx")) - spenderTxid := chainhash.HashH([]byte("spending-tx")) + + // The spender tx body is fetched and now content-verified + // against its txid, so we derive spenderTxid from the actual + // minimalRawTx bytes the mock will return. + spenderTxid := minimalRawTxID(t) srv := mockEsploraServer( t, func(w http.ResponseWriter, r *http.Request) { @@ -488,3 +497,16 @@ func minimalRawTx() []byte { 0x00, 0x00, 0x00, 0x00, } } + +// minimalRawTxID returns the txid of minimalRawTx so tests can wire +// the txid they register through to the bytes the mock Esplora +// server hands back, satisfying the cache fillers' content-hash +// verification. +func minimalRawTxID(t *testing.T) chainhash.Hash { + t.Helper() + + var tx wire.MsgTx + require.NoError(t, tx.Deserialize(bytes.NewReader(minimalRawTx()))) + + return tx.TxHash() +} diff --git a/lwwallet/esplora.go b/lwwallet/esplora.go index c0b6a49ab..41bd99ee3 100644 --- a/lwwallet/esplora.go +++ b/lwwallet/esplora.go @@ -17,12 +17,22 @@ import ( "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/neutrino/cache/lru" + "golang.org/x/sync/singleflight" ) // EsploraClient is an HTTP REST client for the Esplora/mempool.space API. // It provides methods for querying chain state, fetching transactions and // UTXOs, estimating fees, and broadcasting transactions. The client is // safe for concurrent use. +// +// Hash-addressed responses (transactions, full blocks, block headers) +// are cached in process-local LRUs so repeated lookups for the same +// content do not re-issue HTTP requests. The caches are bounded by +// cumulative serialized byte size; see esplora_cache.go for the +// per-cache capacity constants. Live, mutable responses (tip height, +// address UTXOs, tx confirmation status, mempool fee estimates) are +// never cached. type EsploraClient struct { // baseURL is the Esplora API root (e.g. "https://mempool.space/api"). baseURL string @@ -32,6 +42,38 @@ type EsploraClient struct { // log is the structured logger for this Esplora client instance. log btclog.Logger + + // txCache memoizes /tx/:txid/raw responses keyed by txid. A + // confirmed transaction is content-addressed by its txid so the + // cached body can never go stale. + txCache *lru.Cache[chainhash.Hash, cachedTx] + + // rawBlockCache memoizes /block/:hash/raw responses keyed by + // block hash. + rawBlockCache *lru.Cache[chainhash.Hash, cachedBlock] + + // rawHeaderCache memoizes /block/:hash/header responses keyed by + // block hash. + rawHeaderCache *lru.Cache[chainhash.Hash, cachedRawHeader] + + // blockHeaderCache memoizes /block/:hash JSON header responses + // keyed by block hash. + blockHeaderCache *lru.Cache[chainhash.Hash, cachedBlockHeader] + + // txSF, rawBlockSF, rawHeaderSF, and blockHeaderSF coalesce + // concurrent cache misses for the same hash so that the four + // content-addressed cache fillers above do not multiply HTTP + // load against the rate-limited Esplora endpoint when several + // consumers (typically chainBackend and chainSvc reacting to the + // same TipPoller event) race to fetch the same block or + // transaction. Each Group is keyed by the hash string and + // re-checks the cache inside the work function so a sibling + // fetch that populated the cache while we were waiting wins + // without re-issuing the request. + txSF singleflight.Group + rawBlockSF singleflight.Group + rawHeaderSF singleflight.Group + blockHeaderSF singleflight.Group } // NewEsploraClient creates a new Esplora REST API client. The baseURL should @@ -46,6 +88,20 @@ func NewEsploraClient(baseURL string, Timeout: 30 * time.Second, }, log: logger, + txCache: lru.NewCache[chainhash.Hash, cachedTx]( + txCacheCapacity, + ), + rawBlockCache: lru.NewCache[chainhash.Hash, cachedBlock]( + rawBlockCacheCapacity, + ), + rawHeaderCache: lru.NewCache[chainhash.Hash, cachedRawHeader]( + rawHeaderCacheCapacity, + ), + blockHeaderCache: lru.NewCache[ + chainhash.Hash, cachedBlockHeader, + ]( + blockHeaderCacheCapacity, + ), } } @@ -168,21 +224,90 @@ func (c *EsploraClient) GetTipHash() (chainhash.Hash, error) { } // GetBlockHeader returns block metadata (height, timestamp) for the given -// block hash. +// block hash. Results are memoized in blockHeaderCache because the +// header for a confirmed block hash is immutable. Concurrent misses +// for the same hash are coalesced via blockHeaderSF. func (c *EsploraClient) GetBlockHeader( blockHash chainhash.Hash) (*esploraBlock, error) { - body, err := c.get("/block/" + blockHash.String()) + if cached, err := c.blockHeaderCache.Get(blockHash); err == nil && + cached.header != nil { + + return cached.header, nil + } + + v, err, _ := c.blockHeaderSF.Do(blockHash.String(), + func() (interface{}, error) { + // A sibling caller may have populated the cache + // while we were waiting on the singleflight slot; + // re-check before issuing an HTTP request. + if cached, cErr := c.blockHeaderCache.Get( + blockHash, + ); cErr == nil && cached.header != nil { + return cached.header, nil + } + + body, err := c.get("/block/" + blockHash.String()) + if err != nil { + return nil, fmt.Errorf( + "get block header: %w", err, + ) + } + + var block esploraBlock + if err := json.Unmarshal(body, &block); err != nil { + return nil, fmt.Errorf( + "parse block header: %w", err, + ) + } + + // Verify the response actually describes the + // block we asked for before populating the + // cache. Without this check a buggy, MITM'd, or + // compromised Esplora endpoint could pin an + // arbitrary entry under blockHash for the rest + // of this process's lifetime — none of the four + // content-addressed caches have a TTL. + gotID, err := chainhash.NewHashFromStr(block.ID) + if err != nil { + return nil, fmt.Errorf( + "parse block id %q: %w", + block.ID, err, + ) + } + if *gotID != blockHash { + return nil, fmt.Errorf( + "block id mismatch: got %s, want %s", + gotID, blockHash, + ) + } + + if _, putErr := c.blockHeaderCache.Put( + blockHash, + cachedBlockHeader{header: &block}, + ); putErr != nil { + c.log.WarnS(context.Background(), + "Block header cache Put failed", + putErr, + slog.String( + "hash", blockHash.String(), + )) + } + + return &block, nil + }) if err != nil { - return nil, fmt.Errorf("get block header: %w", err) + return nil, err } - var block esploraBlock - if err := json.Unmarshal(body, &block); err != nil { - return nil, fmt.Errorf("parse block header: %w", err) + block, ok := v.(*esploraBlock) + if !ok { + return nil, fmt.Errorf( + "block header singleflight returned %T", v, + ) } - return &block, nil + return block, nil } // GetBlockHashByHeight returns the block hash at the given height. @@ -211,54 +336,172 @@ func (c *EsploraClient) GetBlockHashByHeight( // GetRawBlockHeader returns the deserialized 80-byte block header for // the given block hash. The Esplora /block/:hash/header endpoint returns // the header as a hex-encoded string which we decode and deserialize -// into a wire.BlockHeader. +// into a wire.BlockHeader. Results are memoized in rawHeaderCache +// because the header for a confirmed block hash is immutable. +// Concurrent misses for the same hash are coalesced via rawHeaderSF. func (c *EsploraClient) GetRawBlockHeader( blockHash chainhash.Hash) (*wire.BlockHeader, error) { - body, err := c.get( - "/block/" + blockHash.String() + "/header", - ) + if cached, err := c.rawHeaderCache.Get(blockHash); err == nil && + cached.header != nil { + + return cached.header, nil + } + + v, err, _ := c.rawHeaderSF.Do(blockHash.String(), + func() (interface{}, error) { + if cached, cErr := c.rawHeaderCache.Get( + blockHash, + ); cErr == nil && cached.header != nil { + return cached.header, nil + } + + body, err := c.get( + "/block/" + blockHash.String() + "/header", + ) + if err != nil { + return nil, fmt.Errorf( + "get raw block header: %w", err, + ) + } + + // The response is a hex-encoded 80-byte block + // header. + headerBytes, err := hex.DecodeString( + strings.TrimSpace(string(body)), + ) + if err != nil { + return nil, fmt.Errorf( + "decode block header hex: %w", err, + ) + } + + var header wire.BlockHeader + err = header.Deserialize( + bytes.NewReader(headerBytes), + ) + if err != nil { + return nil, fmt.Errorf( + "deserialize block header: %w", err, + ) + } + + // Hash-verify before caching. See GetBlockHeader + // for the rationale. + if got := header.BlockHash(); got != blockHash { + return nil, fmt.Errorf( + "raw header hash mismatch: got "+ + "%s, want %s", + got, blockHash, + ) + } + + if _, putErr := c.rawHeaderCache.Put( + blockHash, + cachedRawHeader{header: &header}, + ); putErr != nil { + c.log.WarnS(context.Background(), + "Raw header cache Put failed", + putErr, + slog.String( + "hash", blockHash.String(), + )) + } + + return &header, nil + }) if err != nil { - return nil, fmt.Errorf("get raw block header: %w", err) - } - - // The response is a hex-encoded 80-byte block header. - headerBytes, err := hex.DecodeString( - strings.TrimSpace(string(body)), - ) - if err != nil { - return nil, fmt.Errorf( - "decode block header hex: %w", err, - ) + return nil, err } - var header wire.BlockHeader - err = header.Deserialize(bytes.NewReader(headerBytes)) - if err != nil { + header, ok := v.(*wire.BlockHeader) + if !ok { return nil, fmt.Errorf( - "deserialize block header: %w", err, + "raw header singleflight returned %T", v, ) } - return &header, nil + return header, nil } -// GetRawBlock returns the raw serialized block bytes for the given hash. +// GetRawBlock returns the raw serialized block bytes for the given +// hash. Results are memoized in rawBlockCache because a confirmed +// block's contents are content-addressed by their hash. Concurrent +// misses for the same hash are coalesced via rawBlockSF — full +// mainnet blocks approach 4 MiB so collapsing a thundering herd is +// load-bearing for the rate-limit budget of the Esplora endpoint. func (c *EsploraClient) GetRawBlock( blockHash chainhash.Hash) (*wire.MsgBlock, error) { - body, err := c.get("/block/" + blockHash.String() + "/raw") + if cached, err := c.rawBlockCache.Get(blockHash); err == nil && + cached.block != nil { + + return cached.block, nil + } + + v, err, _ := c.rawBlockSF.Do(blockHash.String(), + func() (interface{}, error) { + if cached, cErr := c.rawBlockCache.Get( + blockHash, + ); cErr == nil && cached.block != nil { + return cached.block, nil + } + + body, err := c.get( + "/block/" + blockHash.String() + "/raw", + ) + if err != nil { + return nil, fmt.Errorf( + "get raw block: %w", err, + ) + } + + var block wire.MsgBlock + err = block.Deserialize(bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf( + "deserialize block: %w", err, + ) + } + + // Hash-verify before caching. See GetBlockHeader + // for the rationale. + if got := block.BlockHash(); got != blockHash { + return nil, fmt.Errorf( + "raw block hash mismatch: got "+ + "%s, want %s", + got, blockHash, + ) + } + + if _, putErr := c.rawBlockCache.Put( + blockHash, cachedBlock{ + block: &block, + size: uint64(len(body)), + }, + ); putErr != nil { + c.log.WarnS(context.Background(), + "Raw block cache Put failed", + putErr, + slog.String( + "hash", blockHash.String(), + )) + } + + return &block, nil + }) if err != nil { - return nil, fmt.Errorf("get raw block: %w", err) + return nil, err } - var block wire.MsgBlock - err = block.Deserialize(bytes.NewReader(body)) - if err != nil { - return nil, fmt.Errorf("deserialize block: %w", err) + block, ok := v.(*wire.MsgBlock) + if !ok { + return nil, fmt.Errorf( + "raw block singleflight returned %T", v, + ) } - return &block, nil + return block, nil } // GetScriptUtxos returns all UTXOs for the given pkScript. @@ -337,21 +580,75 @@ func (c *EsploraClient) GetTxStatus( } // GetRawTx returns the raw serialized transaction bytes for a txid. +// Results are memoized in txCache because a confirmed transaction's +// contents are content-addressed by its txid. Concurrent misses for +// the same txid are coalesced via txSF. func (c *EsploraClient) GetRawTx( txid chainhash.Hash) (*wire.MsgTx, error) { - body, err := c.get("/tx/" + txid.String() + "/raw") + if cached, err := c.txCache.Get(txid); err == nil && + cached.tx != nil { + + return cached.tx, nil + } + + v, err, _ := c.txSF.Do(txid.String(), + func() (interface{}, error) { + if cached, cErr := c.txCache.Get( + txid, + ); cErr == nil && cached.tx != nil { + return cached.tx, nil + } + + body, err := c.get("/tx/" + txid.String() + "/raw") + if err != nil { + return nil, fmt.Errorf( + "get raw tx: %w", err, + ) + } + + var tx wire.MsgTx + err = tx.Deserialize(bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf( + "deserialize tx: %w", err, + ) + } + + // Hash-verify before caching. See GetBlockHeader + // for the rationale. TxHash returns the + // witness-stripped txid, which is what + // /tx/:txid/raw is keyed by. + if got := tx.TxHash(); got != txid { + return nil, fmt.Errorf( + "tx hash mismatch: got %s, want %s", + got, txid, + ) + } + + if _, putErr := c.txCache.Put(txid, cachedTx{ + tx: &tx, + size: uint64(len(body)), + }); putErr != nil { + c.log.WarnS(context.Background(), + "Tx cache Put failed", putErr, + slog.String("txid", txid.String())) + } + + return &tx, nil + }) if err != nil { - return nil, fmt.Errorf("get raw tx: %w", err) + return nil, err } - var tx wire.MsgTx - err = tx.Deserialize(bytes.NewReader(body)) - if err != nil { - return nil, fmt.Errorf("deserialize tx: %w", err) + tx, ok := v.(*wire.MsgTx) + if !ok { + return nil, fmt.Errorf( + "raw tx singleflight returned %T", v, + ) } - return &tx, nil + return tx, nil } // BroadcastTx broadcasts a raw transaction to the network. Returns the diff --git a/lwwallet/esplora_cache.go b/lwwallet/esplora_cache.go new file mode 100644 index 000000000..c83bb35f3 --- /dev/null +++ b/lwwallet/esplora_cache.go @@ -0,0 +1,138 @@ +package lwwallet + +import ( + "errors" + + "github.com/btcsuite/btcd/wire" +) + +// errNilCacheEntry is returned by the cached-value Size methods when +// the wrapped pointer is nil. The neutrino LRU treats Size errors as +// a refusal-to-insert, so a stray nil-wrapped value cannot fill the +// LRU map without consuming any byte budget. None of the Put sites +// in esplora.go pass nil values today; this is defense in depth. +var errNilCacheEntry = errors.New("nil cache entry") + +// Cache capacities are expressed in bytes (the unit returned by each +// cached value's Size()). They are sized for a typical client load: +// a few thousand cached transactions, a handful of recent blocks, and +// effectively unbounded room for the small fixed-size header types. +// +// These caches only hold immutable, hash-addressed data (transactions +// and blocks fetched by their content-hash, plus block headers). The +// cache fillers in esplora.go verify that the response body actually +// hashes to the requested key before insertion (txid for txs, +// BlockHash for headers and raw blocks); a buggy or compromised +// Esplora endpoint cannot pin an arbitrary entry under an +// attacker-chosen key. Once an entry is admitted, no reorg can stale +// it because the same hash can only ever map to the same content. +const ( + // txCacheCapacity caps the cumulative serialized size of the + // transaction cache. Sized for a few hundred recent txs on a + // lightweight wallet — we mostly need to absorb the + // duplicate fetch of the same tx by chainBackend confirmation + // lookup and the boarding tx-proof builder. + txCacheCapacity uint64 = 5 * 1024 * 1024 + + // rawBlockCacheCapacity caps the cumulative serialized size of + // the raw-block cache. Sized for ~5 recent full blocks (mainnet + // blocks approach 4 MiB), which covers the duplicate fetch by + // chainSvc filterBlock and chainBackend includeBlock paths + // without committing a hundreds-of-MiB resident set on + // Pi-class hardware. + rawBlockCacheCapacity uint64 = 20 * 1024 * 1024 + + // rawHeaderCacheCapacity caps the cumulative size of the + // raw-block-header cache. Each header is fixed at 80 bytes, so + // 1 MiB holds ~13 K headers. + rawHeaderCacheCapacity uint64 = 1 * 1024 * 1024 + + // blockHeaderCacheCapacity caps the cumulative size of the + // JSON block-header cache (height + timestamp metadata). + blockHeaderCacheCapacity uint64 = 1 * 1024 * 1024 +) + +// cachedTx wraps a *wire.MsgTx so it satisfies the neutrino cache.Value +// interface for LRU bookkeeping. The byte budget is pinned to the +// raw HTTP response length captured at Put time so the LRU's repeated +// Size() calls (one per Put plus one per eviction candidate) do not +// re-serialize the transaction each time. wire.MsgTx.SerializeSize +// walks every input and output, which is wasted work given the same +// number is already on the stack at the cache-fill site. +type cachedTx struct { + tx *wire.MsgTx + size uint64 +} + +// Size returns the cached serialized byte length of the wrapped +// transaction, or errNilCacheEntry when the wrapped pointer is nil +// so the LRU refuses the insert rather than admitting a 0-byte +// entry that would not count against the byte budget. +func (c cachedTx) Size() (uint64, error) { + if c.tx == nil { + return 0, errNilCacheEntry + } + + return c.size, nil +} + +// cachedBlock wraps a *wire.MsgBlock for LRU bookkeeping. Blocks can +// approach 4 MiB on mainnet, and SerializeSize iterates every +// transaction's inputs and outputs, so we pin the byte budget to the +// raw HTTP response length captured at Put time rather than walking +// the block on every LRU bookkeeping or eviction call. +type cachedBlock struct { + block *wire.MsgBlock + size uint64 +} + +// Size returns the cached serialized byte length of the wrapped +// block, or errNilCacheEntry when the wrapped pointer is nil. See +// cachedTx.Size for the rationale. +func (c cachedBlock) Size() (uint64, error) { + if c.block == nil { + return 0, errNilCacheEntry + } + + return c.size, nil +} + +// cachedRawHeader wraps a *wire.BlockHeader for LRU bookkeeping. +// Block headers are always 80 bytes when serialized. +type cachedRawHeader struct { + header *wire.BlockHeader +} + +// Size reports the fixed 80-byte serialized header length, or +// errNilCacheEntry when the wrapped pointer is nil. See cachedTx.Size +// for the rationale. +func (c cachedRawHeader) Size() (uint64, error) { + if c.header == nil { + return 0, errNilCacheEntry + } + + return 80, nil +} + +// cachedBlockHeader wraps an *esploraBlock for LRU bookkeeping. +type cachedBlockHeader struct { + header *esploraBlock +} + +// Size reports a real-ish heap footprint for the wrapped JSON +// header. The struct itself is fixed-size apart from the ID hex +// string, which on a well-behaved Esplora is always 64 bytes; we +// charge len(ID) + the size of the two integer fields plus a +// generous fixed overhead so a hostile Esplora that returns a +// pathologically long ID string still consumes its budget against +// the cache cap rather than slipping under a hardcoded 128-byte +// figure. +func (c cachedBlockHeader) Size() (uint64, error) { + if c.header == nil { + return 0, errNilCacheEntry + } + + const fixedOverhead = 64 + + return uint64(len(c.header.ID)) + fixedOverhead, nil +} diff --git a/lwwallet/esplora_cache_test.go b/lwwallet/esplora_cache_test.go new file mode 100644 index 000000000..9c039b5ee --- /dev/null +++ b/lwwallet/esplora_cache_test.go @@ -0,0 +1,196 @@ +package lwwallet + +import ( + "bytes" + "encoding/hex" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/btcsuite/btclog/v2" + "github.com/stretchr/testify/require" +) + +// TestEsploraCacheRejectsTxHashMismatch verifies that GetRawTx +// refuses to cache (and refuses to return) a tx body whose +// computed TxHash differs from the requested key. +func TestEsploraCacheRejectsTxHashMismatch(t *testing.T) { + t.Parallel() + + wantTxid := chainhash.HashH([]byte("victim")) + + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + // Always serve the same minimal raw tx no matter + // which txid was requested. Its actual TxHash will + // not equal wantTxid, so the verifier must reject. + _, _ = w.Write(minimalRawTx()) + }, + )) + defer srv.Close() + + c := NewEsploraClient(srv.URL, btclog.Disabled) + + tx, err := c.GetRawTx(wantTxid) + require.Error(t, err) + require.Nil(t, tx) + require.Contains(t, err.Error(), "tx hash mismatch") + + // A second request still goes through the network (no cache + // poison) and still rejects. + tx2, err2 := c.GetRawTx(wantTxid) + require.Error(t, err2) + require.Nil(t, tx2) +} + +// TestEsploraCacheRejectsBlockHashMismatch verifies that GetRawBlock +// refuses to cache a block whose computed BlockHash does not equal +// the requested key. +func TestEsploraCacheRejectsBlockHashMismatch(t *testing.T) { + t.Parallel() + + wantHash := chainhash.HashH([]byte("victim-block")) + + // Build a serialized block with a known header so we know its + // hash up front. + var attackerBlock wire.MsgBlock + attackerBlock.Header.Version = 4 + attackerBlock.Header.MerkleRoot = chainhash.HashH( + []byte("attacker-merkle"), + ) + + var blockBuf bytes.Buffer + require.NoError(t, attackerBlock.Serialize(&blockBuf)) + + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(blockBuf.Bytes()) + }, + )) + defer srv.Close() + + c := NewEsploraClient(srv.URL, btclog.Disabled) + + block, err := c.GetRawBlock(wantHash) + require.Error(t, err) + require.Nil(t, block) + require.Contains(t, err.Error(), "raw block hash mismatch") +} + +// TestEsploraCacheRejectsBlockHeaderHashMismatch verifies the JSON +// /block/:hash filler refuses entries where the response's id field +// doesn't equal the requested key. +func TestEsploraCacheRejectsBlockHeaderHashMismatch(t *testing.T) { + t.Parallel() + + wantHash := chainhash.HashH([]byte("victim-jsonblock")) + otherHash := chainhash.HashH([]byte("other")) + + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprintf(w, + `{"id":%q,"height":42,"timestamp":1000}`, + otherHash.String()) + }, + )) + defer srv.Close() + + c := NewEsploraClient(srv.URL, btclog.Disabled) + + hdr, err := c.GetBlockHeader(wantHash) + require.Error(t, err) + require.Nil(t, hdr) + require.Contains(t, err.Error(), "block id mismatch") +} + +// TestEsploraCacheTxHitCount verifies that a successful GetRawTx is +// memoized: a second call returns the cached value without hitting +// the network. We assert this by counting HTTP requests against the +// stub. +func TestEsploraCacheTxHitCount(t *testing.T) { + t.Parallel() + + // Compute the txid of minimalRawTx so the verifier admits + // the response. + var tx wire.MsgTx + require.NoError(t, tx.Deserialize(bytes.NewReader(minimalRawTx()))) + txid := tx.TxHash() + + var hits atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + hits.Add(1) + _, _ = w.Write(minimalRawTx()) + }, + )) + defer srv.Close() + + c := NewEsploraClient(srv.URL, btclog.Disabled) + + // First call: cache miss, one HTTP hit. + tx1, err := c.GetRawTx(txid) + require.NoError(t, err) + require.NotNil(t, tx1) + require.Equal(t, int64(1), hits.Load()) + + // Second call: cache hit, no additional HTTP hit. + tx2, err := c.GetRawTx(txid) + require.NoError(t, err) + require.Same(t, tx1, tx2, + "expected cache to return same pointer") + require.Equal(t, int64(1), hits.Load(), + "second GetRawTx should not have hit the network") +} + +// TestCachedSizeRefusesNil verifies the Size methods on each cache +// value type return errNilCacheEntry when the wrapped pointer is +// nil. This is the M-6 defense that prevents a 0-byte LRU entry +// from filling the map. +func TestCachedSizeRefusesNil(t *testing.T) { + t.Parallel() + + checkRefuse := func(name string, sz func() (uint64, error)) { + t.Helper() + _, err := sz() + require.ErrorIs(t, err, errNilCacheEntry, + "%s: expected errNilCacheEntry", name) + } + + checkRefuse("cachedTx", cachedTx{}.Size) + checkRefuse("cachedBlock", cachedBlock{}.Size) + checkRefuse("cachedRawHeader", cachedRawHeader{}.Size) + checkRefuse("cachedBlockHeader", cachedBlockHeader{}.Size) +} + +// TestCachedBlockHeaderSizeReflectsIDLen verifies the M-3 fix: +// cachedBlockHeader's Size accounts for the variable-length JSON +// id string so a hostile Esplora response with a pathologically +// long id consumes proportional cache budget rather than slipping +// under a hardcoded constant. +func TestCachedBlockHeaderSizeReflectsIDLen(t *testing.T) { + t.Parallel() + + short := cachedBlockHeader{ + header: &esploraBlock{ID: "abcd"}, + } + long := cachedBlockHeader{ + header: &esploraBlock{ + ID: hex.EncodeToString( + make([]byte, 1024), + ), + }, + } + + shortSize, err := short.Size() + require.NoError(t, err) + + longSize, err := long.Size() + require.NoError(t, err) + + require.Greater(t, longSize, shortSize, + "longer ID must contribute to cache budget") +} diff --git a/lwwallet/esplora_chain.go b/lwwallet/esplora_chain.go index 6e05e77f1..9f0982a83 100644 --- a/lwwallet/esplora_chain.go +++ b/lwwallet/esplora_chain.go @@ -19,14 +19,18 @@ import ( ) // EsploraChainService implements btcwallet's chain.Interface using the -// Esplora REST API. It provides the blockchain access layer that -// btcwallet needs for wallet synchronization, block notifications, and -// transaction broadcasting. The service polls Esplora for new blocks -// at a configurable interval and forwards BlockConnected notifications -// to btcwallet via the Notifications() channel. +// Esplora REST API. It is a passive consumer of a shared TipPoller: it +// does not poll Esplora for new blocks itself, but instead subscribes +// to the poller's tip stream and translates each TipBlock event into +// the FilteredBlockConnected + BlockConnected notification pair that +// btcwallet's wallet syncer expects. type EsploraChainService struct { - esplora *EsploraClient - pollInterval time.Duration + esplora *EsploraClient + + // tipPoller is the shared tip-event source. The chain service + // never owns this poller; lifecycle is the caller's + // responsibility (typically the lwwallet integrated wallet). + tipPoller *TipPoller // log is the structured logger for this chain service instance. log btclog.Logger @@ -44,24 +48,25 @@ type EsploraChainService struct { // transactions for RelevantTx notifications. watchedAddrs map[string]btcutil.Address - // bestBlock caches the current chain tip, updated by the poll - // loop on each new block. + // bestBlock caches the current chain tip, updated on each + // processed TipBlock event. bestBlock waddrmgr.BlockStamp - quit chan struct{} - wg sync.WaitGroup + quit chan struct{} + stopOnce sync.Once + wg sync.WaitGroup } // NewEsploraChainService creates a new chain.Interface backed by the -// Esplora REST API. The pollInterval controls how frequently the -// service checks for new blocks. +// Esplora REST API. The provided TipPoller drives new-block +// detection; the caller is responsible for starting and stopping it. func NewEsploraChainService(esplora *EsploraClient, - pollInterval time.Duration, + tipPoller *TipPoller, logger btclog.Logger) *EsploraChainService { return &EsploraChainService{ esplora: esplora, - pollInterval: pollInterval, + tipPoller: tipPoller, log: logger, notifications: make(chan interface{}, 100), watchedAddrs: make(map[string]btcutil.Address), @@ -69,33 +74,22 @@ func NewEsploraChainService(esplora *EsploraClient, } } -// Start fetches the initial chain tip and starts the polling goroutine -// that sends BlockConnected notifications to btcwallet. +// Start seeds the initial chain tip from the configured TipPoller +// (which the caller must have started already) and spawns the +// goroutine that translates each TipBlock event into btcwallet +// chain notifications. func (s *EsploraChainService) Start(ctx context.Context) error { - // Fetch the initial chain tip so BlockStamp() returns correct - // values immediately. We get height first, then resolve the - // hash for that specific height to avoid drift if a new block - // arrives between the two HTTP calls. - tipHeight, err := s.esplora.GetTipHeight() - if err != nil { - return fmt.Errorf("get initial tip height: %w", err) - } - - tipHash, err := s.esplora.GetBlockHashByHeight(tipHeight) + tipHeight, tipHash, tipTime, sub, err := + s.tipPoller.BestBlockAndSubscribe() if err != nil { - return fmt.Errorf("get initial tip hash: %w", err) - } - - tipHeader, err := s.esplora.GetBlockHeader(tipHash) - if err != nil { - return fmt.Errorf("get initial tip header: %w", err) + return fmt.Errorf("subscribe to tip poller: %w", err) } s.mu.Lock() s.bestBlock = waddrmgr.BlockStamp{ Height: tipHeight, Hash: tipHash, - Timestamp: time.Unix(tipHeader.Timestamp, 0), + Timestamp: tipTime, } s.mu.Unlock() @@ -104,7 +98,7 @@ func (s *EsploraChainService) Start(ctx context.Context) error { s.notifications <- chain.ClientConnected{} s.wg.Add(1) - go s.pollLoop() + go s.handleTipEvents(ctx, sub) s.log.InfoS(ctx, "Esplora chain service started", slog.Int("tip_height", int(tipHeight)), @@ -113,19 +107,17 @@ func (s *EsploraChainService) Start(ctx context.Context) error { return nil } -// Stop signals the polling goroutine to exit. +// Stop signals the polling goroutine to exit. Stop is idempotent +// and safe to call concurrently from multiple goroutines; a +// sync.Once guards the close so two simultaneous Stop calls cannot +// both reach close(s.quit) and panic on a double close. func (s *EsploraChainService) Stop() { - select { - case <-s.quit: - // Already stopped. - return - - default: + s.stopOnce.Do(func() { s.log.InfoS(context.Background(), "Stopping Esplora chain service") close(s.quit) - } + }) } // WaitForShutdown blocks until the polling goroutine has exited. @@ -134,23 +126,15 @@ func (s *EsploraChainService) WaitForShutdown() { } // GetBestBlock returns the hash and height of the current best block. -// Height is fetched first, then the hash is resolved for that specific -// height to avoid returning mismatched values if a new block arrives -// between the two HTTP calls. +// The shared TipPoller already maintains a consistent (height, hash, +// timestamp) triple under its own mutex, resolved against the height +// it actually observed; reading that snapshot here avoids two live +// HTTP round trips per call (and the original TOCTOU between the two +// independent fetches that this method used to defend against). func (s *EsploraChainService) GetBestBlock() ( *chainhash.Hash, int32, error) { - height, err := s.esplora.GetTipHeight() - if err != nil { - return nil, 0, fmt.Errorf( - "get best block height: %w", err, - ) - } - - hash, err := s.esplora.GetBlockHashByHeight(height) - if err != nil { - return nil, 0, fmt.Errorf("get best block hash: %w", err) - } + height, hash, _ := s.tipPoller.BestBlock() return &hash, height, nil } @@ -631,19 +615,29 @@ func (s *EsploraChainService) MapRPCErr(err error) error { return err } -// pollLoop periodically checks for new blocks and sends -// BlockConnected notifications to btcwallet. The loop runs until -// Stop() is called. -func (s *EsploraChainService) pollLoop() { - defer s.wg.Done() +// handleTipEvents drains TipBlock events from the shared poller and +// translates each event into the FilteredBlockConnected + +// BlockConnected notification pair that btcwallet's wallet syncer +// expects. The loop exits when the chain service is stopped, when +// the poller signals shutdown via Quit, or when the subscription's +// Updates channel is closed by Cancel. +func (s *EsploraChainService) handleTipEvents(ctx context.Context, + sub *TipSubscription) { - ticker := time.NewTicker(s.pollInterval) - defer ticker.Stop() + defer s.wg.Done() + defer sub.Cancel() for { select { - case <-ticker.C: - s.pollForBlocks() + case event, ok := <-sub.Updates(): + if !ok { + return + } + + s.processTipEvent(ctx, event) + + case <-sub.Quit(): + return case <-s.quit: return @@ -651,136 +645,98 @@ func (s *EsploraChainService) pollLoop() { } } -// pollForBlocks checks for new blocks since the last known tip and -// sends FilteredBlockConnected and BlockConnected notifications for -// each new block. FilteredBlockConnected carries relevant transactions -// (matching watched addresses) so btcwallet can track UTXOs, while -// BlockConnected updates the wallet's sync height. -func (s *EsploraChainService) pollForBlocks() { - newHeight, err := s.esplora.GetTipHeight() - if err != nil { - s.log.WarnS(context.Background(), - "Chain service poll tip height failed", err) +// processTipEvent applies one TipBlock to btcwallet's notification +// channel. The full block is only fetched when there is at least one +// watched address; without watchers there can be no relevant +// transactions, so the EsploraClient's raw-block call (and the +// associated bandwidth) is skipped. +func (s *EsploraChainService) processTipEvent(ctx context.Context, + event *TipBlock) { + if event == nil || event.Header == nil { return } - s.mu.Lock() - oldHeight := s.bestBlock.Height - s.mu.Unlock() - - if newHeight <= oldHeight { - return + blockMeta := wtxmgr.BlockMeta{ + Block: wtxmgr.Block{ + Hash: event.Hash, + Height: event.Height, + }, + Time: time.Unix(event.Header.Timestamp, 0), } - s.log.DebugS(context.Background(), - "New blocks detected", - slog.Int("old_height", int(oldHeight)), - slog.Int("new_height", int(newHeight))) - - // Process each new block in order. - for height := oldHeight + 1; height <= newHeight; height++ { - blockHash, err := s.esplora.GetBlockHashByHeight(height) + // Build pkScript lookup from currently watched addresses so + // we can detect relevant transactions in this block. + s.mu.Lock() + watchedScripts := make(map[string]struct{}, len(s.watchedAddrs)) + for _, addr := range s.watchedAddrs { + pkScript, err := txscript.PayToAddrScript(addr) if err != nil { - s.log.WarnS(context.Background(), - "Chain service poll block hash failed", - err) - - return + continue } - blockInfo, err := s.esplora.GetBlockHeader(blockHash) + watchedScripts[string(pkScript)] = struct{}{} + } + s.mu.Unlock() + + // Filter block for relevant transactions if we have any + // watched addresses. This requires fetching the full block + // from Esplora; the EsploraClient memoizes the raw block by + // hash so concurrent consumers (e.g. boarding tx-proof + // builders) reuse the response. + var relevantTxs []*wtxmgr.TxRecord + if len(watchedScripts) > 0 { + block, err := s.esplora.GetRawBlock(event.Hash) if err != nil { - s.log.WarnS(context.Background(), - "Chain service poll block info failed", - err) + s.log.WarnS(ctx, + "Chain service block fetch failed", err, + slog.Int("height", int(event.Height))) return } - blockMeta := wtxmgr.BlockMeta{ - Block: wtxmgr.Block{ - Hash: blockHash, - Height: height, - }, - Time: time.Unix(blockInfo.Timestamp, 0), - } - - // Build pkScript lookup from currently watched addresses - // so we can detect relevant transactions in this block. - s.mu.Lock() - watchedScripts := make(map[string]struct{}, - len(s.watchedAddrs)) - for _, addr := range s.watchedAddrs { - pkScript, err := txscript.PayToAddrScript(addr) - if err != nil { - continue - } - - watchedScripts[string(pkScript)] = struct{}{} - } - s.mu.Unlock() - - // Filter block for relevant transactions if we have - // any watched addresses. This requires fetching the - // full block from Esplora. - var relevantTxs []*wtxmgr.TxRecord - if len(watchedScripts) > 0 { - block, err := s.esplora.GetRawBlock(blockHash) - if err != nil { - s.log.WarnS(context.Background(), - "Chain service poll block "+ - "fetch failed", err) - - return - } - - relevantTxs = s.filterBlockTxs( - block, watchedScripts, - blockMeta.Time, - ) - } + relevantTxs = s.filterBlockTxs( + block, watchedScripts, blockMeta.Time, + ) + } - // Send FilteredBlockConnected with relevant - // transactions so btcwallet processes them via - // addRelevantTx. This is how btcwallet learns about - // transactions paying to wallet-owned addresses. - // - // We use select with quit to prevent blocking - // indefinitely if the channel is full during initial - // sync (when handleChainNotifications is busy with - // syncWithChain/recovery). - select { - case s.notifications <- chain.FilteredBlockConnected{ - Block: &blockMeta, - RelevantTxs: relevantTxs, - }: + // Send FilteredBlockConnected with relevant transactions so + // btcwallet processes them via addRelevantTx. This is how + // btcwallet learns about transactions paying to wallet-owned + // addresses. + // + // We use select with quit to prevent blocking indefinitely if + // the channel is full during initial sync (when + // handleChainNotifications is busy with syncWithChain / + // recovery). + select { + case s.notifications <- chain.FilteredBlockConnected{ + Block: &blockMeta, + RelevantTxs: relevantTxs, + }: - case <-s.quit: - return - } + case <-s.quit: + return + } - // Send BlockConnected to update btcwallet's sync - // height. FilteredBlockConnected only processes - // transactions but does not update the sync height. - select { - case s.notifications <- chain.BlockConnected( - blockMeta, - ): + // Send BlockConnected to update btcwallet's sync height. + // FilteredBlockConnected only processes transactions but does + // not update the sync height. + select { + case s.notifications <- chain.BlockConnected(blockMeta): - case <-s.quit: - return - } + case <-s.quit: + return + } - // Update the cached best block. - s.mu.Lock() - s.bestBlock = waddrmgr.BlockStamp{ - Height: height, - Hash: blockHash, - Timestamp: blockMeta.Time, - } - s.mu.Unlock() + // Update the cached best block. + s.mu.Lock() + s.bestBlock = waddrmgr.BlockStamp{ + Height: event.Height, + Hash: event.Hash, + Timestamp: blockMeta.Time, } + s.mu.Unlock() } // filterBlockTxs checks all transactions in the block against the diff --git a/lwwallet/subscribe.go b/lwwallet/subscribe.go new file mode 100644 index 000000000..7514fbf1f --- /dev/null +++ b/lwwallet/subscribe.go @@ -0,0 +1,175 @@ +package lwwallet + +import ( + "context" + "fmt" + "log/slog" + "sync" + + "github.com/btcsuite/btclog/v2" + "github.com/lightningnetwork/lnd/subscribe" +) + +// EventServer is a thin generic wrapper around lnd's subscribe.Server +// that delivers typed events to its subscribers. The underlying +// subscribe.Server already handles every concurrency concern we care +// about — single-goroutine subscriber handler (no send-on-closed- +// channel race possible by construction), per-client unbounded +// queue (a slow consumer cannot wedge the broadcaster), idempotent +// Cancel — so EventServer's only job is to keep callers from having +// to type-assert away the interface{} that subscribe.Server speaks. +type EventServer[T any] struct { + inner *subscribe.Server + log btclog.Logger +} + +// NewEventServer constructs a typed event server. Start must be +// called before SendUpdate or Subscribe. +func NewEventServer[T any](log btclog.Logger) *EventServer[T] { + return &EventServer[T]{ + inner: subscribe.NewServer(), + log: log, + } +} + +// Start makes the server ready to accept subscriptions and updates. +// Start is idempotent. +func (s *EventServer[T]) Start() error { + return s.inner.Start() +} + +// Stop tears down the subscriber handler and closes every active +// subscription. Stop is idempotent and safe to call concurrently. +func (s *EventServer[T]) Stop() error { + return s.inner.Stop() +} + +// SendUpdate broadcasts a typed event to every active subscriber. +// The call returns ErrServerShuttingDown if the server has been +// stopped. +func (s *EventServer[T]) SendUpdate(event T) error { + return s.inner.SendUpdate(event) +} + +// Subscribe returns a typed subscription. The returned subscription +// owns a translator goroutine that converts subscribe.Server's +// untyped updates into the typed channel exposed via Updates(). +// Cancel must be called to release the subscription. +func (s *EventServer[T]) Subscribe() (*Subscription[T], error) { + client, err := s.inner.Subscribe() + if err != nil { + return nil, fmt.Errorf("subscribe: %w", err) + } + + sub := &Subscription[T]{ + inner: client, + out: make(chan T, 1), + quit: make(chan struct{}), + log: s.log, + } + + sub.wg.Add(1) + go sub.translate() + + return sub, nil +} + +// Subscription is a typed handle on an active subscribe.Client. It +// converts the inner channel of interface{} updates into a typed +// channel of T events on a dedicated translator goroutine. +type Subscription[T any] struct { + inner *subscribe.Client + + out chan T + quit chan struct{} + + cancelOnce sync.Once + wg sync.WaitGroup + + log btclog.Logger +} + +// Updates returns the typed event channel. The channel is closed +// when the subscription is cancelled or the upstream server is +// stopped. +func (s *Subscription[T]) Updates() <-chan T { + return s.out +} + +// Quit returns a channel closed when the upstream server is +// shutting down. Consumers should select on Updates and Quit to +// react to either a new event or the server going away. +func (s *Subscription[T]) Quit() <-chan struct{} { + return s.inner.Quit() +} + +// Cancel deregisters the subscription and waits for the translator +// goroutine to exit. Cancel is idempotent and safe to call from +// any goroutine; calling it from inside an Updates handler is fine +// because the translator unblocks via the local quit channel. +func (s *Subscription[T]) Cancel() { + s.cancelOnce.Do(func() { + // Close the local quit first so the translator can + // escape a parked send into the typed out channel + // before we block on inner.Cancel(). inner.Cancel + // blocks until the server handler removes us, which in + // turn closes inner.Quit(); the translator picks that + // up next. + close(s.quit) + s.inner.Cancel() + }) + + s.wg.Wait() +} + +// translate is the per-subscription goroutine that pulls untyped +// updates off the inner subscribe.Client, asserts them to T, and +// forwards them to the typed out channel. The goroutine exits when +// either the inner server signals shutdown or the local Cancel +// fires. +func (s *Subscription[T]) translate() { + defer s.wg.Done() + defer close(s.out) + + for { + select { + case upd, ok := <-s.inner.Updates(): + if !ok { + return + } + + typed, ok := upd.(T) + if !ok { + // The inner server should only ever + // deliver T values because SendUpdate + // only accepts T; a type mismatch here + // is a programming bug. + s.log.ErrorS(context.Background(), + "Event server type assertion failed", + fmt.Errorf("got %T", upd), + slog.String( + "want", fmt.Sprintf("%T", + *new(T)), + )) + + continue + } + + select { + case s.out <- typed: + + case <-s.quit: + return + + case <-s.inner.Quit(): + return + } + + case <-s.inner.Quit(): + return + + case <-s.quit: + return + } + } +} diff --git a/lwwallet/tip_poller.go b/lwwallet/tip_poller.go new file mode 100644 index 000000000..ac8100003 --- /dev/null +++ b/lwwallet/tip_poller.go @@ -0,0 +1,360 @@ +package lwwallet + +import ( + "context" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btclog/v2" +) + +// TipBlock describes a newly detected block emitted by TipPoller. +// Each subscriber receives one TipBlock per advance: when the tip +// moves from oldHeight to newHeight the poller fans out (newHeight - +// oldHeight) events in monotonically increasing height order so +// downstream consumers can apply per-block work without skipping or +// re-ordering. +type TipBlock struct { + // Height is the block's height in the canonical chain. + Height int32 + + // Hash is the block's identifying hash. Together with Height + // this uniquely identifies a block for downstream chain-watch + // handlers. + Hash chainhash.Hash + + // Header carries the JSON header response (height + timestamp); + // it is included so subscribers do not need to re-fetch it from + // Esplora. Will be non-nil for events emitted from the poll + // loop. + Header *esploraBlock +} + +// TipSubscription is the typed handle returned by TipPoller.Subscribe. +// It is a thin alias over Subscription[*TipBlock] that exists purely +// for ergonomic call-site naming. +type TipSubscription = Subscription[*TipBlock] + +// TipPoller is the single source of truth for the lwwallet chain +// tip. Exactly one polling goroutine periodically asks the Esplora +// backend for the current best height. When new blocks are detected +// the poller walks oldHeight+1 → newHeight, resolves each block's +// hash + header (which the EsploraClient's caches will absorb on +// repeat callers) and broadcasts a TipBlock event to every active +// subscriber via the embedded EventServer. +// +// Centralizing the tip stream lets multiple downstream chain +// watchers (the btcwallet chain.Interface adapter and the +// chainsource.ChainBackend) share a single Esplora call cadence +// instead of polling independently. Subscriber lifecycle is +// delegated to lnd's subscribe.Server (wrapped by EventServer) so +// the poller itself never has to reason about close-channel races +// or slow-subscriber back-pressure. +type TipPoller struct { + esplora *EsploraClient + pollInterval time.Duration + log btclog.Logger + + // events is the typed event server that fans TipBlock updates + // out to all active subscribers. Its Start/Stop are driven by + // TipPoller.Start/Stop. + events *EventServer[*TipBlock] + + // mu guards the cached tip so BestBlock readers see a + // consistent height/hash/timestamp triple. + mu sync.Mutex + tipHeight int32 + tipHash chainhash.Hash + tipTime time.Time + + // started gates re-entrant Start calls; the underlying + // EventServer is also idempotent on Start. + started bool + + quit chan struct{} + stopOnce sync.Once + wg sync.WaitGroup +} + +// NewTipPoller constructs a TipPoller bound to the given Esplora +// client. The poll interval controls how often the goroutine asks +// Esplora for the latest tip height; subscribers do not influence +// the cadence. +func NewTipPoller(esplora *EsploraClient, pollInterval time.Duration, + log btclog.Logger) *TipPoller { + + return &TipPoller{ + esplora: esplora, + pollInterval: pollInterval, + log: log, + events: NewEventServer[*TipBlock](log), + quit: make(chan struct{}), + } +} + +// Start fetches the initial tip synchronously, starts the embedded +// event server, and then spawns the polling goroutine. Returning +// early without starting the goroutine when the initial fetch fails +// preserves the existing chain-backend contract: a misconfigured +// Esplora endpoint must surface at startup, not silently translate +// into a "stuck at height 0" runtime symptom. +func (t *TipPoller) Start() error { + // Claim the started slot atomically before doing any network + // I/O. Setting the flag in the same critical section as the + // check prevents two concurrent Start() calls from both passing + // the guard, both fetching the initial tip, and both spawning + // poll goroutines that would race on tipHeight/tipHash and + // double the SendUpdate cadence to subscribers. + t.mu.Lock() + if t.started { + t.mu.Unlock() + + return fmt.Errorf("tip poller already started") + } + t.started = true + t.mu.Unlock() + + // resetStarted releases the started slot on any failure path + // below so a caller that retries Start on the same instance + // after a transient Esplora failure is not permanently locked + // out by the atomic claim above. + resetStarted := func() { + t.mu.Lock() + t.started = false + t.mu.Unlock() + } + + height, err := t.esplora.GetTipHeight() + if err != nil { + resetStarted() + return fmt.Errorf("get initial tip height: %w", err) + } + + hash, err := t.esplora.GetBlockHashByHeight(height) + if err != nil { + resetStarted() + return fmt.Errorf("get initial tip hash: %w", err) + } + + // Fetching the initial header is best-effort: a missing + // timestamp at start-up is not load-bearing — the next + // confirmed block updates it from the live header anyway — + // and treating it as fatal would force every test mock to + // serve `/block/` for the initial seed even when the + // tested path never reads `tipTime`. + var tipTime time.Time + if header, hdrErr := t.esplora.GetBlockHeader(hash); hdrErr == nil { + tipTime = time.Unix(header.Timestamp, 0) + } else { + t.log.WarnS(context.Background(), + "Tip poller initial header fetch failed", + hdrErr, slog.String("hash", hash.String())) + } + + if err := t.events.Start(); err != nil { + resetStarted() + return fmt.Errorf("start event server: %w", err) + } + + t.mu.Lock() + t.tipHeight = height + t.tipHash = hash + t.tipTime = tipTime + t.mu.Unlock() + + t.wg.Add(1) + go t.pollLoop() + + t.log.InfoS(context.Background(), "Tip poller started", + slog.Int("tip_height", int(height)), + slog.String("tip_hash", hash.String())) + + return nil +} + +// Stop signals the polling goroutine to exit, waits for it to +// drain, and tears down the event server. Stop is idempotent; the +// second call returns immediately after the first has finished. +func (t *TipPoller) Stop() { + t.stopOnce.Do(func() { + close(t.quit) + }) + + t.wg.Wait() + + // Stop the event server after the poll loop has exited so + // that no SendUpdate is in flight when the server tears down + // its subscriber handler. + if err := t.events.Stop(); err != nil { + t.log.WarnS(context.Background(), + "Tip poller event server stop returned error", err) + } +} + +// BestBlock returns a snapshot of the currently cached tip. Callers +// that just need to read the chain tip without subscribing to new +// block events should use this instead of issuing a fresh Esplora +// request. +func (t *TipPoller) BestBlock() (int32, chainhash.Hash, time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + + return t.tipHeight, t.tipHash, t.tipTime +} + +// Subscribe returns a typed subscription that receives a TipBlock +// for every new block the poller observes. The caller must invoke +// Cancel on the returned subscription when finished. The first +// event a subscriber receives corresponds to the first block +// detected after Subscribe returns; callers that need the current +// tip atomically with the subscription should use +// BestBlockAndSubscribe instead — Subscribe alone leaves a small +// window where a tip event could land between a separate +// BestBlock() read and the subscription registering. +func (t *TipPoller) Subscribe() (*TipSubscription, error) { + return t.events.Subscribe() +} + +// BestBlockAndSubscribe atomically reads the current cached tip and +// registers a new subscription. The poll goroutine holds t.mu +// during the {update tip + SendUpdate} pair, and this function +// holds t.mu around {Subscribe + read tip}, so the two operations +// serialize: callers either see the old tip and receive the next +// tip event, or see the new tip and skip ahead to events strictly +// after it. The non-atomic Subscribe + BestBlock pair leaves a +// race where a tip event could land between the read and the +// register, causing a missed event or a duplicated one. +func (t *TipPoller) BestBlockAndSubscribe() (int32, chainhash.Hash, + time.Time, *TipSubscription, error) { + + t.mu.Lock() + defer t.mu.Unlock() + + sub, err := t.events.Subscribe() + if err != nil { + return 0, chainhash.Hash{}, time.Time{}, nil, + fmt.Errorf("subscribe to tip poller: %w", err) + } + + return t.tipHeight, t.tipHash, t.tipTime, sub, nil +} + +// pollLoop is the single tip-polling goroutine. It ticks at +// pollInterval, asks Esplora for the latest tip, and walks the +// gap from the cached tip to the new tip emitting one TipBlock per +// step. +func (t *TipPoller) pollLoop() { + defer t.wg.Done() + + ticker := time.NewTicker(t.pollInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + t.poll() + + case <-t.quit: + return + } + } +} + +// poll performs one tip-detection cycle. On detected progress it +// fetches the hash and header for each new height, broadcasts a +// TipBlock to every subscriber, and advances the cached tip +// monotonically. A failure to fetch any single block aborts the +// remainder of the cycle so subscribers never see an out-of-order +// event; the next tick re-attempts from the same starting point. +func (t *TipPoller) poll() { + newHeight, err := t.esplora.GetTipHeight() + if err != nil { + t.log.WarnS(context.Background(), + "Tip poller GetTipHeight failed", err) + + return + } + + t.mu.Lock() + oldHeight := t.tipHeight + t.mu.Unlock() + + // Known limitation: a same-height reorg (block at height N + // replaced by a different block at height N) is invisible to + // this loop until the chain advances to N+1, because we gate + // progress on height alone rather than (height, hash). This + // matches the behavior of the per-component pollers that + // preceded the unified TipPoller and has historically been + // acceptable for lwwallet's confirmation-target use case + // (downstream callers re-check status against Esplora on every + // tip event, so a stale hash at height N converges within one + // extra tip advance). Documented here so it is not filed as a + // regression by a future reader. + if newHeight <= oldHeight { + return + } + + t.log.DebugS(context.Background(), "Tip poller advancing", + slog.Int("old_height", int(oldHeight)), + slog.Int("new_height", int(newHeight))) + + for height := oldHeight + 1; height <= newHeight; height++ { + hash, err := t.esplora.GetBlockHashByHeight(height) + if err != nil { + t.log.WarnS(context.Background(), + "Tip poller GetBlockHashByHeight failed", + err, slog.Int("height", int(height))) + + return + } + + header, err := t.esplora.GetBlockHeader(hash) + if err != nil { + t.log.WarnS(context.Background(), + "Tip poller GetBlockHeader failed", err, + slog.String("hash", hash.String())) + + return + } + + event := &TipBlock{ + Height: height, + Hash: hash, + Header: header, + } + + // Hold t.mu across the {update tip + SendUpdate} + // pair so BestBlockAndSubscribe can serialize against + // it: a subscriber that acquires t.mu before us reads + // the OLD tip and is guaranteed to receive THIS event + // once it Subscribes (subscribe.Server's handler is + // single-threaded over Subscribe and SendUpdate, so + // our SendUpdate enqueues behind their Subscribe). A + // subscriber that acquires t.mu after us reads the + // NEW tip and will see only events strictly newer + // than this one. Without holding t.mu here a tip + // reader+subscriber pair has a small window where it + // can read the new tip but miss this event entirely. + t.mu.Lock() + t.tipHeight = height + t.tipHash = hash + t.tipTime = time.Unix(header.Timestamp, 0) + sendErr := t.events.SendUpdate(event) + t.mu.Unlock() + + // SendUpdate failures only happen when the embedded + // subscribe.Server is shutting down; log and exit so + // we do not advance the cached tip past an event we + // failed to fan out. + if sendErr != nil { + t.log.WarnS(context.Background(), + "Tip poller send update failed", sendErr, + slog.Int("height", int(height))) + + return + } + } +} diff --git a/lwwallet/tip_poller_test.go b/lwwallet/tip_poller_test.go new file mode 100644 index 000000000..f0acfc126 --- /dev/null +++ b/lwwallet/tip_poller_test.go @@ -0,0 +1,626 @@ +package lwwallet + +import ( + "fmt" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btclog/v2" + "github.com/stretchr/testify/require" +) + +// stubChain is a tiny test fixture that simulates an Esplora chain +// where the tip height can be advanced under test control. It is +// independent of the larger mockEsploraServer helper so tip-poller +// tests can drive the chain forward synchronously. +type stubChain struct { + mu sync.Mutex + tipHeight int32 + + // hashAt[h] is the hash for height h. We pre-populate as the + // tip advances so GetBlockHashByHeight resolves consistently. + hashAt map[int32]chainhash.Hash +} + +func newStubChain(tipHeight int32) *stubChain { + c := &stubChain{ + tipHeight: tipHeight, + hashAt: make(map[int32]chainhash.Hash), + } + + for h := int32(0); h <= tipHeight; h++ { + c.hashAt[h] = chainhash.HashH( + []byte(fmt.Sprintf("block-%d", h)), + ) + } + + return c +} + +func (c *stubChain) advance(t *testing.T, n int32) { + t.Helper() + + c.mu.Lock() + defer c.mu.Unlock() + + for i := int32(1); i <= n; i++ { + h := c.tipHeight + i + c.hashAt[h] = chainhash.HashH( + []byte(fmt.Sprintf("block-%d", h)), + ) + } + + c.tipHeight += n +} + +// stubEsploraHandler returns an http.HandlerFunc that serves the +// tip-poller's GET requests against the stubChain. Only the routes +// the poller actually hits are implemented. +func stubEsploraHandler(t *testing.T, + chain *stubChain) http.HandlerFunc { + + t.Helper() + + return func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/blocks/tip/height": + chain.mu.Lock() + h := chain.tipHeight + chain.mu.Unlock() + + _, _ = fmt.Fprint(w, h) + + case len(r.URL.Path) > len("/block-height/") && + r.URL.Path[:len("/block-height/")] == + "/block-height/": + + heightStr := r.URL.Path[len("/block-height/"):] + var height int32 + _, err := fmt.Sscanf(heightStr, "%d", &height) + require.NoError(t, err) + + chain.mu.Lock() + h, ok := chain.hashAt[height] + chain.mu.Unlock() + + if !ok { + http.Error(w, "not found", + http.StatusNotFound) + + return + } + + _, _ = fmt.Fprint(w, h.String()) + + case len(r.URL.Path) > len("/block/") && + r.URL.Path[:len("/block/")] == "/block/": + + rest := r.URL.Path[len("/block/"):] + + // Strip optional /header or /raw suffix. + hashStr := rest + suffix := "" + for i := 0; i < len(rest); i++ { + if rest[i] == '/' { + hashStr = rest[:i] + suffix = rest[i:] + break + } + } + + h, err := chainhash.NewHashFromStr(hashStr) + require.NoError(t, err) + + // Find the height for this hash so we can + // craft a header whose BlockHash actually + // matches. + var height int32 = -1 + chain.mu.Lock() + for hh, hash := range chain.hashAt { + if hash == *h { + height = hh + + break + } + } + chain.mu.Unlock() + + if height < 0 { + http.Error(w, "not found", + http.StatusNotFound) + + return + } + + switch suffix { + case "": + // JSON header. + _, _ = fmt.Fprintf(w, + `{"id":%q,"height":%d,`+ + `"timestamp":%d}`, + h.String(), height, + int64(height)*600) + + default: + // Raw header / raw block — synthesize a + // header whose serialized bytes hash to + // h. We take the simple route of using + // h's bytes themselves: the header + // hash-verifier compares header.BlockHash() + // to the requested h, so any header that + // happens to round-trip works. Synthesizing + // such a header from a target hash is + // effectively impossible, so for these + // suffix variants we return 501 — tests + // that need them must use the cache + // pre-fill path. + http.Error(w, "not implemented", + http.StatusNotImplemented) + } + + default: + http.Error(w, "not found", + http.StatusNotFound) + } + } +} + +// TestTipPollerStartStop verifies a clean start/stop cycle with no +// subscribers. +func TestTipPollerStartStop(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 20*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + + height, _, _ := tp.BestBlock() + require.Equal(t, int32(100), height) + + tp.Stop() + + // Stop is idempotent. + tp.Stop() +} + +// TestTipPollerStartTwiceFails ensures double-Start is rejected so +// callers see a clear error rather than silently spawning two poll +// goroutines. +func TestTipPollerStartTwiceFails(t *testing.T) { + t.Parallel() + + chain := newStubChain(50) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 20*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + require.Error(t, tp.Start()) +} + +// TestTipPollerMultiBlockCatchUp verifies that when the chain +// advances by N blocks between polls, the poller emits N events in +// strict height order. +func TestTipPollerMultiBlockCatchUp(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 10*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + _, _, _, sub, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + defer sub.Cancel() + + chain.advance(t, 5) + + for expected := int32(101); expected <= 105; expected++ { + select { + case ev := <-sub.Updates(): + require.NotNil(t, ev) + require.Equal(t, expected, ev.Height, + "events arrived out of order") + + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for height %d", + expected) + } + } +} + +// TestTipPollerSubscribeCancelRace exercises the historical +// send-on-closed-channel hazard: spam Subscribe and cancel +// concurrently with active broadcasts. A pre-fix poller would +// panic; the subscribe.Server-backed implementation must not. +// +// The test runs under the race detector on CI so any latent +// race would surface. +func TestTipPollerSubscribeCancelRace(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 1*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + // Drive the chain forward continuously while Subscribe and + // cancel race in many goroutines. + stop := make(chan struct{}) + defer close(stop) + + go func() { + ticker := time.NewTicker(2 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + chain.advance(t, 1) + case <-stop: + return + } + } + }() + + const workers = 8 + const iterations = 200 + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < iterations; j++ { + sub, err := tp.Subscribe() + if err != nil { + return + } + + // Drain a few events then cancel mid-flight. + select { + case <-sub.Updates(): + case <-time.After(5 * time.Millisecond): + } + + sub.Cancel() + } + }() + } + + wg.Wait() + + // If we got here without panicking, the broadcast/cancel + // race is fixed. +} + +// TestTipPollerSlowSubscriberDoesNotWedge verifies that one slow +// subscriber that never reads its channel does not block other +// subscribers from receiving events. subscribe.Server's per-client +// queue.ConcurrentQueue is unbounded, so the slow subscriber +// accumulates a backlog without affecting fast subscribers. +func TestTipPollerSlowSubscriberDoesNotWedge(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 5*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + _, _, _, slow, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + defer slow.Cancel() + + _, _, _, fast, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + defer fast.Cancel() + + // Advance the chain. The slow subscriber never drains its + // channel; the fast subscriber must still receive every + // event in order. + const advance = int32(20) + chain.advance(t, advance) + + for expected := int32(101); expected <= 100+advance; expected++ { + select { + case ev := <-fast.Updates(): + require.Equal(t, expected, ev.Height) + + case <-time.After(3 * time.Second): + t.Fatalf("fast subscriber wedged at height %d", + expected) + } + } +} + +// TestTipPollerBestBlockAndSubscribeAtomic verifies that the +// atomic helper either returns an old tip and delivers the next +// event, or returns the new tip and skips that event entirely. +// It never returns the new tip and ALSO delivers the same event as +// a duplicate, and never returns an old tip and FAILS to deliver +// the next event. We assert the invariant by registering a race +// between the helper and continuous tip advances and checking +// that received_event.Height > seed_height for every received +// event (no rewind, no duplicate at seed height). +func TestTipPollerBestBlockAndSubscribeAtomic(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 1*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + // Continuous advance. + stop := make(chan struct{}) + defer close(stop) + + go func() { + ticker := time.NewTicker(2 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + chain.advance(t, 1) + case <-stop: + return + } + } + }() + + for i := 0; i < 50; i++ { + seed, _, _, sub, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + + // Read up to one event with a tight timeout. If we + // get one, assert it's strictly newer than seed. + select { + case ev := <-sub.Updates(): + require.Greater(t, ev.Height, seed, + "received duplicate or stale event") + + case <-time.After(50 * time.Millisecond): + // No event in window — also fine; just unwind. + } + + sub.Cancel() + } +} + +// TestTipPollerCancelStopsDelivery verifies that after Cancel, +// no further events are delivered. Sets up a slow consumer that +// cancels mid-stream and asserts the typed Updates channel +// observes either Quit or close after Cancel returns. +func TestTipPollerCancelStopsDelivery(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 1*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + _, _, _, sub, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + + // Drain a few events. + chain.advance(t, 3) + + for i := 0; i < 3; i++ { + select { + case <-sub.Updates(): + case <-time.After(1 * time.Second): + t.Fatal("timed out before initial drain done") + } + } + + sub.Cancel() + + // After Cancel, the typed Updates channel must close eventually. + select { + case _, ok := <-sub.Updates(): + require.False(t, ok, + "expected closed channel after Cancel") + + case <-time.After(2 * time.Second): + t.Fatal("Updates channel did not close after Cancel") + } +} + +// TestTipPollerStopClosesSubscriptions verifies that stopping the +// poller propagates to every active subscription via the inner +// subscribe.Server's quit signal. +func TestTipPollerStopClosesSubscriptions(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 20*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + + subs := make([]*TipSubscription, 0, 4) + for i := 0; i < 4; i++ { + _, _, _, sub, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + subs = append(subs, sub) + } + + tp.Stop() + + // Every sub's Updates channel must close after Stop. + for i, sub := range subs { + select { + case _, ok := <-sub.Updates(): + require.False(t, ok, + "sub %d Updates not closed after Stop", + i) + + case <-time.After(2 * time.Second): + t.Fatalf("sub %d did not close on Stop", i) + } + } +} + +// TestChainBackendWithPollerLifecycle verifies that when a +// ChainBackend is constructed with NewChainBackendWithPoller (i.e. +// ownsTipPoller=false), Stop on the backend does NOT stop the +// poller — the wallet that owns the poller must be the one to +// stop it. +func TestChainBackendWithPollerLifecycle(t *testing.T) { + t.Parallel() + + chain := newStubChain(100) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + esp := NewEsploraClient(srv.URL, btclog.Disabled) + tp := NewTipPoller(esp, 20*time.Millisecond, btclog.Disabled) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + be, err := NewChainBackendWithPoller(esp, tp, btclog.Disabled) + require.NoError(t, err) + require.NoError(t, be.Start()) + require.NoError(t, be.Stop()) + + // Poller must still be alive: BestBlock returns non-zero, + // and the chain advance + a fresh subscription must still + // receive events. + height, _, _ := tp.BestBlock() + require.Equal(t, int32(100), height) + + _, _, _, sub, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + defer sub.Cancel() + + chain.advance(t, 1) + + select { + case ev := <-sub.Updates(): + require.Equal(t, int32(101), ev.Height) + + case <-time.After(2 * time.Second): + t.Fatal("poller stopped after backend.Stop — " + + "ownsTipPoller=false invariant violated") + } +} + +// TestChainBackendWithPollerNilRejected verifies the H-9 nil-check +// surfaces at construction time rather than as a panic in Start. +func TestChainBackendWithPollerNilRejected(t *testing.T) { + t.Parallel() + + srv := mockEsploraServer( + t, func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + }, + ) + esp := NewEsploraClient(srv.URL, btclog.Disabled) + + be, err := NewChainBackendWithPoller(esp, nil, btclog.Disabled) + require.Error(t, err) + require.Nil(t, be) +} + +// TestTipPollerEventsCounter sanity-checks subscribe.Server fan-out +// by comparing the count of advances driven into the chain with the +// count of events observed on a fast consumer. Used to catch a +// regression where SendUpdate silently drops events. +func TestTipPollerEventsCounter(t *testing.T) { + t.Parallel() + + chain := newStubChain(0) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), + 2*time.Millisecond, btclog.Disabled, + ) + + require.NoError(t, tp.Start()) + defer tp.Stop() + + _, _, _, sub, err := tp.BestBlockAndSubscribe() + require.NoError(t, err) + defer sub.Cancel() + + const advances = int32(25) + + var observed atomic.Int32 + + done := make(chan struct{}) + go func() { + defer close(done) + for { + select { + case ev, ok := <-sub.Updates(): + if !ok { + return + } + observed.Add(1) + if ev.Height >= advances { + return + } + + case <-time.After(3 * time.Second): + return + } + } + }() + + chain.advance(t, advances) + + <-done + + require.Equal(t, advances, observed.Load(), + "missed events between SendUpdate and consumer") +} diff --git a/lwwallet/wallet.go b/lwwallet/wallet.go index 260f28013..63eef9329 100644 --- a/lwwallet/wallet.go +++ b/lwwallet/wallet.go @@ -34,6 +34,13 @@ type Wallet struct { // Wallet provides shared btcwallet-backed operations. walletcore.Wallet + // tipPoller is the single Esplora tip-detection goroutine + // shared by chainSvc (btcwallet) and chainBackend (actors). + // Centralizing the poll cadence avoids the prior arrangement + // in which each consumer ran an independent ticker against + // the same Esplora endpoint. + tipPoller *TipPoller + // chainSvc implements btcwallet's chain.Interface, feeding // block notifications to btcwallet for wallet sync. chainSvc *EsploraChainService @@ -63,17 +70,27 @@ func New(cfg Config) (*Wallet, error) { esplora := NewEsploraClient(cfg.EsploraURL, walletLog) + // A single TipPoller owns Esplora tip detection for the whole + // wallet. Both chainSvc and chainBackend subscribe to its + // event stream, so each new block yields exactly one + // GetTipHeight + GetBlockHashByHeight + GetBlockHeader round + // trip rather than two parallel sets. + tipPoller := NewTipPoller(esplora, cfg.PollInterval, walletLog) + // The EsploraChainService implements btcwallet's chain.Interface // and feeds block notifications to btcwallet for wallet sync. - chainSvc := NewEsploraChainService( - esplora, cfg.PollInterval, walletLog, - ) + chainSvc := NewEsploraChainService(esplora, tipPoller, walletLog) // The ChainBackend implements chainsource.ChainBackend for the - // actor system (confirmation/spend/block registrations). - chainBackend := NewChainBackend( - esplora, cfg.PollInterval, walletLog, + // actor system (confirmation/spend/block registrations). It + // shares the wallet's TipPoller, so its Start does not spin up + // a second poll goroutine. + chainBackend, err := NewChainBackendWithPoller( + esplora, tipPoller, walletLog, ) + if err != nil { + return nil, fmt.Errorf("create chain backend: %w", err) + } coinType := walletcore.CoinTypeForNet(cfg.ChainParams) blockCache := blockcache.NewBlockCache( @@ -121,6 +138,7 @@ func New(cfg Config) (*Wallet, error) { ChainParams: cfg.ChainParams, WalletLog: cfg.Log, }, + tipPoller: tipPoller, chainSvc: chainSvc, esplora: esplora, chainBackend: chainBackend, @@ -128,32 +146,66 @@ func New(cfg Config) (*Wallet, error) { }, nil } -// Start initializes the wallet by starting btcwallet (which -// internally starts the EsploraChainService and syncs the wallet) -// and the chainsource ChainBackend. +// Start initializes the wallet. The startup order is load-bearing: +// the TipPoller must be running before either chainSvc or +// chainBackend subscribes, since both rely on TipPoller.BestBlock() +// to seed their initial tip without issuing fresh Esplora calls. +// btcwallet's Start in turn drives chainSvc.Start through its +// chain.Interface contract, so chainSvc inherits the live tip the +// poller already established. func (w *Wallet) Start() error { ctx := context.Background() + // Each successful sub-system Start arms a rollback closure; + // on the happy path we clear the slice at the end and the + // deferred unwind is a no-op. On any error return below the + // already-started subsystems are torn down in reverse order + // so a bad passphrase / locked DB / unreachable Esplora does + // not leak a polling goroutine for the lifetime of the + // process. + var rollback []func() + defer func() { + for i := len(rollback) - 1; i >= 0; i-- { + rollback[i]() + } + }() + + // Spin up the shared tip poller before any consumer subscribes. + if err := w.tipPoller.Start(); err != nil { + return fmt.Errorf("start tip poller: %w", err) + } + rollback = append(rollback, w.tipPoller.Stop) + // btcWallet.Start() unlocks the wallet, creates key scopes, - // starts the chain service, and begins wallet synchronization. + // starts the chain service (which subscribes to the tip + // poller), and begins wallet synchronization. if err := w.BtcWallet.Start(); err != nil { return fmt.Errorf("start btcwallet: %w", err) } + rollback = append(rollback, func() { _ = w.BtcWallet.Stop() }) // Start the chainsource ChainBackend used by the actor system. - // This is separate from the chain service used by btcwallet. + // It also subscribes to the wallet's TipPoller; it does not + // own that poller, so calling Start here only spawns the + // event-handler goroutine. if err := w.chainBackend.Start(); err != nil { return fmt.Errorf("start chain backend: %w", err) } w.Logger(ctx).InfoS(ctx, "Lightweight wallet started") + // All subsystems started cleanly — clear the rollback slice + // so the deferred unwind is a no-op. + rollback = nil + return nil } -// Stop shuts down the wallet, chain service, and chain backend. We -// wait for the chain service goroutine to fully exit before -// returning to avoid racing with btcwallet's internal writes. +// Stop shuts down the wallet, chain service, chain backend, and +// shared tip poller. The teardown order mirrors Start in reverse: +// btcwallet first (so it stops draining notifications), then the +// chain backend (which unsubscribes from the poller), and finally +// the tip poller itself once nobody else can be observing it. func (w *Wallet) Stop() { ctx := context.Background() @@ -163,8 +215,19 @@ func (w *Wallet) Stop() { if err := w.BtcWallet.InternalWallet().Database().Close(); err != nil { w.Logger(ctx).WarnS(ctx, "Failed to close btcwallet DB", err) } + + // Explicitly Stop the chain service before waiting on its + // goroutine. btcwallet.Stop will transitively call + // chainClient.Stop today, but relying on that is brittle — + // any future fast-shutdown path that bypasses + // btcwallet.Stop would leave handleTipEvents blocked on its + // quit channel and deadlock WaitForShutdown. The Stop is + // idempotent (sync.Once), so the duplicate-close path is + // safe even if btcwallet's Stop already fired it. + w.chainSvc.Stop() w.chainSvc.WaitForShutdown() _ = w.chainBackend.Stop() + w.tipPoller.Stop() w.Logger(ctx).InfoS(ctx, "Lightweight wallet stopped") } @@ -204,11 +267,16 @@ func (w *Wallet) FinalizePsbtDirect(packet *psbt.Packet) error { // ListUnspentWitness can return stale results when called right // after a confirmation event because the two pipelines poll // Esplora independently. +// +// The target tip is read from the shared TipPoller's cached +// snapshot rather than via a fresh GetTipHeight HTTP call: the +// poller is the only Esplora tip-detection authority in the wallet, +// and reading its cache lets WaitForSync run at the wallet's +// internal poll cadence rather than firing one extra HTTP request +// per call (which on a hot ListUnspentWitness path could compound +// against an already rate-limited Esplora endpoint). func (w *Wallet) WaitForSync(ctx context.Context) error { - tipHeight, err := w.esplora.GetTipHeight() - if err != nil { - return fmt.Errorf("esplora tip height: %w", err) - } + tipHeight, _, _ := w.tipPoller.BestBlock() for { syncedTo := w.BtcWallet.InternalWallet().SyncedTo() diff --git a/wallet/wallet.go b/wallet/wallet.go index 73e46d33f..23af25f0f 100644 --- a/wallet/wallet.go +++ b/wallet/wallet.go @@ -43,13 +43,18 @@ const ( // epoch notification before the wallet's UTXO set is fully updated. // For neutrino backends, btcwallet's internal block processing // (fetching the full block from P2P, running AddCredit) can take - // over a second after the epoch arrives. - listUnspentMaxRetries = 10 + // over a second after the epoch arrives. The retry budget is kept + // small so idle epochs (no boarding UTXO racing) don't hammer the + // backend; HTTP-backed backends like lwwallet's Esplora adapter + // previously triggered mempool.space rate-limit (HTTP 429) responses + // under the old 10×200ms burst. + listUnspentMaxRetries = 3 // listUnspentRetryDelay is the delay between ListUnspent retries. - // We keep this small so confirmed boarding UTXOs are detected - // promptly without waiting for another block. - listUnspentRetryDelay = 200 * time.Millisecond + // Chosen to give a slow backend (e.g. neutrino fetching a block over + // P2P, or a remote Esplora roundtripping a tip update) a full second + // to catch up between attempts. + listUnspentRetryDelay = 1 * time.Second ) // notifierInfo holds the configuration for a registered confirmation notifier.