From c96448a308d5d775c91ad167e6250684ed788b50 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 30 Apr 2026 17:00:30 -0400 Subject: [PATCH 1/4] wallet: throttle ListUnspent retry burst on block epoch handleBlockEpoch retried ListUnspent up to 10x at 200ms intervals whenever no new boarding UTXO surfaced under a fresh block, burning ~3.5s of HTTP traffic per epoch even when nothing was racing the notification. Against a public Esplora endpoint this was enough to trip mempool.space's HTTP 429 rate limit. Cap the retry budget at 3 attempts and widen the inter-attempt delay to 1s. This still gives a slow neutrino-fed btcwallet a generous window to credit-mark a freshly-confirmed boarding UTXO, while shrinking the per-epoch worst-case from ~30 to ~6 backend round trips on the lwwallet path. --- wallet/wallet.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) 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. From 1959bc484d17818ba4f68346a8d7ffa205cc7918 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 30 Apr 2026 17:00:38 -0400 Subject: [PATCH 2/4] darepod: passive default Esplora poll interval The 5s default was tuned for regtest where blocks land on demand, but it bleeds onto mainnet users where the average inter-block gap is ~600s. At 5s the chain backend issued ~12 GetTipHeight calls per minute against the configured Esplora endpoint -- enough to contribute to mempool.space rate-limit responses when paired with the boarding UTXO retry burst. Bump the default to 30s. Block-detection latency stays an order of magnitude under the average mainnet block interval, while the steady-state request rate drops by 6x. Regtest-driven harnesses that mine blocks on demand should override this knob via the existing wallet.pollinterval config flag. --- darepod/config.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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. From ac97e284518f0164e8d87311da776f8914e028cc Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 30 Apr 2026 17:00:49 -0400 Subject: [PATCH 3/4] lwwallet: LRU-cache immutable Esplora responses Two pollers (the btcwallet chain.Interface adapter and the chainsource.ChainBackend) drive the Esplora client independently, each fetching block hashes, headers, and raw blocks on every new tip they detect. Their per-call work is identical, so under the old wiring every confirmed block was fetched twice, every header was decoded twice, and every TxProof construction triggered a fresh GetRawTx round-trip even when the same boarding txid had been served seconds earlier. Add four neutrino LRU caches keyed by chainhash.Hash for the hash-addressed methods: GetBlockHeader, GetRawBlockHeader, GetRawBlock, and GetRawTx. Capacities are byte-sized via each value's serialized footprint -- 50 MiB for transactions, 200 MiB for raw blocks, and 1 MiB each for the small fixed-size header types. Reorgs cannot stale these entries: each is content- addressed by a hash that uniquely identifies its body, so a stale hit is only possible if the backend lied about the hash in the first place. Mutable endpoints (GetTipHeight, GetAddressUtxos, GetTxStatus, GetFeeEstimates, GetOutspend) deliberately stay uncached; they report live state and a stale answer would silently mask a reorg or a pending spend. --- go.mod | 2 +- lwwallet/chain_backend_test.go | 26 ++- lwwallet/esplora.go | 381 +++++++++++++++++++++++++++++---- lwwallet/esplora_cache.go | 138 ++++++++++++ lwwallet/esplora_cache_test.go | 196 +++++++++++++++++ 5 files changed, 698 insertions(+), 45 deletions(-) create mode 100644 lwwallet/esplora_cache.go create mode 100644 lwwallet/esplora_cache_test.go diff --git a/go.mod b/go.mod index 49ce5c7b3..ddf8d9ab1 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 @@ -124,7 +125,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 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") +} From 5dba8352d747821c0b0a37da3aaeb4dd22ff268b Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 30 Apr 2026 17:09:17 -0400 Subject: [PATCH 4/4] lwwallet: unify tip polling under shared TipPoller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent goroutines used to drive Esplora tip detection: the EsploraChainService (feeding btcwallet's chain.Interface) and the chainsource.ChainBackend (feeding the actor system) each ran their own ticker against the same client. At the prior 5 s default that produced ~24 GetTipHeight calls per minute against a single endpoint, plus duplicated GetBlockHashByHeight + GetBlockHeader work for every block both saw. Introduce TipPoller as the single Esplora tip-detection authority for the lwwallet. It owns one goroutine, walks oldHeight+1 → newHeight on each detected advance, and fans the canonical TipBlock event (height + hash + JSON header) out to subscribers. EsploraChainService and ChainBackend both become passive consumers: each subscribes at Start time and reacts to the events without issuing its own tip queries. The wallet drives lifecycle (Start the poller before either subscriber, Stop it after both have torn down). For the standalone darepod path that constructs a ChainBackend without a wallet, NewChainBackend keeps its existing signature and creates an internal TipPoller it owns end-to-end; NewChainBackendWithPoller is the new shared-poller entry point used by the wallet. Confirmation and spend registrations now schedule a one-shot check at registration time so a tx that is already buried beyond numConfs (or an outpoint that is already spent) fires synchronously, rather than waiting for the next tip event. The prior polling design caught this implicitly via the unconditional checkConfirmations / checkSpends call on every tick. --- darepod/server.go | 38 ++- go.mod | 2 +- lwwallet/chain_backend.go | 517 +++++++++++++++++++++-------- lwwallet/esplora_chain.go | 308 ++++++++---------- lwwallet/subscribe.go | 175 ++++++++++ lwwallet/tip_poller.go | 360 +++++++++++++++++++++ lwwallet/tip_poller_test.go | 626 ++++++++++++++++++++++++++++++++++++ lwwallet/wallet.go | 104 ++++-- 8 files changed, 1786 insertions(+), 344 deletions(-) create mode 100644 lwwallet/subscribe.go create mode 100644 lwwallet/tip_poller.go create mode 100644 lwwallet/tip_poller_test.go 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 ddf8d9ab1..8fd4689a0 100644 --- a/go.mod +++ b/go.mod @@ -40,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 @@ -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/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()