diff --git a/btcwbackend/chain_backend.go b/btcwbackend/chain_backend.go index 4817d880d..4cd8de1e6 100644 --- a/btcwbackend/chain_backend.go +++ b/btcwbackend/chain_backend.go @@ -38,8 +38,13 @@ type ChainBackend struct { neutrinoCS *neutrino.ChainService // notifier provides chain notification services backed by - // neutrino's compact block filter scanning. - notifier *neutrinonotify.NeutrinoNotifier + // neutrino's compact block filter scanning. Stored under the + // chainntnfs interface (concrete type is + // *neutrinonotify.NeutrinoNotifier in production) so reorg-aware + // forwarder tests can substitute a stub that drives the full + // Confirmed / NegativeConf / Done lifecycle without spinning up a + // neutrino chain service. + notifier chainntnfs.ChainNotifier // feeEstimator provides fee estimation from a web API since // neutrino has no mempool visibility. @@ -400,40 +405,89 @@ func (b *ChainBackend) RegisterConf(ctx context.Context, txid *chainhash.Hash, cancelOnce.Do(event.Cancel) } + // Create channels to convert neutrino's confirmation lifecycle + // (which already drives chainntnfs's NegativeConf/Done channels + // directly — neutrino emits these natively from its compact- + // block-filter scanner) into our backend-agnostic types. + // NegativeConf carries a reorg depth that the cross-backend + // chainsource layer intentionally drops, so we forward a bare + // struct{} signal on reorgChan instead. confChan := make(chan *chainsource.TxConfirmation, 1) + reorgChan := make(chan uint64, 1) + doneChan := make(chan struct{}, 1) go func() { + // Defers run in LIFO order. event.Cancel() must run first + // so the upstream notifier stops writing to its internal + // channels before we cancel notifyCtx (which any in-flight + // downstream sends are still using) and finally close the + // outgoing chans. Reversing this order would race the + // upstream notifier against closed channels. defer close(confChan) + defer close(reorgChan) + defer close(doneChan) defer cancel() defer safeCancel() - select { - case lndConf, ok := <-event.Confirmed: - if !ok { - return - } + for { + select { + case lndConf, ok := <-event.Confirmed: + if !ok { + return + } - conf := &chainsource.TxConfirmation{ - BlockHash: lndConf.BlockHash, - BlockHeight: lndConf.BlockHeight, - TxIndex: lndConf.TxIndex, - Tx: lndConf.Tx, - Block: lndConf.Block, - } + conf := &chainsource.TxConfirmation{ + BlockHash: lndConf.BlockHash, + BlockHeight: lndConf.BlockHeight, + TxIndex: lndConf.TxIndex, + Tx: lndConf.Tx, + Block: lndConf.Block, + } + + select { + case confChan <- conf: + case <-notifyCtx.Done(): + return + } + + case _, ok := <-event.NegativeConf: + if !ok { + event.NegativeConf = nil + + continue + } + + select { + case reorgChan <- uint64(0): + case <-notifyCtx.Done(): + return + } + + case _, ok := <-event.Done: + if !ok { + event.Done = nil + + continue + } + + select { + case doneChan <- struct{}{}: + case <-notifyCtx.Done(): + return + } + + return - select { - case confChan <- conf: case <-notifyCtx.Done(): return } - - case <-notifyCtx.Done(): - return } }() return &chainsource.ConfRegistration{ Confirmed: confChan, + Reorged: reorgChan, + Done: doneChan, Cancel: func() { cancel() safeCancel() @@ -470,40 +524,83 @@ func (b *ChainBackend) RegisterSpend(ctx context.Context, cancelOnce.Do(event.Cancel) } + // Create channels to convert neutrino's spend lifecycle into our + // backend-agnostic types. neutrino's Reorg carries no payload, so + // we forward a bare struct{} signal. spendChan := make(chan *chainsource.SpendDetail, 1) + reorgChan := make(chan uint64, 1) + doneChan := make(chan struct{}, 1) go func() { + // LIFO defer: event.Cancel() first so the upstream notifier + // stops writing, then cancel notifyCtx so in-flight downstream + // sends unblock, and finally close the outgoing chans. defer close(spendChan) + defer close(reorgChan) + defer close(doneChan) defer cancel() defer safeCancel() - select { - case lndSpend, ok := <-event.Spend: - if !ok { - return - } + for { + select { + case lndSpend, ok := <-event.Spend: + if !ok { + return + } - spend := &chainsource.SpendDetail{ - SpentOutPoint: lndSpend.SpentOutPoint, - SpenderTxHash: lndSpend.SpenderTxHash, - SpendingTx: lndSpend.SpendingTx, - SpenderInputIndex: lndSpend.SpenderInputIndex, - SpendingHeight: lndSpend.SpendingHeight, - } + spend := &chainsource.SpendDetail{ + SpentOutPoint: lndSpend.SpentOutPoint, + SpenderTxHash: lndSpend.SpenderTxHash, + SpendingTx: lndSpend.SpendingTx, + SpenderInputIndex: lndSpend. + SpenderInputIndex, + SpendingHeight: lndSpend.SpendingHeight, + } + + select { + case spendChan <- spend: + case <-notifyCtx.Done(): + return + } + + case _, ok := <-event.Reorg: + if !ok { + event.Reorg = nil + + continue + } + + select { + case reorgChan <- uint64(0): + case <-notifyCtx.Done(): + return + } + + case _, ok := <-event.Done: + if !ok { + event.Done = nil + + continue + } + + select { + case doneChan <- struct{}{}: + case <-notifyCtx.Done(): + return + } + + return - select { - case spendChan <- spend: case <-notifyCtx.Done(): return } - - case <-notifyCtx.Done(): - return } }() return &chainsource.SpendRegistration{ - Spend: spendChan, + Spend: spendChan, + Reorged: reorgChan, + Done: doneChan, Cancel: func() { cancel() safeCancel() diff --git a/btcwbackend/chain_backend_reorg_test.go b/btcwbackend/chain_backend_reorg_test.go new file mode 100644 index 000000000..da705b44f --- /dev/null +++ b/btcwbackend/chain_backend_reorg_test.go @@ -0,0 +1,359 @@ +package btcwbackend + +import ( + "testing" + "time" + + "github.com/btcsuite/btcd/chainhash/v2" + "github.com/btcsuite/btcd/wire/v2" + "github.com/lightninglabs/darepo-client/chainsource" + "github.com/lightningnetwork/lnd/chainntnfs" + "github.com/stretchr/testify/require" +) + +// reorgWaitTimeout is the per-step deadline used by the btcwbackend +// forwarder reorg tests. The forwarder is a tight goroutine, so the +// timeout exists only to make a hang surface as a fast failure on +// slow CI. +const reorgWaitTimeout = 2 * time.Second + +// fakeNotifier is a minimal chainntnfs.ChainNotifier whose Register* +// methods hand back a caller-supplied event struct. The neutrino-side +// of the real chainntnfs implementation produces events with all four +// channels populated (Confirmed/NegativeConf/Done for conf, +// Spend/Reorg/Done for spend), so the stub lets us drive a synthetic +// upstream lifecycle through the chainbackend forwarder without +// standing up a real neutrino chain service. +type fakeNotifier struct { + confEvent *chainntnfs.ConfirmationEvent + spendEvent *chainntnfs.SpendEvent +} + +// RegisterConfirmationsNtfn returns the stub's pre-built confirmation +// event. +func (n *fakeNotifier) RegisterConfirmationsNtfn(_ *chainhash.Hash, _ []byte, _, + _ uint32, _ ...chainntnfs.NotifierOption) ( + *chainntnfs.ConfirmationEvent, error) { + + return n.confEvent, nil +} + +// RegisterSpendNtfn returns the stub's pre-built spend event. +func (n *fakeNotifier) RegisterSpendNtfn(_ *wire.OutPoint, _ []byte, _ uint32) ( + *chainntnfs.SpendEvent, error) { + + return n.spendEvent, nil +} + +// RegisterBlockEpochNtfn returns an empty block-epoch event. The +// reorg lifecycle tests do not exercise block epochs, so the channels +// are intentionally never written to. +func (n *fakeNotifier) RegisterBlockEpochNtfn(_ *chainntnfs.BlockEpoch) ( + *chainntnfs.BlockEpochEvent, error) { + + return &chainntnfs.BlockEpochEvent{ + Epochs: make(chan *chainntnfs.BlockEpoch), + Cancel: func() {}, + }, nil +} + +// Start satisfies the ChainNotifier interface; the stub has no real +// lifecycle. +func (n *fakeNotifier) Start() error { return nil } + +// Started satisfies the ChainNotifier interface; the stub always +// reports started. +func (n *fakeNotifier) Started() bool { return true } + +// Stop satisfies the ChainNotifier interface; the stub has no real +// lifecycle. +func (n *fakeNotifier) Stop() error { return nil } + +// TestRegisterConfForwardsReorgAndDone drives the full confirmation +// lifecycle through neutrino's chainntnfs notifier into the btcwbackend +// forwarder and asserts each event arrives on the matching chainsource +// registration channel. The lifecycle is: +// +// Confirmed -> NegativeConf -> Confirmed -> Done +// +// and the test additionally verifies the forwarder exits after Done +// by observing that the chainsource channels close. +func TestRegisterConfForwardsReorgAndDone(t *testing.T) { + t.Parallel() + + confChan := make(chan *chainntnfs.TxConfirmation, 2) + negChan := make(chan int32, 1) + doneChan := make(chan struct{}, 1) + notifier := &fakeNotifier{ + confEvent: &chainntnfs.ConfirmationEvent{ + Confirmed: confChan, + NegativeConf: negChan, + Done: doneChan, + Cancel: func() {}, + }, + } + backend := &ChainBackend{notifier: notifier} + + reg, err := backend.RegisterConf( + t.Context(), &chainhash.Hash{0x42}, []byte{0x51}, 1, 100, false, + ) + require.NoError(t, err) + + // 1. First confirmation crosses the forwarder. + hash1 := chainhash.Hash{0xaa} + confChan <- &chainntnfs.TxConfirmation{ + BlockHash: &hash1, + BlockHeight: 123, + Tx: wire.NewMsgTx(2), + } + + conf1 := awaitConfForward(t, reg.Confirmed) + require.Equal(t, uint32(123), conf1.BlockHeight) + require.Equal(t, hash1, *conf1.BlockHash) + + // 2. Reorg ping is forwarded as a single struct{} on Reorged. The + // depth value carried on NegativeConf is intentionally dropped at + // this layer so neutrino and LND backends present the same wire + // shape to chainsource consumers. + negChan <- 1 + + awaitSeq(t, reg.Reorged, "Reorged forward") + + // 3. Transaction re-confirms in a different block on the new tip. + hash2 := chainhash.Hash{0xbb} + confChan <- &chainntnfs.TxConfirmation{ + BlockHash: &hash2, + BlockHeight: 124, + Tx: wire.NewMsgTx(2), + } + + conf2 := awaitConfForward(t, reg.Confirmed) + require.Equal(t, uint32(124), conf2.BlockHeight) + require.Equal(t, hash2, *conf2.BlockHash) + + // 4. Done signal is forwarded; the forwarder then exits and the + // chainsource channels close. + doneChan <- struct{}{} + + awaitStruct(t, reg.Done, "Done forward") + + // All three forwarded channels must close once the forwarder + // exits. + requireConfClosedSoon(t, reg.Confirmed) + requireSeqClosedSoon(t, reg.Reorged) + requireStructClosedSoon(t, reg.Done) +} + +// TestRegisterSpendForwardsReorgAndDone is the spend-side equivalent of +// the confirmation lifecycle test above. +func TestRegisterSpendForwardsReorgAndDone(t *testing.T) { + t.Parallel() + + spendChan := make(chan *chainntnfs.SpendDetail, 2) + reorgChan := make(chan struct{}, 1) + doneChan := make(chan struct{}, 1) + notifier := &fakeNotifier{ + spendEvent: &chainntnfs.SpendEvent{ + Spend: spendChan, + Reorg: reorgChan, + Done: doneChan, + Cancel: func() {}, + }, + } + backend := &ChainBackend{notifier: notifier} + + outpoint := &wire.OutPoint{Index: 1} + reg, err := backend.RegisterSpend( + t.Context(), outpoint, []byte{0x51}, 100, + ) + require.NoError(t, err) + + // 1. First spend. + hash1 := chainhash.Hash{0x10} + spendChan <- &chainntnfs.SpendDetail{ + SpentOutPoint: outpoint, + SpenderTxHash: &hash1, + SpendingTx: wire.NewMsgTx(2), + SpendingHeight: 150, + } + + spend1 := awaitSpendForward(t, reg.Spend) + require.Equal(t, int32(150), spend1.SpendingHeight) + require.Equal(t, hash1, *spend1.SpenderTxHash) + + // 2. Reorg evicts the spend. + reorgChan <- struct{}{} + + awaitSeq(t, reg.Reorged, "spend Reorged forward") + + // 3. A different spender wins the new chain. + hash2 := chainhash.Hash{0x20} + spendChan <- &chainntnfs.SpendDetail{ + SpentOutPoint: outpoint, + SpenderTxHash: &hash2, + SpendingTx: wire.NewMsgTx(2), + SpendingHeight: 151, + } + + spend2 := awaitSpendForward(t, reg.Spend) + require.Equal(t, int32(151), spend2.SpendingHeight) + require.Equal(t, hash2, *spend2.SpenderTxHash) + + // 4. Done. + doneChan <- struct{}{} + + awaitStruct(t, reg.Done, "spend Done forward") + + requireSpendClosedSoon(t, reg.Spend) + requireSeqClosedSoon(t, reg.Reorged) + requireStructClosedSoon(t, reg.Done) +} + +// awaitConfForward reads a single confirmation off the forwarded +// channel with a deadline, failing the test on timeout or unexpected +// close. +func awaitConfForward(t *testing.T, + ch <-chan *chainsource.TxConfirmation) *chainsource.TxConfirmation { + + t.Helper() + + select { + case conf, ok := <-ch: + if !ok { + t.Fatal("conf channel closed before delivery") + } + + return conf + + case <-time.After(reorgWaitTimeout): + t.Fatal("timeout waiting for conf forward") + + return nil + } +} + +// awaitSpendForward reads a single spend off the forwarded channel +// with a deadline, failing the test on timeout or unexpected close. +func awaitSpendForward(t *testing.T, + ch <-chan *chainsource.SpendDetail) *chainsource.SpendDetail { + + t.Helper() + + select { + case spend, ok := <-ch: + if !ok { + t.Fatal("spend channel closed before delivery") + } + + return spend + + case <-time.After(reorgWaitTimeout): + t.Fatal("timeout waiting for spend forward") + + return nil + } +} + +// awaitStruct reads a single struct{} off the forwarded channel with a +// deadline. Used for the Reorged and Done channels. +func awaitStruct(t *testing.T, ch <-chan struct{}, label string) { + t.Helper() + + select { + case _, ok := <-ch: + if !ok { + t.Fatalf("%s channel closed before delivery", label) + } + + case <-time.After(reorgWaitTimeout): + t.Fatalf("timeout waiting for %s", label) + } +} + +// requireConfClosedSoon asserts the confirmation channel closes within +// the reorg wait timeout. Used to verify the forwarder's defer-close +// chain runs after Done is delivered. +func requireConfClosedSoon(t *testing.T, + ch <-chan *chainsource.TxConfirmation) { + + t.Helper() + + require.Eventually(t, func() bool { + select { + case _, ok := <-ch: + return !ok + + default: + return false + } + }, reorgWaitTimeout, 10*time.Millisecond, + "conf channel did not close after Done") +} + +// requireSpendClosedSoon asserts the spend channel closes within the +// reorg wait timeout. +func requireSpendClosedSoon(t *testing.T, ch <-chan *chainsource.SpendDetail) { + t.Helper() + + require.Eventually(t, func() bool { + select { + case _, ok := <-ch: + return !ok + + default: + return false + } + }, reorgWaitTimeout, 10*time.Millisecond, + "spend channel did not close after Done") +} + +// requireStructClosedSoon asserts that a struct{} signal channel +// closes within the reorg wait timeout. +func requireStructClosedSoon(t *testing.T, ch <-chan struct{}) { + t.Helper() + + require.Eventually(t, func() bool { + select { + case _, ok := <-ch: + return !ok + + default: + return false + } + }, reorgWaitTimeout, 10*time.Millisecond, + "struct channel did not close after Done") +} + +// awaitSeq reads a single sequence value off the forwarded Reorged +// channel with a deadline. chainsource retyped Reorged from struct{} +// to the ordering sequence; this backend always sends 0. +func awaitSeq(t *testing.T, ch <-chan uint64, label string) { + t.Helper() + + select { + case _, ok := <-ch: + if !ok { + t.Fatalf("%s channel closed before delivery", label) + } + + case <-time.After(reorgWaitTimeout): + t.Fatalf("timeout waiting for %s", label) + } +} + +// requireSeqClosedSoon asserts that a sequence-carrying signal channel +// closes within the reorg wait timeout. +func requireSeqClosedSoon(t *testing.T, ch <-chan uint64) { + t.Helper() + + require.Eventually(t, func() bool { + select { + case _, ok := <-ch: + return !ok + + default: + return false + } + }, reorgWaitTimeout, 10*time.Millisecond, + "seq channel did not close after Done") +} 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 }