diff --git a/chainsource/finality.go b/chainsource/finality.go index 2b8561b1f..5bdc97ac6 100644 --- a/chainsource/finality.go +++ b/chainsource/finality.go @@ -21,6 +21,16 @@ var finalityBlockSubscriptionBackoffs = []time.Duration{ 2 * time.Second, } +// finalityBlockSubscriptionAttemptTimeout bounds each individual +// RegisterBlocks attempt. Without it a single hung RegisterBlocks call +// (e.g. a wedged lndclient gRPC stream) would block the conf/spend +// monitoring goroutine indefinitely — stalling Confirmed/Reorged/Done +// delivery on that watch — since the retry schedule only bounds the gaps +// between attempts, not the attempts themselves. 10s mirrors the per-call +// registration timeout used in conf_actor.go's handleRegisterConf so the +// whole file behaves consistently under a slow backend. +const finalityBlockSubscriptionAttemptTimeout = 10 * time.Second + // registerBlocksForFinality registers a block-epoch subscription used // to synthesize a Done signal at FinalityDepth past an observed // confirmation or spend. The call is retried with a short bounded @@ -29,22 +39,11 @@ var finalityBlockSubscriptionBackoffs = []time.Duration{ // lndclient over gRPC); a one-shot RegisterBlocks attempt that // briefly hiccups would leak the per-watch sub-actor indefinitely. // -// The retries run in a dedicated arming goroutine (not the sub-actor's -// select loop), so brief blocking here is safe: more confirmation/spend -// events on this specific watch are not expected during the retry window -// (we already consumed the one that triggered the arm), and ctx -// cancellation breaks out promptly. -// -// The passed ctx MUST be the sub-actor's long-lived context, and it is -// handed to RegisterBlocks unwrapped: for in-process backends the -// block-epoch forwarder goroutine is tied to the ctx it receives, so -// bounding each attempt with a cancellable child ctx (and cancelling it -// once the call returns) would tear the subscription down the instant it -// was armed — starving finality synthesis of the very epochs it needs. -// A hung RegisterBlocks can therefore stall this arming goroutine, but -// that is contained: it is off the select loop (fix moved arming there -// precisely so a slow backend cannot wedge Confirmed/Reorged/Done -// delivery), and a genuinely wedged backend is a lost watch regardless. +// The retries run in the calling sub-actor's monitoring goroutine, so +// brief blocking here is safe: more confirmation/spend events on this +// specific watch are not expected during the retry window (we already +// consumed the one that triggered the arm), and ctx cancellation +// breaks out promptly. // // Returns the registration on success, or a non-nil error after // retries are exhausted. Callers should log the error at warn level @@ -55,7 +54,14 @@ func registerBlocksForFinality(ctx context.Context, backend ChainBackend, var lastErr error for attempt, backoff := range finalityBlockSubscriptionBackoffs { - reg, err := backend.RegisterBlocks(ctx) + // Bound each attempt so a hung RegisterBlocks cannot wedge the + // monitoring goroutine; the retry schedule only bounds the gaps + // between attempts, not a single stuck call. + attemptCtx, cancel := context.WithTimeout( + ctx, finalityBlockSubscriptionAttemptTimeout, + ) + reg, err := backend.RegisterBlocks(attemptCtx) + cancel() if err == nil { return reg, nil } diff --git a/lwwallet/AGENTS.md b/lwwallet/AGENTS.md index bfeb325b1..f077ae447 100644 --- a/lwwallet/AGENTS.md +++ b/lwwallet/AGENTS.md @@ -12,28 +12,61 @@ without an external LND node. Implements `wallet.BoardingBackend`, - `TipPoller` — Single source of truth for the chain tip. One goroutine polls Esplora at a configurable interval; when the tip advances it walks each new height, fetches hash + header, and fans out `TipBlock` events to all - subscribers via the embedded `EventServer`. Multiple downstream chain - watchers share one poller cadence instead of polling independently. - Constructor: `NewTipPoller(esplora, pollInterval, logger)`. Key methods: - `Start()`, `Stop()`, `BestBlock()`, `Subscribe()`, - `BestBlockAndSubscribe()` (atomic tip-read + subscribe to avoid missed - events). + subscribers via the embedded `EventServer`. It also detects chain + reorganizations — both same-height hash drift (a block at height N replaced + by a different block at the same height) and deeper reorgs (the first new + height's `PrevBlock` header field does not point at the cached tip hash) — + and fans `ReorgEvent` updates out on a sibling `EventServer`. A unified + `ChainEvent` stream (reachable via `SubscribeChain` / + `BestBlockAndSubscribeChain`) delivers both event types in producer + order on a single channel for consumers that require strict + reorg-before-replacement-tip ordering. A bounded height→hash ring + buffer (`historySize` entries, default `DefaultHashHistorySize=100`) + is seeded back to `tip - historySize + 1` at `Start` so reorg + walk-back can resolve old-chain hashes from the cache rather than + terminating early on the first uncached height. Constructors: + `NewTipPoller(esplora, pollInterval, logger)` (default history) and + `NewTipPollerWithConfig(esplora, pollInterval, historySize, logger)`. + Key methods: `Start()`, `Stop()`, `BestBlock()`, `Subscribe()`, + `SubscribeReorgs()`, `SubscribeChain()`, `BestBlockAndSubscribe()`, + `BestBlockAndSubscribeAll()`, and `BestBlockAndSubscribeChain()`. - `TipBlock` — Event emitted per new block: `Height`, `Hash`, and the `*esploraBlock` header (pre-fetched so subscribers avoid a second Esplora round-trip). - `TipSubscription` — Typed alias `Subscription[*TipBlock]` returned by `TipPoller.Subscribe`. Cancel via `Cancel()`. +- `ReorgEvent` — Event emitted when the poller observes that one or more + previously broadcast blocks are no longer on the canonical chain. Carries + `ForkHeight`, `Disconnected []chainhash.Hash` (ascending), and + `Connected []*TipBlock` (ascending). The poller delivers `ReorgEvent` + BEFORE fanning the connected blocks out on the standard `TipBlock` stream + so consumers can mark registrations dirty before the re-confirmation + arrives. +- `ReorgSubscription` — Typed alias `Subscription[*ReorgEvent]` returned by + `TipPoller.SubscribeReorgs`. - `EventServer[T]` — Generic wrapper around LND's `subscribe.Server` that delivers typed events. `Subscribe()` returns a `Subscription[T]` that converts untyped `interface{}` updates to `T` on a per-subscriber goroutine. - `Subscription[T]` — Typed subscription handle with `Updates() <-chan T`, `Quit() <-chan struct{}`, and idempotent `Cancel()`. -- `ChainBackend` — Implements `chainsource.ChainBackend` by subscribing to a - shared `TipPoller`. On each `TipBlock` event it dispatches block epoch - notifications and re-checks pending confirmation/spend registrations. - Constructor: `NewChainBackend(esplora, pollInterval, logger)` (owns its own - TipPoller) or `NewChainBackendWithPoller(esplora, tipPoller, logger)` (shares - an externally managed poller). +- `ChainBackend` — Implements `chainsource.ChainBackend` by subscribing to + the shared `TipPoller`'s unified `ChainEvent` stream. A single + handler goroutine processes reorg and tip events in producer order: + on each `ReorgEvent` it walks active conf/spend registrations whose + last delivered positive event names a disconnected block hash, fires + their `Reorged` channel, resets state to `stateWatching`, and + re-checks status against the new chain; on each `TipBlock` it + dispatches block-epoch notifications and runs the broad re-check. + Single-goroutine dispatch guarantees a `ReorgEvent` is fully + processed before any post-reorg block-epoch can drive chainsource + finality synthesis on the replacement chain. The `Done` channel on + each returned registration is allocated but never written by the + backend — the chainsource `ConfActor` / `SpendActor` synthesizes + `Done` at its configured `FinalityDepth` using block epochs the + backend already delivers. Constructor: + `NewChainBackend(esplora, pollInterval, logger)` (owns its own + TipPoller) or `NewChainBackendWithPoller(esplora, tipPoller, logger)` + (shares an externally managed poller). - `EsploraClient` — HTTP REST client for the Esplora/mempool.space API. Constructor: `NewEsploraClient(baseURL, logger)`. Hash-addressed responses (transactions, blocks, headers) are cached in LRU caches bounded by @@ -42,7 +75,16 @@ without an external LND node. Implements `wallet.BoardingBackend`, response is verified to hash to the requested key before insertion. - `EsploraChainService` — `chain.Interface` adapter over `EsploraClient`, driven by a shared `TipPoller`. Feeds btcwallet's internal address-credit - pipeline. Constructor: `NewEsploraChainService(esplora, tipPoller, logger)`. + pipeline. Subscribes to the unified `ChainEvent` stream via + `BestBlockAndSubscribeChain`; on each `ReorgEvent` it emits + `chain.BlockDisconnected` notifications for the disconnected hashes + (newest height first), then processes the replacement chain's tip + events as `chain.BlockConnected`. The unified stream guarantees + `BlockDisconnected` lands on btcwallet's notification queue before + any `BlockConnected` for the new chain — without this, btcwallet's + `disconnectBlock` path would refuse the rollback because the cached + hash at the affected height would already be overwritten. + Constructor: `NewEsploraChainService(esplora, tipPoller, logger)`. - `BoardingBackendAdapter` — Implements `wallet.BoardingBackend` and `wallet.OutputLeaser`. Queries Esplora directly for UTXOs (bypasses btcwallet's UTXO tracking because btcwallet skips credit marking for @@ -65,11 +107,55 @@ without an external LND node. Implements `wallet.BoardingBackend`, - Exactly one `TipPoller` goroutine drives both `EsploraChainService` and `ChainBackend`; neither polls Esplora independently. -- `BestBlockAndSubscribe` holds `TipPoller.mu` across `{Subscribe + tip-read}` - while the poll loop holds it across `{update tip + SendUpdate}`, ensuring no - tip event is missed or duplicated on subscribe. -- Same-height reorgs are invisible until the chain advances to the next height - (known limitation; acceptable for confirmation-target use cases). +- `BestBlockAndSubscribe` / `BestBlockAndSubscribeAll` holds `TipPoller.mu` + across `{Subscribe + tip-read}` while the poll loop holds it across + `{update tip + SendUpdate}`, ensuring no tip event is missed or duplicated + on subscribe. +- Same-height reorgs ARE detected: each poll cycle compares the live hash at + the cached tip height against the cached hash and emits a `ReorgEvent` on + mismatch. The previous limitation ("same-height reorgs invisible until the + chain advances") is closed. +- Conf/spend registrations are multi-shot reorg-aware: they are not deleted + after the first positive event, the returned `Reorged` channel fires when + a previously delivered confirmation/spend is reorged out, and a fresh + `Confirmed`/`Spend` may fire on the new canonical chain. `Done` is + synthesized at the chainsource actor layer from block epochs, not by the + backend. +- Reorg detection for registrations whose last delivered positive event + references a block older than the poller's seeded hash history falls + back to a canonical re-query of the live chain state. The fast path + (cached block hash in `ReorgEvent.Disconnected`) covers in-window + reorgs; the canonical re-query covers deeper reorgs where the cached + hash was pruned from `recentHashes` or the registration delivered + against a block below the seeded window on a fresh poller. Both paths + fire `Reorged` followed by a fresh `Confirmed`/`Spend` if the chain + still carries the watched event under a different anchor. +- The tip poller aborts the current poll cycle on raw-block-header fetch + failure rather than optimistically advancing. The raw 80-byte header + is the only carrier of `PrevBlock`, which is the continuity check that + catches a reorg crossing the boundary at the old tip; falling through + on a transient fetch flake would permanently hide that reorg by + caching the new tip hash without ever emitting a `ReorgEvent`. The + next poll tick retries. +- The poller exposes a unified `ChainEvent` stream (`SubscribeChain` / + `BestBlockAndSubscribeChain`) that delivers reorg and tip updates on + a single producer-ordered channel. `ChainBackend` and + `EsploraChainService` both subscribe to this stream rather than the + separate `events` / `reorgs` streams so a `ReorgEvent` is observed + before the subsequent `TipBlock` events for the replacement chain. + Strict ordering is required: btcwallet's `disconnectBlock` rejects + rollback if the cached hash at that height has been overwritten by a + stale `BlockConnected`, and chainsource finality synthesis depends on + registrations being reset before block-epoch driven re-checks run on + the replacement chain. +- The poller seeds `recentHashes` at `Start` by walking back + `historySize - 1` heights from the initial tip. Without the seed a + reorg whose disconnected range extends below the seeded tip but + within `historySize` would terminate walk-back at the first + uncached height, producing a `ReorgEvent` whose `Disconnected` list + carries only the tip-boundary hash and starving the chain.Interface + adapter of the per-height `BlockDisconnected` events btcwallet needs + to roll back its sync state. - LRU caches only hold immutable, hash-addressed data; a verified hash prevents a compromised Esplora endpoint from injecting arbitrary cache entries. - UTXO enumeration queries Esplora directly rather than btcwallet's internal diff --git a/lwwallet/CLAUDE.md b/lwwallet/CLAUDE.md index bfeb325b1..396cef7db 100644 --- a/lwwallet/CLAUDE.md +++ b/lwwallet/CLAUDE.md @@ -12,28 +12,61 @@ without an external LND node. Implements `wallet.BoardingBackend`, - `TipPoller` — Single source of truth for the chain tip. One goroutine polls Esplora at a configurable interval; when the tip advances it walks each new height, fetches hash + header, and fans out `TipBlock` events to all - subscribers via the embedded `EventServer`. Multiple downstream chain - watchers share one poller cadence instead of polling independently. - Constructor: `NewTipPoller(esplora, pollInterval, logger)`. Key methods: - `Start()`, `Stop()`, `BestBlock()`, `Subscribe()`, - `BestBlockAndSubscribe()` (atomic tip-read + subscribe to avoid missed - events). + subscribers via the embedded `EventServer`. It also detects chain + reorganizations — both same-height hash drift (a block at height N replaced + by a different block at the same height) and deeper reorgs (the first new + height's `PrevBlock` header field does not point at the cached tip hash) — + and fans `ReorgEvent` updates out on a sibling `EventServer`. A unified + `ChainEvent` stream (reachable via `SubscribeChain` / + `BestBlockAndSubscribeChain`) delivers both event types in producer + order on a single channel for consumers that require strict + reorg-before-replacement-tip ordering. A bounded height→hash history + map (`historySize` entries, default `DefaultHashHistorySize=100`) + is seeded back to `tip - historySize + 1` at `Start` so reorg + walk-back can resolve old-chain hashes from the cache rather than + terminating early on the first uncached height. Constructors: + `NewTipPoller(esplora, pollInterval, logger)` (default history) and + `NewTipPollerWithConfig(esplora, pollInterval, historySize, logger)`. + Key methods: `Start()`, `Stop()`, `BestBlock()`, `Subscribe()`, + `SubscribeReorgs()`, `SubscribeChain()`, `BestBlockAndSubscribe()`, + `BestBlockAndSubscribeAll()`, and `BestBlockAndSubscribeChain()`. - `TipBlock` — Event emitted per new block: `Height`, `Hash`, and the `*esploraBlock` header (pre-fetched so subscribers avoid a second Esplora round-trip). - `TipSubscription` — Typed alias `Subscription[*TipBlock]` returned by `TipPoller.Subscribe`. Cancel via `Cancel()`. +- `ReorgEvent` — Event emitted when the poller observes that one or more + previously broadcast blocks are no longer on the canonical chain. Carries + `ForkHeight`, `Disconnected []chainhash.Hash` (ascending), and + `Connected []*TipBlock` (ascending). The poller delivers `ReorgEvent` + BEFORE fanning the connected blocks out on the standard `TipBlock` stream + so consumers can mark registrations dirty before the re-confirmation + arrives. +- `ReorgSubscription` — Typed alias `Subscription[*ReorgEvent]` returned by + `TipPoller.SubscribeReorgs`. - `EventServer[T]` — Generic wrapper around LND's `subscribe.Server` that delivers typed events. `Subscribe()` returns a `Subscription[T]` that converts untyped `interface{}` updates to `T` on a per-subscriber goroutine. - `Subscription[T]` — Typed subscription handle with `Updates() <-chan T`, `Quit() <-chan struct{}`, and idempotent `Cancel()`. -- `ChainBackend` — Implements `chainsource.ChainBackend` by subscribing to a - shared `TipPoller`. On each `TipBlock` event it dispatches block epoch - notifications and re-checks pending confirmation/spend registrations. - Constructor: `NewChainBackend(esplora, pollInterval, logger)` (owns its own - TipPoller) or `NewChainBackendWithPoller(esplora, tipPoller, logger)` (shares - an externally managed poller). +- `ChainBackend` — Implements `chainsource.ChainBackend` by subscribing to + the shared `TipPoller`'s unified `ChainEvent` stream. A single + handler goroutine processes reorg and tip events in producer order: + on each `ReorgEvent` it walks active conf/spend registrations whose + last delivered positive event names a disconnected block hash, fires + their `Reorged` channel, resets state to `stateWatching`, and + re-checks status against the new chain; on each `TipBlock` it + dispatches block-epoch notifications and runs the broad re-check. + Single-goroutine dispatch guarantees a `ReorgEvent` is fully + processed before any post-reorg block-epoch can drive chainsource + finality synthesis on the replacement chain. The `Done` channel on + each returned registration is allocated but never written by the + backend — the chainsource `ConfActor` / `SpendActor` synthesizes + `Done` at its configured `FinalityDepth` using block epochs the + backend already delivers. Constructor: + `NewChainBackend(esplora, pollInterval, logger)` (owns its own + TipPoller) or `NewChainBackendWithPoller(esplora, tipPoller, logger)` + (shares an externally managed poller). - `EsploraClient` — HTTP REST client for the Esplora/mempool.space API. Constructor: `NewEsploraClient(baseURL, logger)`. Hash-addressed responses (transactions, blocks, headers) are cached in LRU caches bounded by @@ -42,7 +75,16 @@ without an external LND node. Implements `wallet.BoardingBackend`, response is verified to hash to the requested key before insertion. - `EsploraChainService` — `chain.Interface` adapter over `EsploraClient`, driven by a shared `TipPoller`. Feeds btcwallet's internal address-credit - pipeline. Constructor: `NewEsploraChainService(esplora, tipPoller, logger)`. + pipeline. Subscribes to the unified `ChainEvent` stream via + `BestBlockAndSubscribeChain`; on each `ReorgEvent` it emits + `chain.BlockDisconnected` notifications for the disconnected hashes + (newest height first), then processes the replacement chain's tip + events as `chain.BlockConnected`. The unified stream guarantees + `BlockDisconnected` lands on btcwallet's notification queue before + any `BlockConnected` for the new chain — without this, btcwallet's + `disconnectBlock` path would refuse the rollback because the cached + hash at the affected height would already be overwritten. + Constructor: `NewEsploraChainService(esplora, tipPoller, logger)`. - `BoardingBackendAdapter` — Implements `wallet.BoardingBackend` and `wallet.OutputLeaser`. Queries Esplora directly for UTXOs (bypasses btcwallet's UTXO tracking because btcwallet skips credit marking for @@ -65,11 +107,55 @@ without an external LND node. Implements `wallet.BoardingBackend`, - Exactly one `TipPoller` goroutine drives both `EsploraChainService` and `ChainBackend`; neither polls Esplora independently. -- `BestBlockAndSubscribe` holds `TipPoller.mu` across `{Subscribe + tip-read}` - while the poll loop holds it across `{update tip + SendUpdate}`, ensuring no - tip event is missed or duplicated on subscribe. -- Same-height reorgs are invisible until the chain advances to the next height - (known limitation; acceptable for confirmation-target use cases). +- `BestBlockAndSubscribe` / `BestBlockAndSubscribeAll` holds `TipPoller.mu` + across `{Subscribe + tip-read}` while the poll loop holds it across + `{update tip + SendUpdate}`, ensuring no tip event is missed or duplicated + on subscribe. +- Same-height reorgs ARE detected: each poll cycle compares the live hash at + the cached tip height against the cached hash and emits a `ReorgEvent` on + mismatch. The previous limitation ("same-height reorgs invisible until the + chain advances") is closed. +- Conf/spend registrations are multi-shot reorg-aware: they are not deleted + after the first positive event, the returned `Reorged` channel fires when + a previously delivered confirmation/spend is reorged out, and a fresh + `Confirmed`/`Spend` may fire on the new canonical chain. `Done` is + synthesized at the chainsource actor layer from block epochs, not by the + backend. +- Reorg detection for registrations whose last delivered positive event + references a block older than the poller's seeded hash history falls + back to a canonical re-query of the live chain state. The fast path + (cached block hash in `ReorgEvent.Disconnected`) covers in-window + reorgs; the canonical re-query covers deeper reorgs where the cached + hash was pruned from `recentHashes` or the registration delivered + against a block below the seeded window on a fresh poller. Both paths + fire `Reorged` followed by a fresh `Confirmed`/`Spend` if the chain + still carries the watched event under a different anchor. +- The tip poller aborts the current poll cycle on raw-block-header fetch + failure rather than optimistically advancing. The raw 80-byte header + is the only carrier of `PrevBlock`, which is the continuity check that + catches a reorg crossing the boundary at the old tip; falling through + on a transient fetch flake would permanently hide that reorg by + caching the new tip hash without ever emitting a `ReorgEvent`. The + next poll tick retries. +- The poller exposes a unified `ChainEvent` stream (`SubscribeChain` / + `BestBlockAndSubscribeChain`) that delivers reorg and tip updates on + a single producer-ordered channel. `ChainBackend` and + `EsploraChainService` both subscribe to this stream rather than the + separate `events` / `reorgs` streams so a `ReorgEvent` is observed + before the subsequent `TipBlock` events for the replacement chain. + Strict ordering is required: btcwallet's `disconnectBlock` rejects + rollback if the cached hash at that height has been overwritten by a + stale `BlockConnected`, and chainsource finality synthesis depends on + registrations being reset before block-epoch driven re-checks run on + the replacement chain. +- The poller seeds `recentHashes` at `Start` by walking back + `historySize - 1` heights from the initial tip. Without the seed a + reorg whose disconnected range extends below the seeded tip but + within `historySize` would terminate walk-back at the first + uncached height, producing a `ReorgEvent` whose `Disconnected` list + carries only the tip-boundary hash and starving the chain.Interface + adapter of the per-height `BlockDisconnected` events btcwallet needs + to roll back its sync state. - LRU caches only hold immutable, hash-addressed data; a verified hash prevents a compromised Esplora endpoint from injecting arbitrary cache entries. - UTXO enumeration queries Esplora directly rather than btcwallet's internal diff --git a/lwwallet/chain_backend.go b/lwwallet/chain_backend.go index 573bec48d..442700ac4 100644 --- a/lwwallet/chain_backend.go +++ b/lwwallet/chain_backend.go @@ -16,6 +16,7 @@ import ( "github.com/btcsuite/btcd/wire/v2" "github.com/btcsuite/btclog/v2" "github.com/lightninglabs/darepo-client/chainsource" + fn "github.com/lightningnetwork/lnd/fn/v2" "golang.org/x/sync/singleflight" ) @@ -42,6 +43,29 @@ const ( // raise Esplora load over the default poll cadence. const recheckHeartbeatInterval = 60 * time.Second +// regState is the lifecycle state of a conf/spend registration. +// +// stateWatching: no positive event has been delivered yet (or the +// last positive event was reorged out). The next checkSingle... that +// finds the tx confirmed / outpoint spent will emit Confirmed / +// Spend and transition to statePositive. +// +// statePositive: a Confirmed / Spend event has been delivered and the +// associated block-hash is cached on the registration. The next +// checkSingle... that does NOT find the tx confirmed / outpoint +// spent in the same block leaves the registration alone (the +// chainsource actor handles Done synthesis at FinalityDepth via +// block epochs). A reorg event that names the cached block-hash in +// its Disconnected set fires Reorged, resets cached state, and +// transitions back to stateWatching so the next re-check can fire +// Confirmed / Spend again on the new chain. +type regState uint8 + +const ( + stateWatching regState = iota + statePositive +) + // confRegistration tracks a pending confirmation registration within the // polling loop. type confRegistration struct { @@ -63,8 +87,37 @@ type confRegistration struct { // confChan is the channel to send the confirmation on. confChan chan *chainsource.TxConfirmation + // reorgChan is the channel that fires when a previously delivered + // confirmation is reorged out of the canonical chain. Buffered to + // 1; never written by the backend before stateWatching has been + // re-entered. The value is the chainsource ordering sequence; this + // backend does not stamp sequences (it emits conf/reorg from a + // single ordered handler goroutine), so it always sends 0, which + // the chainsource actor treats as always-apply. + reorgChan chan uint64 + + // doneChan is allocated for API symmetry with the chainsource + // contract. The Esplora-backed backend does not write to it: the + // chainsource ConfActor synthesizes Done at FinalityDepth from + // block epochs once the backend stops re-firing events. + doneChan chan struct{} + // cancelCh signals that this registration has been cancelled. cancelCh chan struct{} + + // regMu guards the fields below that mutate over the + // registration's lifetime. It is held only briefly during state + // transitions; channel sends are performed without it. + regMu sync.Mutex + + // state is the current lifecycle state. + state regState + + // lastBlockHash is the BlockHash of the last delivered + // confirmation when state == statePositive. Used to detect when + // a reorg's Disconnected hash list invalidates this + // registration's last event. + lastBlockHash chainhash.Hash } // spendRegistration tracks a pending spend registration within the @@ -82,8 +135,34 @@ type spendRegistration struct { // spendChan is the channel to send the spend detail on. spendChan chan *chainsource.SpendDetail + // reorgChan is the channel that fires when a previously delivered + // spend is reorged out of the canonical chain. Buffered to 1. + reorgChan chan uint64 + + // doneChan is allocated for API symmetry with the chainsource + // contract. The Esplora-backed backend does not write to it: the + // chainsource SpendActor synthesizes Done at FinalityDepth. + doneChan chan struct{} + // cancelCh signals that this registration has been cancelled. cancelCh chan struct{} + + // regMu guards the fields below. + regMu sync.Mutex + + // state is the current lifecycle state. + state regState + + // lastSpenderHash is the SpenderTxHash of the last delivered + // spend when state == statePositive. + lastSpenderHash chainhash.Hash + + // lastSpendingBlockHash is the block hash that the last + // delivered spend confirmed in, parsed from the same + // outspend response that confirmed the spend. Reorg + // comparison against ReorgEvent.Disconnected uses this so + // the spend-watch path does not need an extra HTTP round-trip. + lastSpendingBlockHash chainhash.Hash } // blockRegistration tracks a block epoch subscription. @@ -205,7 +284,8 @@ func (b *ChainBackend) Start() error { } } - height, hash, _, sub, err := b.tipPoller.BestBlockAndSubscribe() + height, hash, _, chainSub, err := + b.tipPoller.BestBlockAndSubscribeChain() if err != nil { return fmt.Errorf("subscribe to tip poller: %w", err) } @@ -216,7 +296,7 @@ func (b *ChainBackend) Start() error { b.mu.Unlock() b.wg.Add(1) - go b.handleTipEvents(sub) + go b.handleChainEvents(chainSub) b.log.InfoS(context.Background(), "Chain backend started", slog.Int("tip_height", int(height)), @@ -388,11 +468,20 @@ func (b *ChainBackend) SubmitPackage(ctx context.Context, parents []*wire.MsgTx, } // RegisterConf registers for confirmation notifications of a transaction. +// The registration is reorg-aware: the returned ConfRegistration's +// Reorged channel fires when a previously delivered confirmation is +// reorged out of the canonical chain, and the Confirmed channel may +// fire again when the tx re-confirms on the new chain. The Done +// channel is allocated but never written by this backend; the +// chainsource ConfActor synthesizes Done at its configured +// FinalityDepth using block epochs. func (b *ChainBackend) RegisterConf(ctx context.Context, txid *chainhash.Hash, pkScript []byte, numConfs uint32, heightHint uint32, includeBlock bool) (*chainsource.ConfRegistration, error) { confChan := make(chan *chainsource.TxConfirmation, 1) + reorgChan := make(chan uint64, 1) + doneChan := make(chan struct{}, 1) cancelCh := make(chan struct{}) reg := &confRegistration{ @@ -402,6 +491,8 @@ func (b *ChainBackend) RegisterConf(ctx context.Context, txid *chainhash.Hash, heightHint: heightHint, includeBlock: includeBlock, confChan: confChan, + reorgChan: reorgChan, + doneChan: doneChan, cancelCh: cancelCh, } @@ -424,11 +515,7 @@ func (b *ChainBackend) RegisterConf(ctx context.Context, txid *chainhash.Hash, ) cancelFn := func() { - close(cancelCh) - - b.mu.Lock() - delete(b.confRegs, id) - b.mu.Unlock() + b.cancelConfReg(id, reg) } // Run an immediate single-shot check scoped to JUST this @@ -450,16 +537,41 @@ func (b *ChainBackend) RegisterConf(ctx context.Context, txid *chainhash.Hash, return &chainsource.ConfRegistration{ Confirmed: confChan, + Reorged: reorgChan, + Done: doneChan, Cancel: cancelFn, }, nil } +// cancelConfReg tears down a confirmation registration. It is safe +// to invoke from any state; the close(cancelCh) is guarded by a +// once-style check to make double-Cancel a no-op. +func (b *ChainBackend) cancelConfReg(id uint64, reg *confRegistration) { + reg.regMu.Lock() + select { + case <-reg.cancelCh: + // Already cancelled. + reg.regMu.Unlock() + + return + + default: + } + close(reg.cancelCh) + reg.regMu.Unlock() + + b.mu.Lock() + delete(b.confRegs, id) + b.mu.Unlock() +} + // 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. +// registration is NOT deleted after delivery: a reorg may later +// reset it to stateWatching and the next re-check must fire +// Confirmed again on the new chain. func (b *ChainBackend) runConfOneShot(id uint64, reg *confRegistration) { defer b.wg.Done() @@ -477,11 +589,49 @@ func (b *ChainBackend) runConfOneShot(id uint64, reg *confRegistration) { currentHeight := b.bestHeight b.mu.Unlock() + b.deliverConfIfNew(id, reg, currentHeight) +} + +// deliverConfIfNew runs a confirmation re-check and fires the +// Confirmed channel iff the registration is in stateWatching and a +// confirmation is now available. It is the single delivery path +// shared by the one-shot at registration time and the broad +// re-check driven by tip / reorg events. The registration's regMu +// is held only across the state read and the state transition; the +// channel send happens outside the lock so a slow consumer never +// blocks the broad re-check goroutine. +func (b *ChainBackend) deliverConfIfNew(id uint64, reg *confRegistration, + currentHeight int32) { + + reg.regMu.Lock() + if reg.state == statePositive { + reg.regMu.Unlock() + + return + } + reg.regMu.Unlock() + conf := b.checkSingleConf(reg, currentHeight) if conf == nil { return } + // Re-check state under the lock: a concurrent reorg handler + // could have flipped us back to stateWatching after the check + // started, but it could not have flipped us forward to + // statePositive (that's exclusively this function's job). + reg.regMu.Lock() + if reg.state == statePositive { + reg.regMu.Unlock() + + return + } + reg.state = statePositive + if conf.BlockHash != nil { + reg.lastBlockHash = *conf.BlockHash + } + reg.regMu.Unlock() + select { case reg.confChan <- conf: case <-reg.cancelCh: @@ -493,22 +643,26 @@ func (b *ChainBackend) runConfOneShot(id uint64, reg *confRegistration) { b.log.DebugS( context.Background(), - "Confirmation registration fulfilled (one-shot)", + "Confirmation registration fulfilled", 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. +// The registration is reorg-aware: the returned SpendRegistration's +// Reorged channel fires when a previously delivered spend is reorged +// out of the canonical chain, and the Spend channel may fire again +// when the outpoint is re-spent on the new chain. The Done channel +// is allocated but never written by this backend; the chainsource +// SpendActor synthesizes Done at its configured FinalityDepth. func (b *ChainBackend) RegisterSpend(ctx context.Context, outpoint *wire.OutPoint, pkScript []byte, heightHint uint32) ( *chainsource.SpendRegistration, error) { spendChan := make(chan *chainsource.SpendDetail, 1) + reorgChan := make(chan uint64, 1) + doneChan := make(chan struct{}, 1) cancelCh := make(chan struct{}) reg := &spendRegistration{ @@ -516,6 +670,8 @@ func (b *ChainBackend) RegisterSpend(ctx context.Context, pkScript: pkScript, heightHint: heightHint, spendChan: spendChan, + reorgChan: reorgChan, + doneChan: doneChan, cancelCh: cancelCh, } @@ -537,11 +693,7 @@ func (b *ChainBackend) RegisterSpend(ctx context.Context, ) cancelFn := func() { - close(cancelCh) - - b.mu.Lock() - delete(b.spendRegs, id) - b.mu.Unlock() + b.cancelSpendReg(id, reg) } // Per-registration one-shot to handle outpoints that are @@ -552,16 +704,37 @@ func (b *ChainBackend) RegisterSpend(ctx context.Context, go b.runSpendOneShot(id, reg) return &chainsource.SpendRegistration{ - Spend: spendChan, - Cancel: cancelFn, + Spend: spendChan, + Reorged: reorgChan, + Done: doneChan, + Cancel: cancelFn, }, nil } +// cancelSpendReg tears down a spend registration. Idempotent: a +// double-Cancel is treated as a no-op. +func (b *ChainBackend) cancelSpendReg(id uint64, reg *spendRegistration) { + reg.regMu.Lock() + select { + case <-reg.cancelCh: + reg.regMu.Unlock() + + return + + default: + } + close(reg.cancelCh) + reg.regMu.Unlock() + + b.mu.Lock() + delete(b.spendRegs, id) + b.mu.Unlock() +} + // 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. +// triggered at RegisterSpend time. The registration is NOT deleted +// after delivery: a reorg may later reset it to stateWatching and +// the next re-check must fire Spend again on the new chain. func (b *ChainBackend) runSpendOneShot(id uint64, reg *spendRegistration) { defer b.wg.Done() @@ -579,11 +752,46 @@ func (b *ChainBackend) runSpendOneShot(id uint64, reg *spendRegistration) { return } - detail := b.checkSingleSpend(reg) + b.deliverSpendIfNew(id, reg) +} + +// deliverSpendIfNew runs a spend re-check and fires the Spend +// channel iff the registration is in stateWatching and a spend is +// now available. Mirror of deliverConfIfNew; see that function's +// comment for the regMu / channel-send ordering rationale. +func (b *ChainBackend) deliverSpendIfNew(id uint64, reg *spendRegistration) { + reg.regMu.Lock() + if reg.state == statePositive { + reg.regMu.Unlock() + + return + } + reg.regMu.Unlock() + + // checkSingleSpend returns the spending block hash alongside + // the detail (parsed from the same /outspend response that + // confirmed the spend). A zero hash here means the response + // did not parse and reorgSpendReg will fall back to a + // conservative re-check rather than try to match against + // ReorgEvent.Disconnected. + detail, spendingBlockHash := b.checkSingleSpend(reg) if detail == nil { return } + reg.regMu.Lock() + if reg.state == statePositive { + reg.regMu.Unlock() + + return + } + reg.state = statePositive + if detail.SpenderTxHash != nil { + reg.lastSpenderHash = *detail.SpenderTxHash + } + reg.lastSpendingBlockHash = spendingBlockHash + reg.regMu.Unlock() + select { case reg.spendChan <- detail: case <-reg.cancelCh: @@ -594,15 +802,11 @@ func (b *ChainBackend) runSpendOneShot(id uint64, reg *spendRegistration) { } b.log.DebugS(context.Background(), - "Spend registration fulfilled (one-shot)", + "Spend registration fulfilled", 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. @@ -637,17 +841,258 @@ func (b *ChainBackend) RegisterBlocks(_ context.Context) ( }, nil } -// 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) { +// processReorgEvent reconciles all active registrations with one +// ReorgEvent. It is invoked from the unified handleChainEvents +// goroutine in producer order: a ReorgEvent is fully processed +// (registration state reset, Reorged channels fired) before any +// subsequent TipBlock dispatches block epochs or runs broad +// re-check on the replacement chain. Registrations whose last +// positive event names a disconnected hash are reset and +// re-checked; all others are left alone (the broad tip-driven +// re-check still runs separately). +func (b *ChainBackend) processReorgEvent(event *ReorgEvent) { + if event == nil { + return + } + + b.log.InfoS(context.Background(), "Processing reorg", + slog.Int("fork_height", int(event.ForkHeight)), + slog.Int("disconnected", len(event.Disconnected)), + slog.Int("connected", len(event.Connected)), + ) + + disconnectedSet := make( + map[chainhash.Hash]struct{}, len(event.Disconnected), + ) + for _, hash := range event.Disconnected { + disconnectedSet[hash] = struct{}{} + } + + // Snapshot the registration sets under the chain backend + // lock so we don't hold it across the per-registration + // channel sends below. + b.mu.Lock() + confRegs := make(map[uint64]*confRegistration, len(b.confRegs)) + for id, reg := range b.confRegs { + confRegs[id] = reg + } + spendRegs := make(map[uint64]*spendRegistration, len(b.spendRegs)) + for id, reg := range b.spendRegs { + spendRegs[id] = reg + } + currentHeight := b.bestHeight + b.mu.Unlock() + + for id, reg := range confRegs { + b.reorgConfReg(id, reg, disconnectedSet, currentHeight) + } + for id, reg := range spendRegs { + b.reorgSpendReg(id, reg, disconnectedSet) + } +} + +// reorgConfReg processes one confirmation registration against a +// reorg event. If the registration is in statePositive and its +// last-known block hash is in disconnectedSet, fire Reorged, reset +// to stateWatching, and run a fresh check that may immediately +// fire Confirmed against the new chain. +func (b *ChainBackend) reorgConfReg(id uint64, reg *confRegistration, + disconnectedSet map[chainhash.Hash]struct{}, currentHeight int32) { + + select { + case <-reg.cancelCh: + return + + default: + } + + reg.regMu.Lock() + if reg.state != statePositive { + reg.regMu.Unlock() + + return + } + cachedHash := reg.lastBlockHash + reg.regMu.Unlock() + + // Fast path: the registration's cached block hash appears in + // the reorg event's disconnected set. This covers every + // reorg of a block we previously broadcast. + _, fastHit := disconnectedSet[cachedHash] + + if !fastHit { + // Fallback: the registration may have delivered against + // a block older than the poller's seeded hash history + // (e.g. on a fresh daemon where RegisterConf landed an + // immediate historical positive). The reorg's + // disconnected set is bounded by recentHashes, so a + // reorg deep enough to invalidate that historical block + // would not appear here. Re-query canonical status and + // compare against the cached block hash. + // + // confirmedBlockHash returns a tri-state so we never + // fire a spurious reorg: Some(h) means the tx is still + // confirmed in block h (independent of the numConfs + // threshold), None means it is definitively unconfirmed, + // and a non-nil error means canonical status could not + // be determined right now. + gotHash, err := b.confirmedBlockHash(reg) + switch { + // Transient backend failure: we cannot tell whether this + // is a reorg. Leave the registration in statePositive and + // bail; a later tip or reorg event re-evaluates once the + // backend recovers. Firing Reorged here would strand the + // consumer on a false alarm. + case err != nil: + b.log.DebugS( + context.Background(), + "Skipping conf reorg eval; canonical "+ + "status undeterminable", + slog.Uint64("reg_id", id), + btclog.Fmt("err", "%v", err), + ) + + return + + // Still confirmed in the same block: definitively not a + // reorg of this registration. + case gotHash.IsSome() && + gotHash.UnsafeFromSome() == cachedHash: + return + } + + // Otherwise the tx is confirmed elsewhere or definitively + // unconfirmed: fall through and fire Reorged. + } + + reg.regMu.Lock() + if reg.state != statePositive { + reg.regMu.Unlock() + + return + } + reg.state = stateWatching + reg.lastBlockHash = chainhash.Hash{} + reg.regMu.Unlock() + + // Fire Reorged non-blocking: the channel is buffered to 1 + // and the consumer (chainsource conf actor) is single- + // threaded so a missed coalesced reorg is correct — the + // consumer will re-query state anyway. + select { + case reg.reorgChan <- uint64(0): + case <-reg.cancelCh: + return + + default: + b.log.DebugS( + context.Background(), + "Conf reorged signal coalesced", + slog.Uint64("reg_id", id), + ) + } + + b.log.InfoS( + context.Background(), + "Conf registration reorged; re-checking", + slog.Uint64("reg_id", id), + ) + + // Re-check now so a re-confirmation on the new chain fires + // Confirmed in the same reorg-handler turn. + b.deliverConfIfNew(id, reg, currentHeight) +} + +// reorgSpendReg is the spend-side equivalent of reorgConfReg. +func (b *ChainBackend) reorgSpendReg(id uint64, reg *spendRegistration, + disconnectedSet map[chainhash.Hash]struct{}) { + + select { + case <-reg.cancelCh: + return + + default: + } + + reg.regMu.Lock() + if reg.state != statePositive { + reg.regMu.Unlock() + + return + } + cachedBlockHash := reg.lastSpendingBlockHash + cachedSpenderHash := reg.lastSpenderHash + reg.regMu.Unlock() + + // Fast path: the cached spending-block hash appears in the + // reorg event's disconnected set. Empty cachedBlockHash + // (e.g. delivered before the outspend response was parseable) + // falls through to the fallback re-query rather than being + // treated as an automatic hit. + fastHit := false + if cachedBlockHash != (chainhash.Hash{}) { + _, fastHit = disconnectedSet[cachedBlockHash] + } + + if !fastHit { + // Fallback: re-query canonical chain. Covers (a) regs + // delivered against a block older than the poller's + // seeded hash history, and (b) regs whose + // cachedBlockHash is zero because the outspend response + // was unparseable at delivery time. If the outpoint is + // still spent by the same spender in the same block, + // no reorg for this registration. + current, currentBlock := b.checkSingleSpend(reg) + if current != nil && + current.SpenderTxHash != nil && + *current.SpenderTxHash == cachedSpenderHash && + cachedBlockHash != (chainhash.Hash{}) && + currentBlock == cachedBlockHash { + return + } + } + + reg.regMu.Lock() + if reg.state != statePositive { + reg.regMu.Unlock() + + return + } + reg.state = stateWatching + reg.lastSpenderHash = chainhash.Hash{} + reg.lastSpendingBlockHash = chainhash.Hash{} + reg.regMu.Unlock() + + select { + case reg.reorgChan <- uint64(0): + case <-reg.cancelCh: + return + + default: + b.log.DebugS( + context.Background(), + "Spend reorged signal coalesced", + slog.Uint64("reg_id", id), + ) + } + + b.log.InfoS( + context.Background(), + "Spend registration reorged; re-checking", + slog.Uint64("reg_id", id), + ) + + b.deliverSpendIfNew(id, reg) +} + +// handleChainEvents drains the unified chain stream and dispatches +// each event in producer order. Using a single subscription + single +// goroutine guarantees that a ReorgEvent is fully processed +// (registration state reset, Reorged channels fired) before any +// post-reorg TipBlock drives block-epoch dispatch or broad re-check +// — otherwise a block-epoch driven Confirmed could land on a stale +// registration before reorgConfReg has a chance to reset it. +func (b *ChainBackend) handleChainEvents(sub *ChainSubscription) { defer b.wg.Done() defer sub.Cancel() @@ -661,18 +1106,15 @@ func (b *ChainBackend) handleTipEvents(sub *TipSubscription) { return } - b.processTipEvent(event) + switch { + case event.Reorg != nil: + b.processReorgEvent(event.Reorg) + + case event.Tip != nil: + b.processTipEvent(event.Tip) + } 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(): @@ -774,8 +1216,13 @@ func (b *ChainBackend) processTipEvent(event *TipBlock) { }) } -// checkConfirmations iterates over all pending confirmation registrations -// and checks their status via the Esplora API. +// checkConfirmations iterates over all pending confirmation +// registrations and re-checks their status via the Esplora API. +// Registrations already in statePositive are skipped: the reorg +// handler is responsible for transitioning them back to +// stateWatching, and an unconditional re-check would just burn +// Esplora calls for no behavior change (the chainsource actor's +// FinalityDepth synthesis handles Done from block epochs). func (b *ChainBackend) checkConfirmations() { b.mu.Lock() regs := make(map[uint64]*confRegistration, len(b.confRegs)) @@ -794,28 +1241,7 @@ func (b *ChainBackend) checkConfirmations() { default: } - conf := b.checkSingleConf(reg, currentHeight) - if conf == nil { - continue - } - - // Send the confirmation. - select { - case reg.confChan <- conf: - case <-reg.cancelCh: - continue - } - - b.log.DebugS(context.Background(), - "Confirmation registration fulfilled", - slog.Uint64("reg_id", id), - slog.Int("block_height", - int(conf.BlockHeight))) - - // Remove the fulfilled registration. - b.mu.Lock() - delete(b.confRegs, id) - b.mu.Unlock() + b.deliverConfIfNew(id, reg, currentHeight) } } @@ -948,8 +1374,73 @@ func (b *ChainBackend) checkConfByScript(reg *confRegistration, return nil } -// checkSpends iterates over all pending spend registrations and checks -// their status via the Esplora API. +// confirmedBlockHash reports the block hash a registration's transaction is +// currently confirmed in, independent of the registration's numConfs +// threshold. It exists for the reorg-detection fallback, which must +// distinguish three states that checkSingleConf collapses into a single nil: +// +// - Some(hash), nil: the tx is confirmed in block hash (any depth); +// - None, nil: the tx is definitively unconfirmed (genuine reorg); +// - None, err: canonical status could not be determined right now +// (transient backend error) — callers MUST NOT treat this as a reorg. +// +// Unlike checkSingleConf it never gates on numConfs, because a reorg that +// merely reduces a tx's confirmation depth without moving it out of its block +// is not a reorg of that registration. +func (b *ChainBackend) confirmedBlockHash(reg *confRegistration) ( + fn.Option[chainhash.Hash], error) { + + none := fn.None[chainhash.Hash]() + + // Explicit-txid registrations resolve via a direct status lookup. + if reg.txid != nil { + status, err := b.esplora.GetTxStatus( + context.Background(), *reg.txid, + ) + if err != nil { + return none, err + } + if !status.Confirmed { + return none, nil + } + + hash, err := chainhash.NewHashFromStr(status.BlockHash) + if err != nil { + return none, err + } + + return fn.Some(*hash), nil + } + + // Script registrations resolve via the first confirmed UTXO paying + // the watched script. + utxos, err := b.esplora.GetScriptUtxos( + context.Background(), reg.pkScript, + ) + if err != nil { + return none, err + } + + for _, utxo := range utxos { + if !utxo.Status.Confirmed { + continue + } + + hash, err := chainhash.NewHashFromStr(utxo.Status.BlockHash) + if err != nil { + return none, err + } + + return fn.Some(*hash), nil + } + + return none, nil +} + +// checkSpends iterates over all pending spend registrations and +// re-checks their status via the Esplora API. Registrations in +// statePositive are skipped — see checkConfirmations for the +// rationale. func (b *ChainBackend) checkSpends() { b.mu.Lock() regs := make(map[uint64]*spendRegistration, len(b.spendRegs)) @@ -971,68 +1462,59 @@ func (b *ChainBackend) checkSpends() { continue } - detail := b.checkSingleSpend(reg) - if detail == nil { - continue - } - - // Send the spend detail. - select { - case reg.spendChan <- detail: - case <-reg.cancelCh: - continue - } - - b.log.DebugS(context.Background(), - "Spend registration fulfilled", - slog.Uint64("reg_id", id), - slog.String("outpoint", - reg.outpoint.String()), - slog.String( - "spender_txid", detail.SpenderTxHash.String(), - )) - - // Remove the fulfilled registration. - b.mu.Lock() - delete(b.spendRegs, id) - b.mu.Unlock() + b.deliverSpendIfNew(id, reg) } } // 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 +// registration has no outpoint. The second return value is the block +// hash containing the spending tx (parsed from outspend.Status); the +// zero hash indicates the spending block hash was unparseable but the +// spend itself is still valid. 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 { +func (b *ChainBackend) checkSingleSpend(reg *spendRegistration) ( + *chainsource.SpendDetail, chainhash.Hash) { if reg.outpoint == nil { - return nil + return nil, chainhash.Hash{} } outspend, err := b.esplora.GetOutspend( context.Background(), reg.outpoint.Hash, reg.outpoint.Index, ) if err != nil { - return nil + return nil, chainhash.Hash{} } if !outspend.Spent || !outspend.Status.Confirmed { - return nil + return nil, chainhash.Hash{} } spenderHash, err := chainhash.NewHashFromStr(outspend.Txid) if err != nil { - return nil + return nil, chainhash.Hash{} } spendingTx, err := b.esplora.GetRawTx( context.Background(), *spenderHash, ) if err != nil { - return nil + return nil, chainhash.Hash{} + } + + // outspend.Status.BlockHash is hex from the same /outspend + // response that confirmed the spend; an unparseable value is + // not fatal — the spend itself is valid and the reorg path can + // fall back to a conservative re-check (see reorgSpendReg). + var spendingBlockHash chainhash.Hash + if outspend.Status.BlockHash != "" { + h, err := chainhash.NewHashFromStr(outspend.Status.BlockHash) + if err == nil { + spendingBlockHash = *h + } } return &chainsource.SpendDetail{ @@ -1041,7 +1523,7 @@ func (b *ChainBackend) checkSingleSpend( SpendingTx: spendingTx, SpenderInputIndex: outspend.Vin, SpendingHeight: int32(outspend.Status.BlockHeight), - } + }, spendingBlockHash } // Compile-time check that ChainBackend implements diff --git a/lwwallet/chain_backend_reorg_test.go b/lwwallet/chain_backend_reorg_test.go new file mode 100644 index 000000000..593138b38 --- /dev/null +++ b/lwwallet/chain_backend_reorg_test.go @@ -0,0 +1,1136 @@ +package lwwallet + +import ( + "bytes" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "runtime" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/btcsuite/btclog/v2" + "github.com/lightninglabs/darepo-client/chainsource" + "github.com/stretchr/testify/require" +) + +// reorgTestTimeout is the per-step wait used by the lwwallet reorg +// tests. Generous enough to absorb scheduling jitter on overloaded +// CI machines but short enough that a hung backend surfaces as a +// fast failure. +const reorgTestTimeout = 3 * time.Second + +// fakeChain is a mutable chain fixture for the reorg tests. Each +// height maps to a block whose hash, raw header, and contents the +// embedded HTTP handler will serve. Hashes are computable so the +// EsploraClient's content-hash verification passes. +// +// fakeChain is intentionally orthogonal to stubChain (used by the +// existing tip-poller tests): it supports same-height hash +// replacement (reorg), per-tx status overrides, per-outpoint +// outspend overrides, and serves /block//header so the tip +// poller's continuity check can resolve PrevBlock. +type fakeChain struct { + t *testing.T + mu sync.Mutex + + // tip is the current tip height. + tip int32 + + // blocks holds the per-height block model. Replacing the entry + // at height h simulates a same-height reorg; appending new + // heights simulates chain advance. + blocks map[int32]*fakeBlock + + // txStatus is the response to /tx//status. + txStatus map[chainhash.Hash]esploraTxStatus + + // rawTx is the response to /tx//raw. + rawTx map[chainhash.Hash][]byte + + // outspends is the response to /tx//outspend/. + outspends map[wire.OutPoint]esploraOutspend + + // failRawHeader holds block hashes for which the + // /block//header endpoint should return 500. Used to + // simulate a transient Esplora flake during continuity + // checking. + failRawHeader map[chainhash.Hash]struct{} +} + +// fakeBlock describes one height's block in the fakeChain. The hash +// is derived from a synthetic 80-byte header so the EsploraClient's +// content-hash verification accepts the raw-header response. +type fakeBlock struct { + height int32 + hash chainhash.Hash + prevHash chainhash.Hash + header *wire.BlockHeader + timestamp int64 +} + +// newFakeChain seeds a fakeChain with a single block at the given +// tip height. tag is mixed into the block hash so independent +// fakeChains in parallel tests produce distinct hashes. +func newFakeChain(t *testing.T, tip int32, tag string) *fakeChain { + t.Helper() + + c := &fakeChain{ + t: t, + tip: tip, + blocks: make(map[int32]*fakeBlock), + txStatus: make(map[chainhash.Hash]esploraTxStatus), + rawTx: make(map[chainhash.Hash][]byte), + outspends: make(map[wire.OutPoint]esploraOutspend), + failRawHeader: make(map[chainhash.Hash]struct{}), + } + + c.blocks[tip] = c.mintBlock(tip, chainhash.Hash{}, tag) + + return c +} + +// mintBlock builds a fakeBlock at height with the given prev hash +// and a tag that varies the resulting block hash. We synthesize a +// minimal 80-byte header with a unique nonce so BlockHash() varies +// reliably across invocations. +func (c *fakeChain) mintBlock(height int32, prev chainhash.Hash, + tag string) *fakeBlock { + + c.t.Helper() + + hdr := &wire.BlockHeader{ + Version: 1, + PrevBlock: prev, + MerkleRoot: chainhash.HashH([]byte(tag + "-merkle")), + Timestamp: time.Unix(int64(height)*600, 0), + Bits: 0x207fffff, + Nonce: uint32(height) ^ + uint32(chainhash.HashH([]byte(tag)).String()[0])<<16, + } + + // Make the nonce truly unique per (height, tag) so two + // adjacent tags do not collide on PrevBlock=zero replays. + salt := chainhash.HashH([]byte(fmt.Sprintf("%d-%s", height, tag))) + hdr.Nonce = uint32(salt[0])<<24 | uint32(salt[1])<<16 | + uint32(salt[2])<<8 | uint32(salt[3]) + + return &fakeBlock{ + height: height, + hash: hdr.BlockHash(), + prevHash: prev, + header: hdr, + timestamp: hdr.Timestamp.Unix(), + } +} + +// replaceTip swaps in a brand-new block at the current tip height +// (same height, different hash) and returns the new block. Used to +// drive a same-height reorg. +func (c *fakeChain) replaceTip(tag string) *fakeBlock { + c.mu.Lock() + defer c.mu.Unlock() + + prev, ok := c.blocks[c.tip-1] + var prevHash chainhash.Hash + if ok { + prevHash = prev.hash + } + + blk := c.mintBlock(c.tip, prevHash, tag) + c.blocks[c.tip] = blk + + return blk +} + +// extend appends a new block on top of the current tip and returns +// it. tag varies the resulting hash for parallel tests. +func (c *fakeChain) extend(tag string) *fakeBlock { + c.mu.Lock() + defer c.mu.Unlock() + + tipBlk := c.blocks[c.tip] + blk := c.mintBlock(c.tip+1, tipBlk.hash, tag) + c.tip++ + c.blocks[c.tip] = blk + + return blk +} + +// rewriteFrom rebuilds the chain from height start upward, giving +// each new block a different hash than the previous occupant at +// the same height. Used to simulate a deeper reorg where multiple +// heights diverge at once. +func (c *fakeChain) rewriteFrom(start int32, tagPrefix string) { + c.mu.Lock() + defer c.mu.Unlock() + + var prev chainhash.Hash + if prior, ok := c.blocks[start-1]; ok { + prev = prior.hash + } + for h := start; h <= c.tip; h++ { + blk := c.mintBlock(h, prev, + fmt.Sprintf("%s-%d", tagPrefix, h)) + c.blocks[h] = blk + prev = blk.hash + } +} + +// setTxStatus pins the response for /tx//status. +func (c *fakeChain) setTxStatus(txid chainhash.Hash, status esploraTxStatus) { + c.mu.Lock() + defer c.mu.Unlock() + + c.txStatus[txid] = status +} + +// setRawTx pins the response for /tx//raw. +func (c *fakeChain) setRawTx(txid chainhash.Hash, raw []byte) { + c.mu.Lock() + defer c.mu.Unlock() + + c.rawTx[txid] = raw +} + +// setOutspend pins the response for /tx//outspend/. +func (c *fakeChain) setOutspend(op wire.OutPoint, outspend esploraOutspend) { + c.mu.Lock() + defer c.mu.Unlock() + + c.outspends[op] = outspend +} + +// handler returns an http.HandlerFunc that serves the routes the +// chain backend / tip poller exercise. Anything else returns 404. +func (c *fakeChain) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + c.mu.Lock() + defer c.mu.Unlock() + + path := r.URL.Path + switch { + case path == "/blocks/tip/height": + _, _ = fmt.Fprint(w, c.tip) + + case path == "/blocks/tip/hash": + blk, ok := c.blocks[c.tip] + if !ok { + http.Error(w, "no tip", http.StatusNotFound) + + return + } + + _, _ = fmt.Fprint(w, blk.hash.String()) + + case strings.HasPrefix(path, "/block-height/"): + heightStr := strings.TrimPrefix( + path, "/block-height/", + ) + height, err := strconv.ParseInt(heightStr, 10, 32) + if err != nil { + http.Error( + w, "bad height", http.StatusBadRequest, + ) + + return + } + blk, ok := c.blocks[int32(height)] + if !ok { + http.Error(w, "not found", + http.StatusNotFound) + + return + } + + _, _ = fmt.Fprint(w, blk.hash.String()) + + case strings.HasPrefix(path, "/block/"): + c.serveBlockReq(w, r, path) + + case strings.HasPrefix(path, "/tx/"): + c.serveTxReq(w, r, path) + + default: + http.Error(w, "not found", http.StatusNotFound) + } + } +} + +// serveBlockReq handles /block/ and /block//header. +// Caller holds c.mu. +func (c *fakeChain) serveBlockReq(w http.ResponseWriter, _ *http.Request, + path string) { + + rest := strings.TrimPrefix(path, "/block/") + hashStr := rest + suffix := "" + if idx := strings.Index(rest, "/"); idx >= 0 { + hashStr = rest[:idx] + suffix = rest[idx:] + } + + hash, err := chainhash.NewHashFromStr(hashStr) + if err != nil { + http.Error(w, "bad hash", http.StatusBadRequest) + + return + } + + var found *fakeBlock + for _, b := range c.blocks { + if b.hash == *hash { + found = b + + break + } + } + if found == nil { + http.Error(w, "not found", http.StatusNotFound) + + return + } + + switch suffix { + case "": + // JSON header. + _, _ = fmt.Fprintf( + w, `{"id":%q,"height":%d,"timestamp":%d}`, + found.hash.String(), found.height, found.timestamp, + ) + + case "/header": + // Raw 80-byte header, hex-encoded. + if _, fail := c.failRawHeader[found.hash]; fail { + http.Error( + w, "raw header unavailable", + http.StatusInternalServerError, + ) + + return + } + var buf bytes.Buffer + require.NoError(c.t, found.header.Serialize(&buf)) + _, _ = fmt.Fprint(w, hex.EncodeToString(buf.Bytes())) + + default: + http.Error(w, "not implemented", + http.StatusNotImplemented) + } +} + +// serveTxReq handles /tx//status, /tx//raw, and +// /tx//outspend/. Caller holds c.mu. +func (c *fakeChain) serveTxReq(w http.ResponseWriter, _ *http.Request, + path string) { + + rest := strings.TrimPrefix(path, "/tx/") + parts := strings.SplitN(rest, "/", 3) + if len(parts) < 2 { + http.Error(w, "not found", http.StatusNotFound) + + return + } + + txid, err := chainhash.NewHashFromStr(parts[0]) + if err != nil { + http.Error(w, "bad txid", http.StatusBadRequest) + + return + } + + switch parts[1] { + case "status": + status, ok := c.txStatus[*txid] + if !ok { + // Default to unconfirmed when no override exists. + status = esploraTxStatus{Confirmed: false} + } + + err := json.NewEncoder(w).Encode(status) + require.NoError(c.t, err) + + case "raw": + raw, ok := c.rawTx[*txid] + if !ok { + http.Error(w, "not found", + http.StatusNotFound) + + return + } + _, err := w.Write(raw) + require.NoError(c.t, err) + + case "outspend": + if len(parts) < 3 { + http.Error(w, "not found", + http.StatusNotFound) + + return + } + vout, err := strconv.ParseUint(parts[2], 10, 32) + if err != nil { + http.Error(w, "bad vout", + http.StatusBadRequest) + + return + } + op := wire.OutPoint{ + Hash: *txid, Index: uint32(vout), + } + outspend, ok := c.outspends[op] + if !ok { + outspend = esploraOutspend{Spent: false} + } + err = json.NewEncoder(w).Encode(outspend) + require.NoError(c.t, err) + + default: + http.Error(w, "not found", http.StatusNotFound) + } +} + +// fakeChainServer wraps a fakeChain with an httptest.Server for +// drop-in use by the backend tests. The server is auto-closed via +// t.Cleanup. +func fakeChainServer(t *testing.T, chain *fakeChain) *httptest.Server { + t.Helper() + + srv := httptest.NewServer(chain.handler()) + t.Cleanup(srv.Close) + + return srv +} + +// awaitConf reads one TxConfirmation with a deadline. +func awaitConf(t *testing.T, ch <-chan *chainsourceConf) *chainsourceConf { + t.Helper() + + select { + case c, ok := <-ch: + require.True(t, ok, "conf channel closed unexpectedly") + + return c + + case <-time.After(reorgTestTimeout): + t.Fatal("timeout waiting for confirmation") + + return nil + } +} + +// awaitSpend reads one SpendDetail with a deadline. +func awaitSpend(t *testing.T, ch <-chan *chainsourceSpend) *chainsourceSpend { + t.Helper() + + select { + case s, ok := <-ch: + require.True(t, ok, "spend channel closed unexpectedly") + + return s + + case <-time.After(reorgTestTimeout): + t.Fatal("timeout waiting for spend") + + return nil + } +} + +// awaitSeqSignal is awaitSignal for the sequence-carrying Reorged +// channel (chainsource retyped it from struct{} to the ordering +// sequence; this backend always sends 0). +func awaitSeqSignal(t *testing.T, ch <-chan uint64, label string) { + t.Helper() + + select { + case _, ok := <-ch: + require.True(t, ok, + "%s channel closed unexpectedly", label) + + case <-time.After(reorgTestTimeout): + t.Fatalf("timeout waiting for %s", label) + } +} + +// requireQuiet asserts that ch has no event for a short window. +// Used to ensure registrations are NOT double-firing. +func requireQuiet(t *testing.T, ch <-chan struct{}, label string, + dur time.Duration) { + + t.Helper() + + select { + case <-ch: + t.Fatalf("unexpected %s signal", label) + + case <-time.After(dur): + } +} + +// chainsourceConf / chainsourceSpend are type aliases to keep the +// test helper signatures short. +type ( + chainsourceConf = chainsource.TxConfirmation + chainsourceSpend = chainsource.SpendDetail +) + +// TestChainBackendSameHeightHashDrift verifies that a same-height +// hash replacement is detected by the tip poller and routed to the +// chain backend as a reorg, even when the chain height does not +// advance. This is the core "same-height reorgs invisible" gap the +// PR closes. +func TestChainBackendSameHeightHashDrift(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "drift-init") + srv := fakeChainServer(t, chain) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + backend := NewChainBackend( + esplora, 20*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + // Subscribe directly to the poller's reorg stream to verify + // detection. The backend also subscribes; both subscribers + // must observe the reorg. + reorgSub, err := backend.tipPoller.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + // Replace the tip block at the SAME height with a new hash. + chain.replaceTip("drift-new") + + select { + case ev := <-reorgSub.Updates(): + require.NotNil(t, ev) + require.Equal(t, int32(99), ev.ForkHeight) + require.Len(t, ev.Disconnected, 1) + require.Len(t, ev.Connected, 1) + require.Equal(t, int32(100), ev.Connected[0].Height) + + case <-time.After(reorgTestTimeout): + t.Fatal("timed out waiting for same-height reorg") + } +} + +// TestChainBackendConfReorgRoundTrip drives a registration through +// Confirmed -> Reorged -> Confirmed by replacing the block that +// confirmed the tx with a new block at the same height that still +// confirms the tx (different BlockHash). +func TestChainBackendConfReorgRoundTrip(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "conf-init") + srv := fakeChainServer(t, chain) + + // Use minimalRawTx so the EsploraClient's content-hash + // verification accepts the raw-tx response. + rawTx := minimalRawTx() + txid := minimalRawTxID(t) + chain.setRawTx(txid, rawTx) + + confBlock := chain.blocks[100] + chain.setTxStatus(txid, esploraTxStatus{ + Confirmed: true, + BlockHeight: 100, + BlockHash: confBlock.hash.String(), + }) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + backend := NewChainBackend( + esplora, 20*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + reg, err := backend.RegisterConf( + t.Context(), &txid, nil, 1, 99, false, + ) + require.NoError(t, err) + defer reg.Cancel() + + conf1 := awaitConf(t, reg.Confirmed) + require.Equal(t, uint32(100), conf1.BlockHeight) + require.Equal(t, confBlock.hash, *conf1.BlockHash) + + // Same-height reorg: replace the confirming block with a + // different one at height 100, and pin the tx status to + // the new block hash so the post-reorg re-check finds the + // tx confirmed in the new block. + newBlock := chain.replaceTip("conf-replaced") + chain.setTxStatus(txid, esploraTxStatus{ + Confirmed: true, + BlockHeight: 100, + BlockHash: newBlock.hash.String(), + }) + + awaitSeqSignal(t, reg.Reorged, "conf Reorged") + + conf2 := awaitConf(t, reg.Confirmed) + require.Equal(t, uint32(100), conf2.BlockHeight) + require.Equal(t, newBlock.hash, *conf2.BlockHash) + require.NotEqual( + t, conf1.BlockHash.String(), conf2.BlockHash.String(), + "reorg should surface a different block hash", + ) +} + +// TestChainBackendConfReorgEvictedDoesNotReConfirm verifies that +// when the post-reorg chain no longer contains the tx, Reorged +// fires but Confirmed does not re-fire. +func TestChainBackendConfReorgEvictedDoesNotReConfirm(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "evict-init") + srv := fakeChainServer(t, chain) + + rawTx := minimalRawTx() + txid := minimalRawTxID(t) + chain.setRawTx(txid, rawTx) + + confBlock := chain.blocks[100] + chain.setTxStatus(txid, esploraTxStatus{ + Confirmed: true, + BlockHeight: 100, + BlockHash: confBlock.hash.String(), + }) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + backend := NewChainBackend( + esplora, 20*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + reg, err := backend.RegisterConf( + t.Context(), &txid, nil, 1, 99, false, + ) + require.NoError(t, err) + defer reg.Cancel() + + awaitConf(t, reg.Confirmed) + + // Reorg evicts the confirming block AND the tx is no longer + // found on the new chain. + chain.replaceTip("evict-new") + chain.setTxStatus(txid, esploraTxStatus{Confirmed: false}) + + awaitSeqSignal(t, reg.Reorged, "conf Reorged") + + // No re-confirmation: assert silence. + requireQuiet( + t, structOnlyChan(reg.Confirmed), + "unexpected re-Confirmed", 500*time.Millisecond, + ) +} + +// structOnlyChan adapts a TxConfirmation channel to a struct{} +// channel for use with requireQuiet. The forwarder goroutine is +// short-lived; once the test ends both channels go out of scope. +func structOnlyChan(in <-chan *chainsourceConf) <-chan struct{} { + out := make(chan struct{}, 4) + + go func() { + for c := range in { + if c == nil { + continue + } + + select { + case out <- struct{}{}: + default: + } + } + }() + + return out +} + +// TestChainBackendSpendReorgRoundTrip drives a spend registration +// through Spend -> Reorged -> Spend by reorging the block that +// confirmed the spending tx and pinning a different (still spent) +// outspend on the new chain. +func TestChainBackendSpendReorgRoundTrip(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "spend-init") + srv := fakeChainServer(t, chain) + + rawTx := minimalRawTx() + spenderTxid := minimalRawTxID(t) + chain.setRawTx(spenderTxid, rawTx) + + fundingTxid := chainhash.HashH([]byte("funding")) + outpoint := wire.OutPoint{Hash: fundingTxid, Index: 0} + + chain.setOutspend(outpoint, esploraOutspend{ + Spent: true, Txid: spenderTxid.String(), Vin: 0, + Status: esploraStatus{ + Confirmed: true, BlockHeight: 100, + }, + }) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + backend := NewChainBackend( + esplora, 20*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + reg, err := backend.RegisterSpend( + t.Context(), &outpoint, nil, 90, + ) + require.NoError(t, err) + defer reg.Cancel() + + spend1 := awaitSpend(t, reg.Spend) + require.Equal(t, spenderTxid, *spend1.SpenderTxHash) + require.Equal(t, int32(100), spend1.SpendingHeight) + + // Replace the confirming block AND swap the spender on the + // new chain. The spender tx id is the same (we're reusing + // minimalRawTx) but the block hash differs, which is what + // the reorg handler keys off. + chain.replaceTip("spend-replaced") + chain.setOutspend(outpoint, esploraOutspend{ + Spent: true, Txid: spenderTxid.String(), Vin: 0, + Status: esploraStatus{ + Confirmed: true, BlockHeight: 100, + }, + }) + + awaitSeqSignal(t, reg.Reorged, "spend Reorged") + + spend2 := awaitSpend(t, reg.Spend) + require.Equal(t, spenderTxid, *spend2.SpenderTxHash) + require.Equal(t, int32(100), spend2.SpendingHeight) +} + +// TestTipPollerSeedsHashHistoryOnStart pins the property that the +// poller seeds its recent-hash ring back through historySize-1 +// heights at Start. Without this, a fresh poller would only ever +// cache the initial tip, leaving any reorg-event consumer with an +// incomplete disconnected set for reorgs deeper than 1 block but +// within the configured history. The chain.Interface consumer +// (EsploraChainService) depends on a complete disconnected set to +// emit a BlockDisconnected for every height btcwallet must roll +// back. +func TestTipPollerSeedsHashHistoryOnStart(t *testing.T) { + t.Parallel() + + // Build a chain at tip 100 plus three lower heights. The poller + // starts at tip 100; with seed-history, recentHashes will be + // pre-populated with {97, 98, 99, 100}. + chain := newFakeChain(t, 100, "seed-100") + // fakeChain only populates the tip block; the test needs lower + // heights in the response. Extend backwards by minting blocks + // at 99, 98, 97 with the correct PrevBlock chain so the + // poller's seed walk resolves each height. + chain.mu.Lock() + var prev chainhash.Hash + for h := int32(97); h <= 100; h++ { + blk := chain.mintBlock(h, prev, fmt.Sprintf("seed-%d", h)) + chain.blocks[h] = blk + prev = blk.hash + } + chain.mu.Unlock() + + srv := fakeChainServer(t, chain) + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + + // Cap historySize at 4 so the seed walk fills exactly the + // heights we care about. + tipPoller := NewTipPollerWithConfig( + esplora, 20*time.Millisecond, 4, btclog.Disabled, + ) + require.NoError(t, tipPoller.Start()) + t.Cleanup(tipPoller.Stop) + + reorgSub, err := tipPoller.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + // Reorg the deepest 3 heights (98, 99, 100). With seed-history + // the poller's recentHashes contains 97, 98, 99, 100 so the + // walk-back terminates at fork point 97 and disconnects 3 + // hashes. Without the seed, recentHashes would only contain + // 100; walk-back would stop at 99 (no cached hash) and the + // disconnected list would carry only 1 hash. + chain.rewriteFrom(98, "seed-rewritten") + + select { + case ev := <-reorgSub.Updates(): + require.NotNil(t, ev) + require.Equal( + t, int32(97), ev.ForkHeight, "fork point should be "+ + "the deepest unchanged height; "+ + "seed-history failed if it lands above 97", + ) + require.Len( + t, ev.Disconnected, 3, "all three reorged heights "+ + "must appear in Disconnected; seed-history "+ + "is what makes this property hold for "+ + "reorgs of depth > 1 against a freshly "+ + "started poller", + ) + + case <-time.After(reorgTestTimeout): + t.Fatal("timed out waiting for reorg event") + } +} + +// TestChainBackendDeeperReorgDetected verifies that a multi-block +// reorg (where height advances AND PrevBlock continuity is broken) +// produces a ReorgEvent with the correct fork point. +// +// The test first advances the chain past the eventual fork point so +// the poller's recent-hash ring buffer contains the to-be-rewritten +// heights. Without that, the walk-back would terminate early on the +// first !haveCached probe and report a shallower fork than the test +// intends to exercise. +func TestChainBackendDeeperReorgDetected(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "deep-init") + + srv := fakeChainServer(t, chain) + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + tp := NewTipPoller(esplora, 20*time.Millisecond, btclog.Disabled) + require.NoError(t, tp.Start()) + t.Cleanup(tp.Stop) + + // Subscribe BEFORE the rewrite so we receive the reorg + // event live. + reorgSub, err := tp.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + tipSub, err := tp.Subscribe() + require.NoError(t, err) + defer tipSub.Cancel() + + // Advance the chain through the poller so it caches + // heights 101 and 102 in its ring buffer. + chain.extend("init-101") + chain.extend("init-102") + + for expected := int32(101); expected <= 102; expected++ { + select { + case ev := <-tipSub.Updates(): + require.Equal(t, expected, ev.Height) + + case <-time.After(reorgTestTimeout): + t.Fatalf("timed out waiting for height %d to be cached", + expected) + } + } + + // Now rewrite from height 101 upward (heights 101 and 102 + // get new hashes) and extend by one to push the new tip + // past the old. + chain.rewriteFrom(101, "fork") + chain.extend("fork-103") + + // Wait for ReorgEvent. ForkHeight should be 100 because + // the rewrite started at 101 (so 100 is the last height + // that agrees between old and new chains). + select { + case ev := <-reorgSub.Updates(): + require.NotNil(t, ev) + require.Equal(t, int32(100), ev.ForkHeight) + require.GreaterOrEqual(t, len(ev.Connected), 2) + require.GreaterOrEqual(t, len(ev.Disconnected), 2) + + case <-time.After(reorgTestTimeout): + t.Fatal("timed out waiting for deeper reorg") + } +} + +// TestChainBackendCancelCleanup verifies that Cancel on a +// conf/spend registration removes it from internal maps and does +// not leak goroutines. +func TestChainBackendCancelCleanup(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "cancel-init") + srv := fakeChainServer(t, chain) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + backend := NewChainBackend( + esplora, time.Hour, btclog.Disabled, + ) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + txid := minimalRawTxID(t) + outpoint := wire.OutPoint{ + Hash: chainhash.HashH([]byte("funding")), Index: 0, + } + + // Sample the baseline goroutine count AFTER the backend is + // started so the poller / tip handler / reorg handler + // goroutines are already included. + time.Sleep(50 * time.Millisecond) + baseline := runtime.NumGoroutine() + + const iterations = 20 + for i := 0; i < iterations; i++ { + confReg, err := backend.RegisterConf( + t.Context(), &txid, nil, 1, 99, false, + ) + require.NoError(t, err) + + spendReg, err := backend.RegisterSpend( + t.Context(), &outpoint, nil, 90, + ) + require.NoError(t, err) + + confReg.Cancel() + spendReg.Cancel() + + // Double Cancel must be a safe no-op. + confReg.Cancel() + spendReg.Cancel() + } + + // Give the spawned one-shot goroutines time to exit. + require.Eventually(t, func() bool { + + // Allow a small tolerance for scheduling jitter; the + // upper bound here is generous (20) versus the + // per-iteration spawn count (2) to absorb any + // in-flight Esplora request goroutines that have not + // yet returned. The key invariant is that the count + // does not GROW unbounded. + return runtime.NumGoroutine() <= baseline+5 + }, 2*time.Second, 50*time.Millisecond, + "goroutines leaked after Cancel: baseline=%d current=%d", + baseline, runtime.NumGoroutine()) + + // After Cancel, internal maps should be empty. + backend.mu.Lock() + confLen := len(backend.confRegs) + spendLen := len(backend.spendRegs) + backend.mu.Unlock() + require.Equal(t, 0, confLen, + "confRegs not cleaned up after Cancel") + require.Equal(t, 0, spendLen, + "spendRegs not cleaned up after Cancel") +} + +// TestChainBackendConfStaysAliveAfterFirstFire verifies that the +// confirmation registration is NOT deleted after the first +// Confirmed delivery, since the chainsource ConfActor needs the +// registration to remain alive to receive future Reorged events +// and finally a synthesized Done. +func TestChainBackendConfStaysAliveAfterFirstFire(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "alive-init") + srv := fakeChainServer(t, chain) + + rawTx := minimalRawTx() + txid := minimalRawTxID(t) + chain.setRawTx(txid, rawTx) + chain.setTxStatus(txid, esploraTxStatus{ + Confirmed: true, + BlockHeight: 100, + BlockHash: chain.blocks[100].hash.String(), + }) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + backend := NewChainBackend( + esplora, 50*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + reg, err := backend.RegisterConf( + t.Context(), &txid, nil, 1, 99, false, + ) + require.NoError(t, err) + defer reg.Cancel() + + awaitConf(t, reg.Confirmed) + + // Give the polling loop time to fire several heartbeats. + // The registration must still be present. + time.Sleep(200 * time.Millisecond) + + backend.mu.Lock() + confLen := len(backend.confRegs) + backend.mu.Unlock() + + require.Equal( + t, 1, confLen, "reg deleted after first fire; reorg-aware "+ + "contract requires it to stay alive", + ) +} + +// TestChainBackendConfReorgBelowSeededHistory pins the property that a +// reorg deeper than the poller's seeded hash history still surfaces +// Reorged for a registration that delivered against the +// now-out-of-window block. The poller's disconnected set is bounded +// by the cached history, so without a canonical re-query fallback the +// reorg signal would be silently dropped. +func TestChainBackendConfReorgBelowSeededHistory(t *testing.T) { + t.Parallel() + + // Seed a chain at height 100 and extend up to 105 so the + // walk-back has live block hashes to compare against. Use a + // tiny history size so the registration's conf block falls + // outside the seeded window from the poller's perspective. + chain := newFakeChain(t, 100, "below-100") + chain.extend("below-101") + chain.extend("below-102") + chain.extend("below-103") + chain.extend("below-104") + chain.extend("below-105") + + confBlock := chain.blocks[100] + srv := fakeChainServer(t, chain) + + rawTx := minimalRawTx() + txid := minimalRawTxID(t) + chain.setRawTx(txid, rawTx) + chain.setTxStatus(txid, esploraTxStatus{ + Confirmed: true, + BlockHeight: 100, + BlockHash: confBlock.hash.String(), + }) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + // historySize=3 means the poller only retains the last three + // canonical heights it has observed. The conf block at height + // 100 will be five heights below the seeded tip (105) and is + // guaranteed never to enter recentHashes. + tipPoller := NewTipPollerWithConfig( + esplora, 20*time.Millisecond, 3, btclog.Disabled, + ) + require.NoError(t, tipPoller.Start()) + t.Cleanup(tipPoller.Stop) + backend, err := NewChainBackendWithPoller( + esplora, tipPoller, btclog.Disabled, + ) + require.NoError(t, err) + require.NoError(t, backend.Start()) + t.Cleanup(func() { + require.NoError(t, backend.Stop()) + }) + + reg, err := backend.RegisterConf( + t.Context(), &txid, nil, 1, 99, false, + ) + require.NoError(t, err) + defer reg.Cancel() + + conf1 := awaitConf(t, reg.Confirmed) + require.Equal(t, confBlock.hash, *conf1.BlockHash) + + // Deep reorg: rewrite every height from 100 upward with + // different hashes. The new block at height 100 has a new + // hash, but the tx is still pinned there so the canonical + // re-query in reorgConfReg can fire a fresh Confirmed. + chain.rewriteFrom(100, "below-new") + newConfBlock := chain.blocks[100] + chain.setTxStatus(txid, esploraTxStatus{ + Confirmed: true, + BlockHeight: 100, + BlockHash: newConfBlock.hash.String(), + }) + + awaitSeqSignal(t, reg.Reorged, "conf Reorged on below-history reorg") + + conf2 := awaitConf(t, reg.Confirmed) + require.Equal(t, newConfBlock.hash, *conf2.BlockHash) + require.NotEqual( + t, conf1.BlockHash.String(), conf2.BlockHash.String(), + "re-confirmation must surface the new block hash", + ) +} + +// TestChainBackendAbortsOnRawHeaderFailure pins the property that a +// failed raw-header fetch during continuity check aborts the poll +// cycle rather than optimistically advancing. The optimistic path +// would permanently hide a reorg that crossed the old-tip boundary +// if the raw-header fetch flaked at exactly the wrong moment. +func TestChainBackendAbortsOnRawHeaderFailure(t *testing.T) { + t.Parallel() + + chain := newFakeChain(t, 100, "abort-init") + srv := fakeChainServer(t, chain) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + tipPoller := NewTipPoller( + esplora, 20*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, tipPoller.Start()) + t.Cleanup(tipPoller.Stop) + + tipSub, err := tipPoller.Subscribe() + require.NoError(t, err) + defer tipSub.Cancel() + reorgSub, err := tipPoller.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + // Add a height-101 block whose PrevBlock does NOT chain off + // the seeded tip at height 100. Then poison the raw-header + // endpoint for that block so the continuity check cannot + // resolve. The poller must abort the cycle: not broadcast + // the new block, not update its cached tip, not fire a reorg. + chain.mu.Lock() + stranger := chain.mintBlock( + 101, chainhash.Hash{0xde, 0xad}, "abort-stranger", + ) + chain.blocks[101] = stranger + chain.tip = 101 + chain.failRawHeader[stranger.hash] = struct{}{} + chain.mu.Unlock() + + // Sleep long enough for a few poll cycles to fire; verify + // neither tip advance nor reorg ever fires. + select { + case ev := <-tipSub.Updates(): + t.Fatalf("poller broadcast a tip event despite raw-header "+ + "failure: height=%d hash=%s", ev.Height, ev.Hash) + + case ev := <-reorgSub.Updates(): + t.Fatalf("poller broadcast a reorg event despite raw-header "+ + "failure: fork=%d", ev.ForkHeight) + + case <-time.After(200 * time.Millisecond): + // Expected: silent. + } + + height, hash, _ := tipPoller.BestBlock() + require.Equal( + t, int32(100), height, + "cached tip height advanced despite aborted cycle", + ) + require.Equal( + t, chain.blocks[100].hash, hash, + "cached tip hash advanced despite aborted cycle", + ) +} diff --git a/lwwallet/chain_backend_test.go b/lwwallet/chain_backend_test.go index bcea6be99..6ee39c5ad 100644 --- a/lwwallet/chain_backend_test.go +++ b/lwwallet/chain_backend_test.go @@ -2,6 +2,7 @@ package lwwallet import ( "bytes" + "encoding/hex" "encoding/json" "fmt" "net/http" @@ -80,15 +81,20 @@ func TestChainBackendBlockNotification(t *testing.T) { mu sync.Mutex tipHeight int32 = 100 blockHashes = make(map[int32]chainhash.Hash) + blockHdrs = make(map[int32]*wire.BlockHeader) ) - // Pre-generate block hashes. + // Pre-generate real chained block headers so the tip poller's + // PrevBlock continuity check during advance can resolve the raw + // header endpoint with a header that actually links back to the + // previous height. + var prevHash chainhash.Hash for h := int32(100); h <= 103; h++ { - blockHashes[h] = chainhash.HashH( - []byte( - fmt.Sprintf("block-%d", h), - ), - ) + hdr := mintStubHeader(h, prevHash) + blockHdrs[h] = hdr + hash := hdr.BlockHash() + blockHashes[h] = hash + prevHash = hash } srv := mockEsploraServer( @@ -109,7 +115,7 @@ func TestChainBackendBlockNotification(t *testing.T) { default: handleBlockReqs( - t, w, r, &mu, blockHashes, + t, w, r, &mu, blockHashes, blockHdrs, ) } }, @@ -147,10 +153,13 @@ func TestChainBackendBlockNotification(t *testing.T) { } } -// handleBlockReqs handles /block-height/:height and /block/:hash -// requests in the mock Esplora server. +// handleBlockReqs handles /block-height/:height, /block/:hash, and +// /block/:hash/header requests in the mock Esplora server. The +// blockHdrs map supplies real wire.BlockHeaders so the raw-header +// continuity check the tip poller runs on advance can succeed. func handleBlockReqs(t *testing.T, w http.ResponseWriter, r *http.Request, - mu *sync.Mutex, blockHashes map[int32]chainhash.Hash) { + mu *sync.Mutex, blockHashes map[int32]chainhash.Hash, + blockHdrs map[int32]*wire.BlockHeader) { t.Helper() @@ -176,20 +185,40 @@ func handleBlockReqs(t *testing.T, w http.ResponseWriter, r *http.Request, return } - // Handle /block/:hash (block header). - for _, h := range blockHashes { + // Handle /block/:hash and /block/:hash/header. + mu.Lock() + defer mu.Unlock() + for height, h := range blockHashes { hashStr := h.String() path := "/block/" + hashStr - if r.URL.Path == path { + headerPath := path + "/header" + + switch r.URL.Path { + case path: resp := esploraBlock{ ID: hashStr, - Height: 100, + Height: height, Timestamp: 1700000000, } - err := json.NewEncoder(w).Encode(resp) require.NoError(t, err) + return + + case headerPath: + hdr := blockHdrs[height] + if hdr == nil { + http.Error(w, "no header", + http.StatusNotFound) + + return + } + var buf bytes.Buffer + require.NoError(t, hdr.Serialize(&buf)) + _, _ = fmt.Fprint( + w, hex.EncodeToString(buf.Bytes()), + ) + return } } diff --git a/lwwallet/esplora_chain.go b/lwwallet/esplora_chain.go index bd37534ee..2d1c944fc 100644 --- a/lwwallet/esplora_chain.go +++ b/lwwallet/esplora_chain.go @@ -110,12 +110,15 @@ func NewEsploraChainService(esplora *EsploraClient, tipPoller *TipPoller, // 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. +// goroutine that translates each chain event into btcwallet chain +// notifications. The service subscribes to the unified chain stream +// so reorg and tip updates arrive on a single producer-ordered +// channel; this is what lets chain.BlockDisconnected reliably precede +// the replacement chain.BlockConnected events for the new tip. func (s *EsploraChainService) Start(ctx context.Context) error { //nolint:contextcheck // tip subscription lifecycle is owned by Stop - tipHeight, tipHash, tipTime, sub, err := - s.tipPoller.BestBlockAndSubscribe() + tipHeight, tipHash, tipTime, chainSub, err := + s.tipPoller.BestBlockAndSubscribeChain() if err != nil { return fmt.Errorf("subscribe to tip poller: %w", err) } @@ -134,7 +137,7 @@ func (s *EsploraChainService) Start(ctx context.Context) error { s.notifications <- chain.ClientConnected{} s.wg.Add(1) - go s.handleTipEvents(ctx, sub) + go s.handleChainEvents(ctx, chainSub) s.log.InfoS(ctx, "Esplora chain service started", slog.Int("tip_height", int(tipHeight)), @@ -685,26 +688,35 @@ func (s *EsploraChainService) MapRPCErr(err error) error { return err } -// 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) { +// handleChainEvents drains the unified chain stream and translates +// each event into btcwallet chain notifications. Using a single +// subscription delivers ReorgEvent and TipBlock updates in producer +// order through one channel, which is what guarantees +// BlockDisconnected lands on btcwallet's notification queue before +// the BlockConnected events for the replacement chain (btcwallet's +// disconnectBlock would otherwise refuse the rollback if the cached +// hash at that height had already been overwritten by a stale +// BlockConnected). +func (s *EsploraChainService) handleChainEvents(ctx context.Context, + sub *ChainSubscription) { defer s.wg.Done() defer sub.Cancel() for { select { - case event, ok := <-sub.Updates(): + case ev, ok := <-sub.Updates(): if !ok { return } - s.processTipEvent(ctx, event) + switch { + case ev.Reorg != nil: + s.processReorgEvent(ctx, ev.Reorg) + + case ev.Tip != nil: + s.processTipEvent(ctx, ev.Tip) + } case <-sub.Quit(): return @@ -729,6 +741,63 @@ func (s *EsploraChainService) handleTipEvents(ctx context.Context, // per-event cap branch without producing 256+ heights of traffic. const defaultMaxGapFillPerTipEvent int32 = 256 +// processReorgEvent translates a ReorgEvent into chain.BlockDisconnected +// notifications. The replacement Connected blocks arrive separately on +// the tip stream and are processed by processTipEvent in the usual way; +// emitting Disconnected here is what lets btcwallet roll back the +// reorged-out heights via its disconnectBlock path before it re-sees +// the canonical chain. Newest height is disconnected first so btcwallet +// walks back from its own cached tip. +// +// After emitting the disconnects, s.bestBlock is rolled back to +// event.ForkHeight so the subsequent replacement-chain TipBlock events +// (which arrive at heights forkHeight+1..newTipHeight) pass +// processTipEvent's "event.Height <= lastDelivered" duplicate guard. +// Without this rollback, the replacement BlockConnected events would +// silently be dropped because s.bestBlock still points at the old tip. +// The Hash is intentionally cleared because the reorg event does not +// carry the fork-point hash; the next BlockConnected fully repopulates +// s.bestBlock and only the Height is load-bearing for the duplicate +// guard. +func (s *EsploraChainService) processReorgEvent(ctx context.Context, + event *ReorgEvent) { + + if event == nil { + return + } + + for i := len(event.Disconnected) - 1; i >= 0; i-- { + hash := event.Disconnected[i] + height := event.ForkHeight + int32(i) + 1 + + meta := wtxmgr.BlockMeta{ + Block: wtxmgr.Block{ + Hash: hash, + Height: height, + }, + } + + select { + case s.notifications <- chain.BlockDisconnected(meta): + case <-s.quit: + return + } + } + + s.mu.Lock() + if s.bestBlock.Height > event.ForkHeight { + s.bestBlock = waddrmgr.BlockStamp{ + Height: event.ForkHeight, + } + } + s.mu.Unlock() + + s.log.InfoS(ctx, "Chain service emitted BlockDisconnected", + slog.Int("fork_height", int(event.ForkHeight)), + slog.Int("disconnected", len(event.Disconnected)), + ) +} + // processTipEvent applies one TipBlock to btcwallet's notification // channel. The chain service owns its own delivery cursor // (s.bestBlock); this function first walks any gap between diff --git a/lwwallet/esplora_chain_reorg_test.go b/lwwallet/esplora_chain_reorg_test.go new file mode 100644 index 000000000..a4b22c1c1 --- /dev/null +++ b/lwwallet/esplora_chain_reorg_test.go @@ -0,0 +1,175 @@ +package lwwallet + +import ( + "testing" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btclog/v2" + "github.com/btcsuite/btcwallet/chain" + "github.com/btcsuite/btcwallet/wtxmgr" + "github.com/stretchr/testify/require" +) + +// awaitNotification pulls the next notification from the chain +// service or fails the test on timeout. +func awaitNotification(t *testing.T, s *EsploraChainService) interface{} { + t.Helper() + + select { + case n := <-s.Notifications(): + return n + + case <-time.After(reorgTestTimeout): + t.Fatalf("timed out waiting for chain notification") + + return nil + } +} + +// drainUntilConnected drains notifications until a BlockConnected +// event at the given height is observed (or the test times out). +// Used to skip over the startup ClientConnected and initial connected +// events so the reorg-specific assertions can run on a known cursor. +func drainUntilConnected(t *testing.T, s *EsploraChainService, height int32) { + t.Helper() + + deadline := time.After(reorgTestTimeout) + + for { + select { + case n := <-s.Notifications(): + conn, ok := n.(chain.BlockConnected) + if !ok { + continue + } + if conn.Block.Height == height { + return + } + + case <-deadline: + t.Fatalf("timed out waiting for BlockConnected at %d", + height) + } + } +} + +// TestEsploraChainServiceReorgEmitsBlockDisconnected pins the property +// that a chain reorg surfaces chain.BlockDisconnected notifications +// before btcwallet sees the BlockConnected events that announce the +// new canonical chain, AND that the disconnect events arrive in +// newest-height-first order (the order btcwallet's disconnectBlock +// expects when walking back from its cached tip). Both ordering +// properties are load-bearing: without disconnect-before-connect, +// btcwallet would refuse the rollback because the cached hash at +// each height would already be overwritten; without +// newest-first, btcwallet's per-step rollback could trip on a +// height it hasn't seen disconnected yet. +func TestEsploraChainServiceReorgEmitsBlockDisconnected(t *testing.T) { + t.Parallel() + + chainModel := newFakeChain(t, 100, "svc-reorg-100") + chainModel.extend("svc-reorg-101") + chainModel.extend("svc-reorg-102") + + srv := fakeChainServer(t, chainModel) + + esplora := NewEsploraClient(srv.URL, btclog.Disabled) + tipPoller := NewTipPoller( + esplora, 20*time.Millisecond, btclog.Disabled, + ) + require.NoError(t, tipPoller.Start()) + t.Cleanup(tipPoller.Stop) + + service := NewEsploraChainService( + esplora, tipPoller, btclog.Disabled, + ) + require.NoError(t, service.Start(t.Context())) + t.Cleanup(func() { + service.Stop() + service.WaitForShutdown() + }) + + // First notification is always ClientConnected. + first := awaitNotification(t, service) + _, ok := first.(chain.ClientConnected) + require.True(t, ok, "expected ClientConnected first, got %T", first) + + // Drain forward to a known tip + a couple of new blocks so the + // service has a known set of recently-emitted connected blocks + // before the reorg fires. + chainModel.extend("svc-reorg-103") + drainUntilConnected(t, service, 103) + + // Stash hashes of the blocks the reorg is about to invalidate. + oldHash102 := chainModel.blocks[102].hash + oldHash103 := chainModel.blocks[103].hash + + // Rewrite heights 102..103 with a new chain. The tip stays at + // 103 but with a different hash, so the poller detects a + // same-height drift at 103 and walks back to fork point 101. + chainModel.rewriteFrom(102, "svc-reorg-new") + newHash102 := chainModel.blocks[102].hash + newHash103 := chainModel.blocks[103].hash + + // Drain notifications and pin the strict ordering. Each + // BlockConnected for a new-chain hash must arrive AFTER both + // BlockDisconnected events for the old chain — the unified + // chain stream guarantees this by serializing reorg + tip + // events through a single producer-ordered channel. + disconnectsSeen := 0 + gotDisconnected := []chainhash.Hash{} + gotConnectedNew := map[chainhash.Hash]struct{}{} + deadline := time.After(reorgTestTimeout) + for len(gotConnectedNew) < 2 { + select { + case n := <-service.Notifications(): + switch ev := n.(type) { + case chain.BlockDisconnected: + gotDisconnected = append( + gotDisconnected, ev.Block.Hash, + ) + disconnectsSeen++ + + case chain.BlockConnected: + meta := wtxmgr.BlockMeta(ev) + if meta.Hash != newHash102 && + meta.Hash != newHash103 { + + continue + } + + require.Equal( + t, 2, disconnectsSeen, "BlockConnect"+ + "ed for replacement chain "+ + "arrived before all "+ + "BlockDisconnected events: "+ + "saw %d disconnects so far", + disconnectsSeen, + ) + gotConnectedNew[meta.Hash] = struct{}{} + } + + case <-deadline: + t.Fatalf("timed out: disconnects=%v connected_new=%v", + gotDisconnected, gotConnectedNew) + } + } + + require.Equal( + t, oldHash103, gotDisconnected[0], "first "+ + "BlockDisconnected should be the newest old-chain "+ + "height (103) for btcwallet's per-step walk-back", + ) + require.Equal( + t, oldHash102, gotDisconnected[1], + "second BlockDisconnected should be height 102", + ) + + _, sawConn102 := gotConnectedNew[newHash102] + _, sawConn103 := gotConnectedNew[newHash103] + require.True(t, sawConn102, + "BlockConnected for new 102 not emitted") + require.True(t, sawConn103, + "BlockConnected for new 103 not emitted") +} diff --git a/lwwallet/esplora_chain_test.go b/lwwallet/esplora_chain_test.go index ee43b022f..151363cc0 100644 --- a/lwwallet/esplora_chain_test.go +++ b/lwwallet/esplora_chain_test.go @@ -2,6 +2,7 @@ package lwwallet import ( "bytes" + "encoding/hex" "encoding/json" "fmt" "net/http" @@ -299,6 +300,16 @@ func (c *rawBlockStubChain) serveBlockRoute(t *testing.T, w http.ResponseWriter, require.NoError(t, block.Serialize(&buf)) _, _ = w.Write(buf.Bytes()) + case "/header": + // Hex-encoded 80-byte block header. The TipPoller fetches + // this per new tip to verify PrevBlock continuity against + // the cached tip hash; without it the poller cannot + // distinguish a clean tip advance from a reorg crossing the + // boundary and aborts the cycle. + var buf bytes.Buffer + require.NoError(t, block.Header.Serialize(&buf)) + _, _ = fmt.Fprint(w, hex.EncodeToString(buf.Bytes())) + default: http.Error(w, "not implemented", http.StatusNotImplemented) diff --git a/lwwallet/tip_poller.go b/lwwallet/tip_poller.go index dbce6c97d..f2fd3557f 100644 --- a/lwwallet/tip_poller.go +++ b/lwwallet/tip_poller.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "sort" "sync" "time" @@ -11,6 +12,13 @@ import ( "github.com/btcsuite/btclog/v2" ) +// DefaultHashHistorySize is the default upper bound on entries retained +// in the TipPoller's bounded height -> hash history map. It is sized to +// at least twice the conventional Bitcoin reorg-safety depth (6) so a +// reorg at finality depth still has its disconnected hashes available +// for walk-back, with headroom. Configurable via NewTipPollerWithConfig. +const DefaultHashHistorySize = 100 + // 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 - @@ -38,6 +46,62 @@ type TipBlock struct { // for ergonomic call-site naming. type TipSubscription = Subscription[*TipBlock] +// ReorgEvent describes a chain reorganization observed by the +// TipPoller. The poller emits one ReorgEvent every time it detects +// that one or more previously broadcast blocks are no longer in the +// canonical chain. Disconnected hashes are listed in canonical +// (low-height first) order; Connected blocks are the new tip's path +// from ForkHeight+1 onward, also in canonical order. Either slice may +// be empty: a same-height hash-replacement reorg has Disconnected of +// length 1 and Connected of length 1. +type ReorgEvent struct { + // ForkHeight is the highest height at which the old chain and the + // new chain agree on the block hash. The first disconnected / + // connected block is at ForkHeight+1. + ForkHeight int32 + + // Disconnected lists block hashes that were previously broadcast + // as part of the canonical chain and are no longer on it, in + // ascending height order (ForkHeight+1 first). + Disconnected []chainhash.Hash + + // Connected lists blocks now on the canonical chain starting at + // ForkHeight+1, in ascending height order. The poller will also + // fan these out individually as TipBlock events on the standard + // Subscribe channel after the ReorgEvent is delivered. + Connected []*TipBlock +} + +// ReorgSubscription is the typed handle returned by +// TipPoller.SubscribeReorgs. +type ReorgSubscription = Subscription[*ReorgEvent] + +// ChainEvent is the unified update delivered on the ordered chain +// stream returned by SubscribeChain. Exactly one of Reorg and Tip is +// non-nil per event: +// +// - Reorg-only events announce a reorg's disconnected range and +// precede the replacement tip events on the same stream. +// - Tip-only events announce a new block (either a forward advance +// or a post-reorg connected block). +// +// Cross-event ordering on this stream is producer-ordered: the +// embedded EventServer delivers updates in SendUpdate order to a +// single subscriber goroutine, so a downstream consumer that +// dispatches on event type from one channel observes the same order +// the TipPoller emitted. That is the load-bearing property +// downstream needs to emit BlockDisconnected before BlockConnected +// for the replacement chain (btcwallet's disconnectBlock requires +// it; chainsource finality synthesis requires it too). +type ChainEvent struct { + Reorg *ReorgEvent + Tip *TipBlock +} + +// ChainSubscription is the typed handle returned by +// TipPoller.SubscribeChain. It carries the unified ChainEvent stream. +type ChainSubscription = Subscription[*ChainEvent] + // 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 @@ -58,18 +122,46 @@ type TipPoller struct { pollInterval time.Duration log btclog.Logger + // historySize caps the bounded height -> hash history map that + // lets the poller resolve old-chain hashes during reorg walk-back. + historySize int + // 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] + // reorgs is the typed event server that fans ReorgEvent updates + // out to subscribers that opt in via SubscribeReorgs. Reorgs are + // rare enough that a separate server is cheaper than re-fitting + // every TipBlock consumer with reorg-aware logic. + reorgs *EventServer[*ReorgEvent] + + // chain is the unified event server that fans both tip and reorg + // updates out on a single ordered stream. Consumers that need + // strict reorg-before-replacement-tip ordering (chain.Interface + // adapter, chainsource backend) subscribe here rather than to + // the separate events / reorgs servers, which would race across + // two independent translator goroutines. + chain *EventServer[*ChainEvent] + // mu guards the cached tip so BestBlock readers see a - // consistent height/hash/timestamp triple. + // consistent height/hash/timestamp triple. It also guards + // recentHashes, which is sized small enough that linear scans + // under the lock are cheap. mu sync.Mutex tipHeight int32 tipHash chainhash.Hash tipTime time.Time + // recentHashes maps a recent canonical-chain height to the hash + // the poller broadcast for it. It is bounded to historySize + // entries: on every insert we prune any entry whose height is + // more than historySize below the current tip. The buffer lets + // reorg detection walk back to the fork point using the cached + // hashes rather than re-fetching the entire pre-fork chain. + recentHashes map[int32]chainhash.Hash + // started gates re-entrant Start calls; the underlying // EventServer is also idempotent on Start. started bool @@ -82,15 +174,35 @@ type TipPoller struct { // 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. +// the cadence. The bounded hash-history map is sized to +// DefaultHashHistorySize. func NewTipPoller(esplora *EsploraClient, pollInterval time.Duration, log btclog.Logger) *TipPoller { + return NewTipPollerWithConfig( + esplora, pollInterval, DefaultHashHistorySize, log, + ) +} + +// NewTipPollerWithConfig is the explicit-history-size constructor. +// historySize <= 0 falls back to DefaultHashHistorySize so callers +// cannot accidentally disable reorg walk-back by passing a zero value. +func NewTipPollerWithConfig(esplora *EsploraClient, pollInterval time.Duration, + historySize int, log btclog.Logger) *TipPoller { + + if historySize <= 0 { + historySize = DefaultHashHistorySize + } + return &TipPoller{ esplora: esplora, pollInterval: pollInterval, + historySize: historySize, log: log, events: NewEventServer[*TipBlock](log), + reorgs: NewEventServer[*ReorgEvent](log), + chain: NewEventServer[*ChainEvent](log), + recentHashes: make(map[int32]chainhash.Hash), quit: make(chan struct{}), } } @@ -170,12 +282,60 @@ func (t *TipPoller) Start() error { return fmt.Errorf("start event server: %w", err) } + if err := t.reorgs.Start(); err != nil { + // Roll back the tip-event server so a partial Start does + // not leave a half-running poller behind. + _ = t.events.Stop() + resetStarted() + + return fmt.Errorf("start reorg event server: %w", err) + } + + if err := t.chain.Start(); err != nil { + _ = t.events.Stop() + _ = t.reorgs.Stop() + resetStarted() + + return fmt.Errorf("start chain event server: %w", err) + } + + // Seed recentHashes by walking back historySize-1 heights so a + // reorg whose disconnected range extends below the seeded tip + // but within the configured history can still resolve every + // disconnected hash from the cache. Without this, a fresh + // poller would only ever cache the initial tip, leaving a + // downstream chain.Interface consumer unable to enumerate every + // hash btcwallet must roll back on a multi-block reorg below + // the seeded tip. The walk-back is best-effort: a single + // per-height fetch failure ends the seed loop early; the next + // poll tick still drives the cache forward as the chain grows. t.mu.Lock() t.tipHeight = height t.tipHash = hash t.tipTime = tipTime + t.recordHashLocked(height, hash) t.mu.Unlock() + for h := height - 1; h > height-int32(t.historySize) && h >= 0; h-- { + liveHash, err := t.esplora.GetBlockHashByHeight( + context.Background(), h, + ) + if err != nil { + t.log.WarnS( + context.Background(), + "Tip poller history seed fetch failed", + err, + slog.Int("height", int(h)), + ) + + break + } + + t.mu.Lock() + t.recordHashLocked(h, liveHash) + t.mu.Unlock() + } + t.wg.Add(1) go t.pollLoop() @@ -197,7 +357,7 @@ func (t *TipPoller) Stop() { t.wg.Wait() - // Stop the event server after the poll loop has exited so + // Stop the event servers 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 { @@ -207,6 +367,22 @@ func (t *TipPoller) Stop() { err, ) } + + if err := t.reorgs.Stop(); err != nil { + t.log.WarnS( + context.Background(), + "Tip poller reorg event server stop returned error", + err, + ) + } + + if err := t.chain.Stop(); err != nil { + t.log.WarnS( + context.Background(), + "Tip poller chain event server stop returned error", + err, + ) + } } // BestBlock returns a snapshot of the currently cached tip. Callers @@ -233,6 +409,18 @@ func (t *TipPoller) Subscribe() (*TipSubscription, error) { return t.events.Subscribe() } +// SubscribeReorgs returns a typed subscription that receives a +// ReorgEvent every time the poller detects that one or more +// previously broadcast blocks have been replaced on the canonical +// chain. Callers that need both tip events and reorg events should +// subscribe to both streams; reorg events are delivered BEFORE the +// connected blocks are fanned out on the TipBlock stream so that +// consumers can mark their registrations dirty before the +// re-confirmation arrives. +func (t *TipPoller) SubscribeReorgs() (*ReorgSubscription, error) { + return t.reorgs.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 @@ -257,6 +445,85 @@ func (t *TipPoller) BestBlockAndSubscribe() (int32, chainhash.Hash, time.Time, return t.tipHeight, t.tipHash, t.tipTime, sub, nil } +// BestBlockAndSubscribeAll atomically reads the cached tip and +// registers both a TipBlock subscription and a ReorgEvent +// subscription. Reorg-aware consumers (e.g. ChainBackend) use this +// to avoid the small window where a reorg could land between +// independently registering on the two streams. +func (t *TipPoller) BestBlockAndSubscribeAll() (int32, chainhash.Hash, + time.Time, *TipSubscription, *ReorgSubscription, error) { + + t.mu.Lock() + defer t.mu.Unlock() + + sub, err := t.events.Subscribe() + if err != nil { + return 0, chainhash.Hash{}, time.Time{}, nil, nil, + fmt.Errorf("subscribe to tip events: %w", err) + } + + reorgSub, err := t.reorgs.Subscribe() + if err != nil { + sub.Cancel() + + return 0, chainhash.Hash{}, time.Time{}, nil, nil, + fmt.Errorf("subscribe to reorg events: %w", err) + } + + return t.tipHeight, t.tipHash, t.tipTime, sub, reorgSub, nil +} + +// SubscribeChain returns a typed subscription on the unified chain +// event stream. Updates arrive in producer order through a single +// channel; consumers that need strict reorg-before-replacement-tip +// ordering must subscribe here rather than to the separate tip / +// reorg streams (which race across two independent translator +// goroutines). +func (t *TipPoller) SubscribeChain() (*ChainSubscription, error) { + return t.chain.Subscribe() +} + +// BestBlockAndSubscribeChain atomically reads the cached tip and +// registers a ChainSubscription. Same atomicity guarantee as +// BestBlockAndSubscribe, applied to the unified stream. +func (t *TipPoller) BestBlockAndSubscribeChain() (int32, chainhash.Hash, + time.Time, *ChainSubscription, error) { + + t.mu.Lock() + defer t.mu.Unlock() + + sub, err := t.chain.Subscribe() + if err != nil { + return 0, chainhash.Hash{}, time.Time{}, nil, + fmt.Errorf("subscribe to chain events: %w", err) + } + + return t.tipHeight, t.tipHash, t.tipTime, sub, nil +} + +// recordHashLocked inserts a (height, hash) pair into the recent- +// hash history map and prunes any entry whose height has fallen out +// of the historySize window relative to the new entry. Caller must +// hold t.mu. +func (t *TipPoller) recordHashLocked(height int32, hash chainhash.Hash) { + t.recentHashes[height] = hash + + cutoff := height - int32(t.historySize) + for h := range t.recentHashes { + if h <= cutoff { + delete(t.recentHashes, h) + } + } +} + +// hashAtHeightLocked returns the cached hash for a height plus +// whether it was present. Caller must hold t.mu. +func (t *TipPoller) hashAtHeightLocked(height int32) (chainhash.Hash, bool) { + h, ok := t.recentHashes[height] + + return h, ok +} + // 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 @@ -278,12 +545,26 @@ func (t *TipPoller) pollLoop() { } } -// 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. +// poll performs one tip-detection cycle. The cycle is reorg-aware: +// +// 1. Query (tip height, tip hash). +// 2. If the tip is at the same height as the cached tip but the hash +// differs, walk back through cached hashes to find the fork point +// and emit a ReorgEvent + a re-broadcast TipBlock at the same +// height. This is the "same-height reorg" case that earlier +// versions of the poller could not detect. +// 3. If the new height is strictly greater than the cached height, +// fetch each intervening hash; if the first new block's PrevBlock +// header field does not point at the cached tip hash, walk back +// to find the fork point, emit a ReorgEvent for the disconnected +// range, and then fan TipBlock events for the new chain in +// ascending order. If the new chain extends the old tip cleanly +// (the common case), no ReorgEvent fires and TipBlock events are +// dispatched as before. +// +// 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(context.Background()) if err != nil { @@ -298,23 +579,123 @@ func (t *TipPoller) poll() { t.mu.Lock() oldHeight := t.tipHeight + oldHash := t.tipHash 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 { + switch { + case newHeight < oldHeight: + // The remote reports fewer blocks than we have cached. + // Two distinct cases hide behind a lower height: + // + // 1. A transient indexer hiccup where the remote + // momentarily lags our cached tip but is still on + // the same chain (our tip remains canonical). + // 2. A genuine reorg onto a SHORTER but higher-work + // chain: blocks above newHeight were orphaned and + // the surviving chain has fewer total blocks. Bitcoin + // follows most-work, not most-blocks, so a shorter + // chain can legitimately win. + // + // Disambiguate by comparing the live hash at newHeight to + // our cached hash for that height. Agreement means the + // remote is merely behind (case 1, no-op); divergence + // means the chain reorged out from under us (case 2) and + // we must walk back to the fork point. Without this an + // orphaning reorg to a shorter chain would be silently + // ignored until the new chain grew past our stale tip. + newTipHash, err := t.esplora.GetBlockHashByHeight( + context.Background(), newHeight, + ) + if err != nil { + t.log.WarnS( + context.Background(), + "Tip poller shorter-chain hash check failed", + err, + slog.Int("height", int(newHeight)), + ) + + return + } + + t.mu.Lock() + cachedHash, haveCached := t.hashAtHeightLocked(newHeight) + t.mu.Unlock() + + switch { + // newHeight predates our retained history floor, so we + // cannot prove a reorg by hash comparison. A reorg this + // deep exceeds historySize and is an operator-level + // problem rather than a productionally recoverable one + // (mirrors handleReorg's own deep-reorg guard); log and + // re-check next tick rather than walk a pruned history. + case !haveCached: + t.log.WarnS( + context.Background(), + "Tip poller remote tip below retained "+ + "history floor; ignoring", + fmt.Errorf("history floor exceeded"), + slog.Int("new_height", int(newHeight)), + slog.Int("old_height", int(oldHeight)), + ) + + return + + // Same hash at newHeight: the remote is merely lagging + // our cached tip on the same chain. No-op. + case newTipHash == cachedHash: + return + } + + // Divergent hash at newHeight: a reorg onto a shorter, + // higher-work chain. handleReorg builds Disconnected up + // to our old cached tip (covering the orphaned blocks + // above newHeight) and Connected up to newHeight. + t.handleReorg(newHeight, newTipHash) + + return + + case newHeight == oldHeight: + // Same-height: query the hash at this height (which + // the BlockHashByHeight endpoint resolves directly). + // If it differs from the cached hash at the same + // height we have a same-height reorg; if it matches + // the chain has not moved. + newTipHash, err := t.esplora.GetBlockHashByHeight( + context.Background(), newHeight, + ) + if err != nil { + t.log.WarnS( + context.Background(), + "Tip poller same-height hash check failed", + err, + slog.Int("height", int(newHeight)), + ) + + return + } + + if newTipHash == oldHash { + return + } + + t.handleReorg(newHeight, newTipHash) + return + + default: + // newHeight > oldHeight: forward advance, possibly with + // a deeper reorg if the first new height's predecessor + // is not the cached tip hash. + t.advance(oldHeight, newHeight) } +} +// advance walks the chain forward from oldHeight+1 to newHeight, +// detecting reorgs along the way. The very first new height's raw +// header is consulted to verify its PrevBlock matches the cached +// tip hash; mismatch triggers a walk-back to the fork point and a +// reorg emission before any new TipBlock events fire. +func (t *TipPoller) advance(oldHeight, newHeight int32) { t.log.DebugS(context.Background(), "Tip poller advancing", slog.Int("old_height", int(oldHeight)), slog.Int("new_height", int(newHeight)), @@ -349,44 +730,397 @@ func (t *TipPoller) poll() { return } + // On the FIRST new height, verify continuity against + // the cached tip. We use the raw 80-byte header (which + // carries PrevBlock) because the JSON header endpoint + // does not include the previous-block hash. If + // PrevBlock does not match the cached tip hash a reorg + // crossed the boundary; walk back to the fork point + // and emit a ReorgEvent before continuing. + if height == oldHeight+1 { + rawHdr, err := t.esplora.GetRawBlockHeader( + context.Background(), hash, + ) + if err != nil { + // Without the raw header we cannot verify + // that the new block connects to the cached + // tip. Abort the cycle without broadcasting + // or updating the cached tip; the next tick + // will retry. Optimistically advancing here + // would permanently hide a reorg that + // crossed the oldHeight boundary if the + // raw-header fetch flaked exactly when the + // reorg happened. + t.log.WarnS( + context.Background(), + "Tip poller raw header fetch "+ + "failed; aborting cycle", + err, + slog.String("hash", hash.String()), + ) + + return + } + + t.mu.Lock() + cachedOldHash, ok := t.hashAtHeightLocked( + oldHeight, + ) + t.mu.Unlock() + + if ok && rawHdr.PrevBlock != cachedOldHash { + // Deeper reorg: handle it as a new-tip + // reorg from oldHeight, pointing at the + // actual new tip at newHeight. handleReorg + // emits ReorgEvent + TipBlock events + // itself. + tipHash, tipErr := t.esplora. + GetBlockHashByHeight( + context.Background(), newHeight, + ) + if tipErr != nil { + t.log.WarnS( + context.Background(), + "Tip poller "+ + "reorg tip "+ + "fetch failed", + tipErr, + slog.Int( + "height", + int(newHeight), + ), + ) + + return + } + + t.handleReorg(newHeight, tipHash) + + 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. + if !t.broadcastTipBlock(event) { + return + } + } +} + +// broadcastTipBlock updates the cached tip, records the hash in the +// history map, and fans the TipBlock out to subscribers. Returns +// false if the send failed (server shutting down), telling the +// caller to abort the cycle so the cached tip is not advanced past +// an event subscribers did not receive. +func (t *TipPoller) broadcastTipBlock(event *TipBlock) bool { + // 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. + t.mu.Lock() + t.tipHeight = event.Height + t.tipHash = event.Hash + if event.Header != nil { + t.tipTime = time.Unix(event.Header.Timestamp, 0) + } + t.recordHashLocked(event.Height, event.Hash) + tipErr := t.events.SendUpdate(event) + chainErr := t.chain.SendUpdate(&ChainEvent{Tip: event}) + t.mu.Unlock() + + if tipErr != nil { + t.log.WarnS( + context.Background(), + "Tip poller send update failed", + tipErr, + slog.Int("height", int(event.Height)), + ) + + return false + } + + if chainErr != nil { + t.log.WarnS( + context.Background(), + "Tip poller chain stream send failed", + chainErr, + slog.Int("height", int(event.Height)), + ) + + return false + } + + return true +} + +// handleReorg walks back through the recent-hash history map until it +// finds the fork point between the cached chain and the new chain +// whose tip is (newTipHeight, newTipHash). It then constructs a +// ReorgEvent listing the disconnected hashes and connected blocks, +// broadcasts the reorg, and finally fans out the connected blocks +// as TipBlock events in ascending height order so consumers can +// re-check registrations against each new block. +func (t *TipPoller) handleReorg(newTipHeight int32, newTipHash chainhash.Hash) { + t.log.InfoS(context.Background(), "Tip poller detected reorg", + slog.Int("new_tip_height", int(newTipHeight)), + slog.String("new_tip_hash", newTipHash.String()), + ) + + // Walk back to find the fork point. We probe heights from + // newTipHeight downwards, comparing the live Esplora hash at + // each height to the cached hash. The first height at which + // they agree is the fork point. We bound the search by the + // retained history depth so a misbehaving Esplora cannot drag + // us into an unbounded loop. + t.mu.Lock() + cachedTipHeight := t.tipHeight + cutoff := cachedTipHeight - int32(t.historySize) + t.mu.Unlock() + + if cutoff < 0 { + cutoff = 0 + } + + // connectedByHeight collects new-chain blocks discovered + // during walk-back so we can re-broadcast them in ascending + // order after the fork point is found. + connectedByHeight := make(map[int32]chainhash.Hash) + connectedByHeight[newTipHeight] = newTipHash + + forkHeight := int32(-1) + probeHeight := newTipHeight + for probeHeight > cutoff { + probeHeight-- + + // Heights strictly above our cached tip cannot be a fork + // point: we never broadcast a block there, so the live + // block is a NEW connected block on a longer chain (a + // forward reorg that both reorganized old blocks AND + // extended past our tip). Record it and keep walking + // down toward the real fork. Without this, the + // !haveCached branch below would mistake the first such + // height for the fork point and compute a forkHeight + // above the cached tip, underflowing the Disconnected + // slice capacity. + if probeHeight > cachedTipHeight { + liveHash, err := t.esplora.GetBlockHashByHeight( + context.Background(), probeHeight, + ) + if err != nil { + t.log.WarnS( + context.Background(), + "Tip poller reorg walk-back failed", + err, + slog.Int("height", int(probeHeight)), + ) + + return + } + + connectedByHeight[probeHeight] = liveHash + + continue + } + t.mu.Lock() - t.tipHeight = height - t.tipHash = hash - t.tipTime = time.Unix(header.Timestamp, 0) - sendErr := t.events.SendUpdate(event) + cachedHash, haveCached := t.hashAtHeightLocked(probeHeight) 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 { + if !haveCached { + // We never broadcast a block at this height + // (typically because it's older than our + // retained history's start, e.g. on a fresh + // poller that only ever cached the initial + // tip). Treat probeHeight as the fork point: + // anything at or below it is not part of our + // old broadcast set, so it cannot be + // "disconnected" from a downstream consumer's + // point of view. The live hash at this height + // is not a new connected block we owe + // subscribers either; only blocks strictly + // above probeHeight that we previously + // broadcast (and that have now changed) form + // the reorg boundary. + forkHeight = probeHeight + + break + } + + liveHash, err := t.esplora.GetBlockHashByHeight( + context.Background(), probeHeight, + ) + if err != nil { t.log.WarnS( context.Background(), - "Tip poller send update failed", - sendErr, - slog.Int("height", int(height)), + "Tip poller reorg walk-back failed", + err, + slog.Int("height", int(probeHeight)), + ) + + return + } + + if liveHash == cachedHash { + // Genuine fork point: the new chain agrees with + // our cached canonical chain at this height. + forkHeight = probeHeight + + break + } + + // haveCached && liveHash != cachedHash: this height is + // part of the reorg. Record the live hash so we can + // rebroadcast it after the fork point is found. + connectedByHeight[probeHeight] = liveHash + } + + if forkHeight < 0 { + // We exhausted the retained history without finding a + // fork point. Be conservative: log and bail. A future + // tick will re-attempt with the now-pruned history; in + // practice a reorg deeper than DefaultHashHistorySize + // is a developer / operator problem, not a + // productionally recoverable condition. + t.log.WarnS( + context.Background(), + "Tip poller reorg deeper than retained history; "+ + "giving up walk-back", + fmt.Errorf("history exhausted"), + slog.Int("new_tip_height", int(newTipHeight)), + slog.Int("history_size", t.historySize), + ) + + return + } + + // Build the Disconnected slice from cached hashes between + // forkHeight+1 and the cached tipHeight in one critical + // section. The walk is already in ascending order so no sort + // is needed; we cap at the history size to bound the worst + // case under t.mu. + t.mu.Lock() + cachedTipHeight = t.tipHeight + disconnected := make( + []chainhash.Hash, 0, int(cachedTipHeight-forkHeight), + ) + for h := forkHeight + 1; h <= cachedTipHeight; h++ { + if hash, ok := t.hashAtHeightLocked(h); ok { + disconnected = append(disconnected, hash) + } + } + t.mu.Unlock() + + // Build the Connected slice. Heights range from forkHeight+1 + // to newTipHeight; for each, fetch the header so the + // TipBlock carries a populated Header field for downstream + // consumers. + connectedHeights := make([]int32, 0, len(connectedByHeight)) + for h := range connectedByHeight { + if h <= forkHeight { + continue + } + connectedHeights = append(connectedHeights, h) + } + sort.Slice(connectedHeights, func(i, j int) bool { + return connectedHeights[i] < connectedHeights[j] + }) + + connected := make([]*TipBlock, 0, len(connectedHeights)) + for _, h := range connectedHeights { + hash := connectedByHeight[h] + header, err := t.esplora.GetBlockHeader( + context.Background(), hash, + ) + if err != nil { + t.log.WarnS( + context.Background(), + "Tip poller reorg header fetch failed", + err, + slog.String("hash", hash.String()), ) return } + + connected = append(connected, &TipBlock{ + Height: h, + Hash: hash, + Header: header, + }) + } + + // Prune the now-stale hashes from the history before we + // broadcast the reorg so a subscriber that immediately calls + // back into BestBlock sees the new tip. + t.mu.Lock() + for h := forkHeight + 1; h <= cachedTipHeight; h++ { + delete(t.recentHashes, h) + } + t.mu.Unlock() + + reorgEvent := &ReorgEvent{ + ForkHeight: forkHeight, + Disconnected: disconnected, + Connected: connected, + } + + t.log.InfoS(context.Background(), "Tip poller emitting reorg", + slog.Int("fork_height", int(forkHeight)), + slog.Int("disconnected", len(disconnected)), + slog.Int("connected", len(connected)), + ) + + if err := t.reorgs.SendUpdate(reorgEvent); err != nil { + t.log.WarnS( + context.Background(), + "Tip poller reorg send failed", + err, + ) + + return + } + + // Also emit the reorg on the unified chain stream BEFORE any + // connected-block tip events land on it, so a single consumer + // reading the chain subscription sees Reorged before any of + // the replacement Connected blocks. This is the load-bearing + // ordering property the chain.Interface adapter needs to emit + // BlockDisconnected before BlockConnected, and the chainsource + // backend needs to reset registrations before block-epoch + // driven re-checks run on the replacement chain. + if err := t.chain.SendUpdate( + &ChainEvent{Reorg: reorgEvent}, + ); err != nil { + + t.log.WarnS( + context.Background(), + "Tip poller chain reorg send failed", + err, + ) + + return + } + + // Finally, fan the connected blocks out on the standard + // TipBlock stream so existing consumers re-check on each + // new block exactly as they would on a non-reorg advance. + // broadcastTipBlock also pushes each block onto the unified + // chain stream so the single-channel consumer observes the + // connected blocks immediately after the reorg event in the + // same producer-ordered sequence. + for _, block := range connected { + if !t.broadcastTipBlock(block) { + return + } } } diff --git a/lwwallet/tip_poller_reorg_test.go b/lwwallet/tip_poller_reorg_test.go new file mode 100644 index 000000000..0b9dd2cbc --- /dev/null +++ b/lwwallet/tip_poller_reorg_test.go @@ -0,0 +1,274 @@ +package lwwallet + +import ( + "fmt" + "testing" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/btcsuite/btclog/v2" + "github.com/stretchr/testify/require" +) + +// mintStubHeaderGen is like mintStubHeader but mixes a fork generation +// into the salt so re-minting the same height on a forked chain yields +// a distinct BlockHash. This lets a test rebuild a height range on a +// different chain whose hashes provably diverge from the original. +func mintStubHeaderGen(height, gen int32, + prev chainhash.Hash) *wire.BlockHeader { + + salt := chainhash.HashH( + fmt.Appendf(nil, "stub-block-%d-gen-%d", height, gen), + ) + + return &wire.BlockHeader{ + Version: 1, + PrevBlock: prev, + MerkleRoot: chainhash.HashH( + fmt.Appendf(nil, "merkle-%d-gen-%d", height, gen), + ), + Timestamp: time.Unix(int64(height)*600, 0), + Bits: 0x207fffff, + Nonce: uint32(salt[0])<<24 | uint32(salt[1])<<16 | + uint32(salt[2])<<8 | uint32(salt[3]), + } +} + +// reorgTo rewrites the chain onto a fork that branches at forkHeight and +// extends to newTip. newTip may be lower than the current tip (a shorter +// but higher-work chain), equal to it (a same-height hash replacement), +// or higher (a deeper forward reorg). Heights above forkHeight are +// replaced with generation-salted blocks whose hashes diverge from the +// old chain; any old heights above newTip are orphaned (removed) so the +// stubbed backend no longer serves them. +func (c *stubChain) reorgTo(t *testing.T, forkHeight, newTip, gen int32) { + t.Helper() + + c.mu.Lock() + defer c.mu.Unlock() + + // Drop every height above the fork point; the new chain rebuilds + // forkHeight+1 .. newTip and orphans anything beyond newTip. + for h := range c.hashAt { + if h > forkHeight { + delete(c.hashAt, h) + delete(c.blocks, h) + } + } + + prev := c.hashAt[forkHeight] + for h := forkHeight + 1; h <= newTip; h++ { + hdr := mintStubHeaderGen(h, gen, prev) + c.blocks[h] = hdr + hash := hdr.BlockHash() + c.hashAt[h] = hash + prev = hash + } + + c.tipHeight = newTip +} + +// setTipHeight lowers (or raises) the reported tip height WITHOUT +// touching any cached hashes. It models a transient indexer hiccup +// where the remote momentarily reports fewer blocks than it served a +// moment ago, but is still on the same chain. +func (c *stubChain) setTipHeight(h int32) { + c.mu.Lock() + defer c.mu.Unlock() + + c.tipHeight = h +} + +// requireTipEventually blocks until the poller's BestBlock height +// reaches want, failing the test if it does not within the deadline. +func requireTipEventually(t *testing.T, tp *TipPoller, want int32) { + t.Helper() + + require.Eventually(t, func() bool { + h, _, _ := tp.BestBlock() + + return h == want + }, 2*time.Second, 5*time.Millisecond, + "poller never reached tip height %d", want) +} + +// newReorgTestPoller spins up a TipPoller over a stubChain seeded at the +// given height and returns both so a test can drive reorgs. +func newReorgTestPoller(t *testing.T, + seedHeight int32) (*TipPoller, *stubChain) { + + t.Helper() + + chain := newStubChain(seedHeight) + srv := mockEsploraServer(t, stubEsploraHandler(t, chain)) + + tp := NewTipPoller( + NewEsploraClient(srv.URL, btclog.Disabled), 10*time.Millisecond, + btclog.Disabled, + ) + require.NoError(t, tp.Start()) + t.Cleanup(tp.Stop) + + return tp, chain +} + +// TestTipPollerShorterChainReorg covers the case Roasbeef flagged: a +// reorg onto a SHORTER but higher-work chain, where blocks above the new +// tip are orphaned. The poller must detect this even though the remote +// reports a lower height than the cached tip, and roll the tip back to +// the shorter chain's tip. +func TestTipPollerShorterChainReorg(t *testing.T) { + t.Parallel() + + tp, chain := newReorgTestPoller(t, 100) + + reorgSub, err := tp.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + // Grow to 110 so the poller caches 101..110, then confirm it + // observed the advance before we reorg out from under it. + chain.advance(t, 10) + requireTipEventually(t, tp, 110) + + // Fork at 105 onto a shorter chain that tips at 108 (< 110). + chain.reorgTo(t, 105, 108, 1) + + select { + case ev := <-reorgSub.Updates(): + require.NotNil(t, ev) + require.Equal(t, int32(105), ev.ForkHeight) + + // Old chain 106..110 disconnects (5 blocks), even though + // the new chain only carries 106..108. + require.Len(t, ev.Disconnected, 5) + + // New chain connects 106..108 (3 blocks) in ascending + // order, tipping at 108. + require.Len(t, ev.Connected, 3) + require.Equal( + t, int32(106), ev.Connected[0].Height, + ) + require.Equal( + t, int32(108), ev.Connected[len(ev.Connected)-1].Height, + ) + + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for shorter-chain reorg event") + } + + // The poller's tip must now be the shorter chain's tip. + requireTipEventually(t, tp, 108) +} + +// TestTipPollerShorterHeightSameChainNoReorg verifies the no-op half of +// the shorter-height branch: a transient indexer hiccup where the remote +// reports fewer blocks but is still on the same chain (the hash at the +// reported height matches our cache). No reorg must fire and the cached +// tip must NOT roll back. +func TestTipPollerShorterHeightSameChainNoReorg(t *testing.T) { + t.Parallel() + + tp, chain := newReorgTestPoller(t, 100) + + reorgSub, err := tp.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + chain.advance(t, 10) + requireTipEventually(t, tp, 110) + + // Remote briefly reports height 108 with the SAME hash history + // (no fork). This is lag, not a reorg. + chain.setTipHeight(108) + + select { + case ev := <-reorgSub.Updates(): + t.Fatalf("unexpected reorg on transient lag: %+v", ev) + + case <-time.After(300 * time.Millisecond): + // No reorg fired: correct. + } + + // The tip must remain at 110: we never roll back on a lagging + // remote that is still on our chain. + h, _, _ := tp.BestBlock() + require.Equal(t, int32(110), h) +} + +// TestTipPollerSameHeightReorg covers a same-height hash replacement: the +// tip height does not change but the block at the tip is replaced by a +// different block on a competing chain. +func TestTipPollerSameHeightReorg(t *testing.T) { + t.Parallel() + + tp, chain := newReorgTestPoller(t, 100) + + reorgSub, err := tp.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + chain.advance(t, 10) + requireTipEventually(t, tp, 110) + + oldHash := chain.hashAt[110] + + // Replace block 110 with a different block at the same height. + chain.reorgTo(t, 109, 110, 1) + + select { + case ev := <-reorgSub.Updates(): + require.NotNil(t, ev) + require.Equal(t, int32(109), ev.ForkHeight) + require.Len(t, ev.Disconnected, 1) + require.Equal(t, oldHash, ev.Disconnected[0]) + require.Len(t, ev.Connected, 1) + require.Equal(t, int32(110), ev.Connected[0].Height) + + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for same-height reorg event") + } + + requireTipEventually(t, tp, 110) +} + +// TestTipPollerDeeperForwardReorg covers a reorg that also extends the +// chain: the new chain forks below the old tip yet ends higher than it. +// The forward-advance path must notice the broken PrevBlock continuity +// at the old tip boundary and emit a reorg before the new tip events. +func TestTipPollerDeeperForwardReorg(t *testing.T) { + t.Parallel() + + tp, chain := newReorgTestPoller(t, 100) + + reorgSub, err := tp.SubscribeReorgs() + require.NoError(t, err) + defer reorgSub.Cancel() + + chain.advance(t, 10) + requireTipEventually(t, tp, 110) + + // Fork at 107 onto a longer chain that tips at 113. + chain.reorgTo(t, 107, 113, 1) + + select { + case ev := <-reorgSub.Updates(): + require.NotNil(t, ev) + require.Equal(t, int32(107), ev.ForkHeight) + + // Old chain 108..110 disconnects (3 blocks). + require.Len(t, ev.Disconnected, 3) + + // New chain connects 108..113 (6 blocks). + require.Len(t, ev.Connected, 6) + require.Equal( + t, int32(113), ev.Connected[len(ev.Connected)-1].Height, + ) + + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for deeper forward reorg event") + } + + requireTipEventually(t, tp, 113) +} diff --git a/lwwallet/tip_poller_test.go b/lwwallet/tip_poller_test.go index e4aed4a9c..cf75fa16b 100644 --- a/lwwallet/tip_poller_test.go +++ b/lwwallet/tip_poller_test.go @@ -1,6 +1,8 @@ package lwwallet import ( + "bytes" + "encoding/hex" "fmt" "net/http" "sync" @@ -9,6 +11,7 @@ import ( "time" "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" "github.com/btcsuite/btclog/v2" "github.com/stretchr/testify/require" ) @@ -16,28 +19,59 @@ import ( // 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. +// tests can drive the chain forward synchronously. Each height holds +// a real wire.BlockHeader whose PrevBlock chains back to the previous +// height; the poller's PrevBlock continuity check on advance therefore +// passes naturally rather than aborting. 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. + // blocks[h] holds the wire.BlockHeader for height h. Pre- + // populated through 0..tipHeight at construction and grown + // monotonically by advance. + blocks map[int32]*wire.BlockHeader + + // hashAt[h] is the chainhash.Hash for height h. hashAt map[int32]chainhash.Hash } +// mintStubHeader builds a deterministic wire.BlockHeader for a given +// height, chaining off the supplied previous hash. The salted nonce +// makes the resulting BlockHash stable per (height, prev) and unique +// across heights. +func mintStubHeader(height int32, prev chainhash.Hash) *wire.BlockHeader { + salt := chainhash.HashH([]byte(fmt.Sprintf("stub-block-%d", height))) + + return &wire.BlockHeader{ + Version: 1, + PrevBlock: prev, + MerkleRoot: chainhash.HashH( + []byte( + fmt.Sprintf("merkle-%d", height), + ), + ), + Timestamp: time.Unix(int64(height)*600, 0), + Bits: 0x207fffff, + Nonce: uint32(salt[0])<<24 | uint32(salt[1])<<16 | + uint32(salt[2])<<8 | uint32(salt[3]), + } +} + func newStubChain(tipHeight int32) *stubChain { c := &stubChain{ tipHeight: tipHeight, + blocks: make(map[int32]*wire.BlockHeader), hashAt: make(map[int32]chainhash.Hash), } + var prev chainhash.Hash for h := int32(0); h <= tipHeight; h++ { - c.hashAt[h] = chainhash.HashH( - []byte( - fmt.Sprintf("block-%d", h), - ), - ) + hdr := mintStubHeader(h, prev) + c.blocks[h] = hdr + hash := hdr.BlockHash() + c.hashAt[h] = hash + prev = hash } return c @@ -49,13 +83,14 @@ func (c *stubChain) advance(t *testing.T, n int32) { c.mu.Lock() defer c.mu.Unlock() + prev := c.hashAt[c.tipHeight] for i := int32(1); i <= n; i++ { h := c.tipHeight + i - c.hashAt[h] = chainhash.HashH( - []byte( - fmt.Sprintf("block-%d", h), - ), - ) + hdr := mintStubHeader(h, prev) + c.blocks[h] = hdr + hash := hdr.BlockHash() + c.hashAt[h] = hash + prev = hash } c.tipHeight += n @@ -147,19 +182,29 @@ func stubEsploraHandler(t *testing.T, chain *stubChain) http.HandlerFunc { h.String(), height, int64(height)*600) + case "/header": + // Raw 80-byte header, hex-encoded. + chain.mu.Lock() + hdr := chain.blocks[height] + chain.mu.Unlock() + if hdr == nil { + http.Error( + w, "not found", + http.StatusNotFound, + ) + + return + } + var buf bytes.Buffer + require.NoError(t, hdr.Serialize(&buf)) + _, _ = fmt.Fprint( + w, + hex.EncodeToString( + buf.Bytes(), + ), + ) + 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,