From 5089311ca1d2b5775b70aa8f6f51b25c4a0b7d7b Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 16 Aug 2019 19:55:26 -0700 Subject: [PATCH 1/9] chainntnfs: refactor common registration code within RegisterConfirmationsNtfn --- chainntnfs/bitcoindnotify/bitcoind.go | 29 +-- chainntnfs/btcdnotify/btcd.go | 29 +-- chainntnfs/neutrinonotify/neutrino.go | 35 +--- chainntnfs/txnotifier.go | 105 ++++++++-- chainntnfs/txnotifier_test.go | 289 ++++++++------------------ 5 files changed, 192 insertions(+), 295 deletions(-) diff --git a/chainntnfs/bitcoindnotify/bitcoind.go b/chainntnfs/bitcoindnotify/bitcoind.go index 2c02167b8..d5203de25 100644 --- a/chainntnfs/bitcoindnotify/bitcoind.go +++ b/chainntnfs/bitcoindnotify/bitcoind.go @@ -39,7 +39,6 @@ type chainUpdate struct { // chain client. Multiple concurrent clients are supported. All notifications // are achieved via non-blocking sends on client channels. type BitcoindNotifier struct { - confClientCounter uint64 // To be used atomically. spendClientCounter uint64 // To be used atomically. epochClientCounter uint64 // To be used atomically. @@ -827,41 +826,23 @@ func (b *BitcoindNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, pkScript []byte, numConfs, heightHint uint32) (*chainntnfs.ConfirmationEvent, error) { - // Construct a notification request for the transaction and send it to - // the main event loop. - confID := atomic.AddUint64(&b.confClientCounter, 1) - confRequest, err := chainntnfs.NewConfRequest(txid, pkScript) - if err != nil { - return nil, err - } - ntfn := &chainntnfs.ConfNtfn{ - ConfID: confID, - ConfRequest: confRequest, - NumConfirmations: numConfs, - Event: chainntnfs.NewConfirmationEvent(numConfs, func() { - b.txNotifier.CancelConf(confRequest, confID) - }), - HeightHint: heightHint, - } - - chainntnfs.Log.Infof("New confirmation subscription: %v, num_confs=%v", - confRequest, numConfs) - // Register the conf notification with the TxNotifier. A non-nil value // for `dispatch` will be returned if we are required to perform a // manual scan for the confirmation. Otherwise the notifier will begin // watching at tip for the transaction to confirm. - dispatch, _, err := b.txNotifier.RegisterConf(ntfn) + ntfn, err := b.txNotifier.RegisterConf( + txid, pkScript, numConfs, heightHint, + ) if err != nil { return nil, err } - if dispatch == nil { + if ntfn.HistoricalDispatch == nil { return ntfn.Event, nil } select { - case b.notificationRegistry <- dispatch: + case b.notificationRegistry <- ntfn.HistoricalDispatch: return ntfn.Event, nil case <-b.quit: return nil, chainntnfs.ErrChainNotifierShuttingDown diff --git a/chainntnfs/btcdnotify/btcd.go b/chainntnfs/btcdnotify/btcd.go index bfe42da09..c174f4c85 100644 --- a/chainntnfs/btcdnotify/btcd.go +++ b/chainntnfs/btcdnotify/btcd.go @@ -50,7 +50,6 @@ type txUpdate struct { // notifications. Multiple concurrent clients are supported. All notifications // are achieved via non-blocking sends on client channels. type BtcdNotifier struct { - confClientCounter uint64 // To be used aotmically. spendClientCounter uint64 // To be used atomically. epochClientCounter uint64 // To be used atomically. @@ -851,41 +850,23 @@ func (b *BtcdNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, pkScript []byte, numConfs, heightHint uint32) (*chainntnfs.ConfirmationEvent, error) { - // Construct a notification request for the transaction and send it to - // the main event loop. - confID := atomic.AddUint64(&b.confClientCounter, 1) - confRequest, err := chainntnfs.NewConfRequest(txid, pkScript) - if err != nil { - return nil, err - } - ntfn := &chainntnfs.ConfNtfn{ - ConfID: confID, - ConfRequest: confRequest, - NumConfirmations: numConfs, - Event: chainntnfs.NewConfirmationEvent(numConfs, func() { - b.txNotifier.CancelConf(confRequest, confID) - }), - HeightHint: heightHint, - } - - chainntnfs.Log.Infof("New confirmation subscription: %v, num_confs=%v ", - confRequest, numConfs) - // Register the conf notification with the TxNotifier. A non-nil value // for `dispatch` will be returned if we are required to perform a // manual scan for the confirmation. Otherwise the notifier will begin // watching at tip for the transaction to confirm. - dispatch, _, err := b.txNotifier.RegisterConf(ntfn) + ntfn, err := b.txNotifier.RegisterConf( + txid, pkScript, numConfs, heightHint, + ) if err != nil { return nil, err } - if dispatch == nil { + if ntfn.HistoricalDispatch == nil { return ntfn.Event, nil } select { - case b.notificationRegistry <- dispatch: + case b.notificationRegistry <- ntfn.HistoricalDispatch: return ntfn.Event, nil case <-b.quit: return nil, chainntnfs.ErrChainNotifierShuttingDown diff --git a/chainntnfs/neutrinonotify/neutrino.go b/chainntnfs/neutrinonotify/neutrino.go index bc5c27d7d..76f6dcbcd 100644 --- a/chainntnfs/neutrinonotify/neutrino.go +++ b/chainntnfs/neutrinonotify/neutrino.go @@ -37,7 +37,6 @@ const ( // TODO(roasbeef): heavily consolidate with NeutrinoNotifier code // * maybe combine into single package? type NeutrinoNotifier struct { - confClientCounter uint64 // To be used atomically. spendClientCounter uint64 // To be used atomically. epochClientCounter uint64 // To be used atomically. @@ -820,31 +819,13 @@ func (n *NeutrinoNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, pkScript []byte, numConfs, heightHint uint32) (*chainntnfs.ConfirmationEvent, error) { - // Construct a notification request for the transaction and send it to - // the main event loop. - confID := atomic.AddUint64(&n.confClientCounter, 1) - confRequest, err := chainntnfs.NewConfRequest(txid, pkScript) - if err != nil { - return nil, err - } - ntfn := &chainntnfs.ConfNtfn{ - ConfID: confID, - ConfRequest: confRequest, - NumConfirmations: numConfs, - Event: chainntnfs.NewConfirmationEvent(numConfs, func() { - n.txNotifier.CancelConf(confRequest, confID) - }), - HeightHint: heightHint, - } - - chainntnfs.Log.Infof("New confirmation subscription: %v, num_confs=%v", - confRequest, numConfs) - // Register the conf notification with the TxNotifier. A non-nil value // for `dispatch` will be returned if we are required to perform a // manual scan for the confirmation. Otherwise the notifier will begin // watching at tip for the transaction to confirm. - dispatch, txNotifierTip, err := n.txNotifier.RegisterConf(ntfn) + ntfn, err := n.txNotifier.RegisterConf( + txid, pkScript, numConfs, heightHint, + ) if err != nil { return nil, err } @@ -859,9 +840,7 @@ func (n *NeutrinoNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, // type so we can instruct neutrino to match if the transaction // containing the script is found in a block. params := n.p2pNode.ChainParams() - _, addrs, _, err := txscript.ExtractPkScriptAddrs( - confRequest.PkScript.Script(), ¶ms, - ) + _, addrs, _, err := txscript.ExtractPkScriptAddrs(pkScript, ¶ms) if err != nil { return nil, fmt.Errorf("unable to extract script: %v", err) } @@ -873,7 +852,7 @@ func (n *NeutrinoNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, case n.notificationRegistry <- &rescanFilterUpdate{ updateOptions: []neutrino.UpdateOption{ neutrino.AddAddrs(addrs...), - neutrino.Rewind(txNotifierTip), + neutrino.Rewind(ntfn.Height), neutrino.DisableDisconnectedNtfns(true), }, errChan: errChan, @@ -893,14 +872,14 @@ func (n *NeutrinoNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, // If a historical rescan was not requested by the txNotifier, then we // can return to the caller. - if dispatch == nil { + if ntfn.HistoricalDispatch == nil { return ntfn.Event, nil } // Finally, with the filter updated, we can dispatch the historical // rescan to ensure we can detect if the event happened in the past. select { - case n.notificationRegistry <- dispatch: + case n.notificationRegistry <- ntfn.HistoricalDispatch: case <-n.quit: return nil, chainntnfs.ErrChainNotifierShuttingDown } diff --git a/chainntnfs/txnotifier.go b/chainntnfs/txnotifier.go index 0cc6948b0..5e503f217 100644 --- a/chainntnfs/txnotifier.go +++ b/chainntnfs/txnotifier.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/txscript" @@ -255,6 +256,25 @@ type HistoricalConfDispatch struct { EndHeight uint32 } +// ConfRegistration encompasses all of the information required for callers to +// retrieve details about a confirmation event. +type ConfRegistration struct { + // Event contains references to the channels that the notifications are + // to be sent over. + Event *ConfirmationEvent + + // HistoricalDispatch, if non-nil, signals to the client who registered + // the notification that they are responsible for attempting to manually + // rescan blocks for the txid/output script between the start and end + // heights. + HistoricalDispatch *HistoricalConfDispatch + + // Height is the height of the TxNotifier at the time the confirmation + // notification was registered. This can be used so that backends can + // request to be notified of confirmations from this point forwards. + Height uint32 +} + // SpendRequest encapsulates a request for a spend notification of either an // outpoint or output script. type SpendRequest struct { @@ -401,6 +421,8 @@ type HistoricalSpendDispatch struct { // The TxNotifier will watch the blockchain as new blocks come in, in order to // satisfy its client requests. type TxNotifier struct { + confClientCounter uint64 // To be used atomically. + // currentHeight is the height of the tracked blockchain. It is used to // determine the number of confirmations a tx has and ensure blocks are // connected and disconnected in order. @@ -484,34 +506,60 @@ func NewTxNotifier(startHeight uint32, reorgSafetyLimit uint32, } } +// newConfNtfn validates all of the parameters required to successfully create +// and register a confirmation notification. +func (n *TxNotifier) newConfNtfn(txid *chainhash.Hash, + pkScript []byte, numConfs, heightHint uint32) (*ConfNtfn, error) { + + confRequest, err := NewConfRequest(txid, pkScript) + if err != nil { + return nil, err + } + + // Enforce that we will not dispatch confirmations beyond the reorg + // safety limit. + if numConfs > n.reorgSafetyLimit { + return nil, ErrTxMaxConfs + } + + confID := atomic.AddUint64(&n.confClientCounter, 1) + return &ConfNtfn{ + ConfID: confID, + ConfRequest: confRequest, + NumConfirmations: numConfs, + Event: NewConfirmationEvent(numConfs, func() { + n.CancelConf(confRequest, confID) + }), + HeightHint: heightHint, + }, nil +} + // RegisterConf handles a new confirmation notification request. The client will // be notified when the transaction/output script gets a sufficient number of -// confirmations in the blockchain. The registration succeeds if no error is -// returned. If the returned HistoricalConfDispatch is non-nil, the caller is -// responsible for attempting to manually rescan blocks for the txid/output -// script between the start and end heights. The notifier's current height is -// also returned so that backends can request to be notified of confirmations -// from this point forwards. +// confirmations in the blockchain. // // NOTE: If the transaction/output script has already been included in a block // on the chain, the confirmation details must be provided with the // UpdateConfDetails method, otherwise we will wait for the transaction/output // script to confirm even though it already has. -func (n *TxNotifier) RegisterConf(ntfn *ConfNtfn) (*HistoricalConfDispatch, - uint32, error) { +func (n *TxNotifier) RegisterConf(txid *chainhash.Hash, pkScript []byte, + numConfs, heightHint uint32) (*ConfRegistration, error) { select { case <-n.quit: - return nil, 0, ErrTxNotifierExiting + return nil, ErrTxNotifierExiting default: } - // Enforce that we will not dispatch confirmations beyond the reorg - // safety limit. - if ntfn.NumConfirmations > n.reorgSafetyLimit { - return nil, 0, ErrTxMaxConfs + // We'll start by performing a series of validation checks. + ntfn, err := n.newConfNtfn(txid, pkScript, numConfs, heightHint) + if err != nil { + return nil, err } + Log.Infof("New confirmation subscription: %v, num_confs=%v ", + ntfn.ConfRequest, numConfs) + // Before proceeding to register the notification, we'll query our // height hint cache to determine whether a better one exists. // @@ -557,9 +605,16 @@ func (n *TxNotifier) RegisterConf(ntfn *ConfNtfn) (*HistoricalConfDispatch, "registration since rescan has finished", ntfn.ConfRequest) - return nil, n.currentHeight, n.dispatchConfDetails( - ntfn, confSet.details, - ) + err := n.dispatchConfDetails(ntfn, confSet.details) + if err != nil { + return nil, err + } + + return &ConfRegistration{ + Event: ntfn.Event, + HistoricalDispatch: nil, + Height: n.currentHeight, + }, nil // A rescan is already in progress, return here to prevent dispatching // another. When the rescan returns, this notification's details will be @@ -568,7 +623,11 @@ func (n *TxNotifier) RegisterConf(ntfn *ConfNtfn) (*HistoricalConfDispatch, Log.Debugf("Waiting for pending rescan to finish before "+ "notifying %v at tip", ntfn.ConfRequest) - return nil, n.currentHeight, nil + return &ConfRegistration{ + Event: ntfn.Event, + HistoricalDispatch: nil, + Height: n.currentHeight, + }, nil // If no rescan has been dispatched, attempt to do so now. case rescanNotStarted: @@ -587,7 +646,11 @@ func (n *TxNotifier) RegisterConf(ntfn *ConfNtfn) (*HistoricalConfDispatch, // notifier to start delivering messages for this set // immediately. confSet.rescanStatus = rescanComplete - return nil, n.currentHeight, nil + return &ConfRegistration{ + Event: ntfn.Event, + HistoricalDispatch: nil, + Height: n.currentHeight, + }, nil } Log.Debugf("Dispatching historical confirmation rescan for %v", @@ -607,7 +670,11 @@ func (n *TxNotifier) RegisterConf(ntfn *ConfNtfn) (*HistoricalConfDispatch, // registrations don't also attempt a dispatch. confSet.rescanStatus = rescanPending - return dispatch, n.currentHeight, nil + return &ConfRegistration{ + Event: ntfn.Event, + HistoricalDispatch: dispatch, + Height: n.currentHeight, + }, nil } // CancelConf cancels an existing request for a spend notification of an diff --git a/chainntnfs/txnotifier_test.go b/chainntnfs/txnotifier_test.go index 0023e3b07..ecfc327f7 100644 --- a/chainntnfs/txnotifier_test.go +++ b/chainntnfs/txnotifier_test.go @@ -136,22 +136,17 @@ func TestTxNotifierMaxConfs(t *testing.T) { // Registering one confirmation above the maximum should fail with // ErrTxMaxConfs. - ntfn := &chainntnfs.ConfNtfn{ - ConfRequest: chainntnfs.ConfRequest{ - TxID: chainntnfs.ZeroHash, - PkScript: testScript, - }, - NumConfirmations: chainntnfs.MaxNumConfs + 1, - Event: chainntnfs.NewConfirmationEvent( - chainntnfs.MaxNumConfs, nil, - ), - } - if _, _, err := n.RegisterConf(ntfn); err != chainntnfs.ErrTxMaxConfs { + _, err := n.RegisterConf( + &chainntnfs.ZeroHash, testRawScript, chainntnfs.MaxNumConfs+1, 1, + ) + if err != chainntnfs.ErrTxMaxConfs { t.Fatalf("expected chainntnfs.ErrTxMaxConfs, got %v", err) } - ntfn.NumConfirmations-- - if _, _, err := n.RegisterConf(ntfn); err != nil { + _, err = n.RegisterConf( + &chainntnfs.ZeroHash, testRawScript, chainntnfs.MaxNumConfs, 1, + ) + if err != nil { t.Fatalf("unable to register conf ntfn: %v", err) } } @@ -176,29 +171,17 @@ func TestTxNotifierFutureConfDispatch(t *testing.T) { // notifications. tx1 := wire.MsgTx{Version: 1} tx1.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn1 := chainntnfs.ConfNtfn{ - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx1.TxHash(), - PkScript: testScript, - }, - NumConfirmations: tx1NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx1NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn1); err != nil { + tx1Hash := tx1.TxHash() + ntfn1, err := n.RegisterConf(&tx1Hash, testRawScript, tx1NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } tx2 := wire.MsgTx{Version: 2} tx2.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn2 := chainntnfs.ConfNtfn{ - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx2.TxHash(), - PkScript: testScript, - }, - NumConfirmations: tx2NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx2NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn2); err != nil { + tx2Hash := tx2.TxHash() + ntfn2, err := n.RegisterConf(&tx2Hash, testRawScript, tx2NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } @@ -226,7 +209,7 @@ func TestTxNotifierFutureConfDispatch(t *testing.T) { Transactions: []*wire.MsgTx{&tx1, &tx2}, }) - err := n.ConnectTip(block1.Hash(), 11, block1.Transactions()) + err = n.ConnectTip(block1.Hash(), 11, block1.Transactions()) if err != nil { t.Fatalf("Failed to connect block: %v", err) } @@ -361,24 +344,14 @@ func TestTxNotifierHistoricalConfDispatch(t *testing.T) { // Create the test transactions at a height before the TxNotifier's // starting height so that they are confirmed once registering them. tx1Hash := tx1.TxHash() - ntfn1 := chainntnfs.ConfNtfn{ - ConfID: 0, - ConfRequest: chainntnfs.ConfRequest{TxID: tx1Hash}, - NumConfirmations: tx1NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx1NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn1); err != nil { + ntfn1, err := n.RegisterConf(&tx1Hash, testRawScript, tx1NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } tx2Hash := tx2.TxHash() - ntfn2 := chainntnfs.ConfNtfn{ - ConfID: 1, - ConfRequest: chainntnfs.ConfRequest{TxID: tx2Hash}, - NumConfirmations: tx2NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx2NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn2); err != nil { + ntfn2, err := n.RegisterConf(&tx2Hash, testRawScript, tx2NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } @@ -390,7 +363,7 @@ func TestTxNotifierHistoricalConfDispatch(t *testing.T) { TxIndex: 1, Tx: &tx1, } - err := n.UpdateConfDetails(ntfn1.ConfRequest, &txConf1) + err = n.UpdateConfDetails(ntfn1.HistoricalDispatch.ConfRequest, &txConf1) if err != nil { t.Fatalf("unable to update conf details: %v", err) } @@ -424,7 +397,7 @@ func TestTxNotifierHistoricalConfDispatch(t *testing.T) { TxIndex: 2, Tx: &tx2, } - err = n.UpdateConfDetails(ntfn2.ConfRequest, &txConf2) + err = n.UpdateConfDetails(ntfn2.HistoricalDispatch.ConfRequest, &txConf2) if err != nil { t.Fatalf("unable to update conf details: %v", err) } @@ -693,34 +666,22 @@ func TestTxNotifierMultipleHistoricalConfRescans(t *testing.T) { // The first registration for a transaction in the notifier should // request a historical confirmation rescan as it does not have a // historical view of the chain. - confNtfn1 := &chainntnfs.ConfNtfn{ - ConfID: 0, - // TODO(wilmer): set pkScript. - ConfRequest: chainntnfs.ConfRequest{TxID: chainntnfs.ZeroHash}, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - historicalConfDispatch1, _, err := n.RegisterConf(confNtfn1) + ntfn1, err := n.RegisterConf(&chainntnfs.ZeroHash, testRawScript, 1, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalConfDispatch1 == nil { + if ntfn1.HistoricalDispatch == nil { t.Fatal("expected to receive historical dispatch request") } // We'll register another confirmation notification for the same // transaction. This should not request a historical confirmation rescan // since the first one is still pending. - confNtfn2 := &chainntnfs.ConfNtfn{ - ConfID: 1, - // TODO(wilmer): set pkScript. - ConfRequest: chainntnfs.ConfRequest{TxID: chainntnfs.ZeroHash}, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - historicalConfDispatch2, _, err := n.RegisterConf(confNtfn2) + ntfn2, err := n.RegisterConf(&chainntnfs.ZeroHash, testRawScript, 1, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalConfDispatch2 != nil { + if ntfn2.HistoricalDispatch != nil { t.Fatal("received unexpected historical rescan request") } @@ -731,21 +692,16 @@ func TestTxNotifierMultipleHistoricalConfRescans(t *testing.T) { confDetails := &chainntnfs.TxConfirmation{ BlockHeight: startingHeight - 1, } - err = n.UpdateConfDetails(confNtfn2.ConfRequest, confDetails) + err = n.UpdateConfDetails(ntfn1.HistoricalDispatch.ConfRequest, confDetails) if err != nil { t.Fatalf("unable to update conf details: %v", err) } - confNtfn3 := &chainntnfs.ConfNtfn{ - ConfID: 2, - ConfRequest: chainntnfs.ConfRequest{TxID: chainntnfs.ZeroHash}, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - historicalConfDispatch3, _, err := n.RegisterConf(confNtfn3) + ntfn3, err := n.RegisterConf(&chainntnfs.ZeroHash, testRawScript, 1, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalConfDispatch3 != nil { + if ntfn3.HistoricalDispatch != nil { t.Fatal("received unexpected historical rescan request") } } @@ -848,23 +804,16 @@ func TestTxNotifierMultipleHistoricalNtfns(t *testing.T) { var txid chainhash.Hash copy(txid[:], bytes.Repeat([]byte{0x01}, 32)) - confRequest := chainntnfs.ConfRequest{ - // TODO(wilmer): set pkScript. - TxID: txid, - } // We'll start off by registered 5 clients for a confirmation // notification on the same transaction. - confNtfns := make([]*chainntnfs.ConfNtfn, numNtfns) + confNtfns := make([]*chainntnfs.ConfRegistration, numNtfns) for i := uint64(0); i < numNtfns; i++ { - confNtfns[i] = &chainntnfs.ConfNtfn{ - ConfID: i, - ConfRequest: confRequest, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - if _, _, err := n.RegisterConf(confNtfns[i]); err != nil { + ntfn, err := n.RegisterConf(&txid, testRawScript, 1, 1) + if err != nil { t.Fatalf("unable to register conf ntfn #%d: %v", i, err) } + confNtfns[i] = ntfn } // Ensure none of them have received the confirmation details. @@ -884,7 +833,9 @@ func TestTxNotifierMultipleHistoricalNtfns(t *testing.T) { BlockHeight: startingHeight - 1, Tx: wire.NewMsgTx(1), } - err := n.UpdateConfDetails(confNtfns[0].ConfRequest, expectedConfDetails) + err := n.UpdateConfDetails( + confNtfns[0].HistoricalDispatch.ConfRequest, expectedConfDetails, + ) if err != nil { t.Fatalf("unable to update conf details: %v", err) } @@ -905,16 +856,11 @@ func TestTxNotifierMultipleHistoricalNtfns(t *testing.T) { // we'll register another client for the same transaction. We should not // see a historical rescan request and the confirmation notification // should come through immediately. - extraConfNtfn := &chainntnfs.ConfNtfn{ - ConfID: numNtfns + 1, - ConfRequest: confRequest, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - historicalConfRescan, _, err := n.RegisterConf(extraConfNtfn) + extraConfNtfn, err := n.RegisterConf(&txid, testRawScript, 1, 1) if err != nil { t.Fatalf("unable to register conf ntfn: %v", err) } - if historicalConfRescan != nil { + if extraConfNtfn.HistoricalDispatch != nil { t.Fatal("received unexpected historical rescan request") } @@ -1016,31 +962,17 @@ func TestTxNotifierCancelConf(t *testing.T) { // canceled. tx1 := wire.NewMsgTx(1) tx1.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn1 := &chainntnfs.ConfNtfn{ - ConfID: 1, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx1.TxHash(), - PkScript: testScript, - }, - NumConfirmations: 1, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - if _, _, err := n.RegisterConf(ntfn1); err != nil { + tx1Hash := tx1.TxHash() + ntfn1, err := n.RegisterConf(&tx1Hash, testRawScript, 1, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } tx2 := wire.NewMsgTx(2) tx2.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn2 := &chainntnfs.ConfNtfn{ - ConfID: 2, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx2.TxHash(), - PkScript: testScript, - }, - NumConfirmations: 1, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - if _, _, err := n.RegisterConf(ntfn2); err != nil { + tx2Hash := tx2.TxHash() + ntfn2, err := n.RegisterConf(&tx2Hash, testRawScript, 1, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } @@ -1057,9 +989,9 @@ func TestTxNotifierCancelConf(t *testing.T) { // Before extending the notifier's tip with the block above, we'll // cancel the second request. - n.CancelConf(ntfn2.ConfRequest, ntfn2.ConfID) + n.CancelConf(ntfn2.HistoricalDispatch.ConfRequest, 2) - err := n.ConnectTip(block.Hash(), startingHeight+1, block.Transactions()) + err = n.ConnectTip(block.Hash(), startingHeight+1, block.Transactions()) if err != nil { t.Fatalf("unable to connect block: %v", err) } @@ -1200,60 +1132,42 @@ func TestTxNotifierConfReorg(t *testing.T) { // Tx 1 will be confirmed in block 9 and requires 2 confs. tx1 := wire.MsgTx{Version: 1} tx1.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn1 := chainntnfs.ConfNtfn{ - ConfID: 1, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx1.TxHash(), - PkScript: testScript, - }, - NumConfirmations: tx1NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx1NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn1); err != nil { + tx1Hash := tx1.TxHash() + ntfn1, err := n.RegisterConf(&tx1Hash, testRawScript, tx1NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } - if err := n.UpdateConfDetails(ntfn1.ConfRequest, nil); err != nil { + err = n.UpdateConfDetails(ntfn1.HistoricalDispatch.ConfRequest, nil) + if err != nil { t.Fatalf("unable to deliver conf details: %v", err) } // Tx 2 will be confirmed in block 10 and requires 1 conf. tx2 := wire.MsgTx{Version: 2} tx2.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn2 := chainntnfs.ConfNtfn{ - ConfID: 2, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx2.TxHash(), - PkScript: testScript, - }, - NumConfirmations: tx2NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx2NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn2); err != nil { + tx2Hash := tx2.TxHash() + ntfn2, err := n.RegisterConf(&tx2Hash, testRawScript, tx2NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } - if err := n.UpdateConfDetails(ntfn2.ConfRequest, nil); err != nil { + err = n.UpdateConfDetails(ntfn2.HistoricalDispatch.ConfRequest, nil) + if err != nil { t.Fatalf("unable to deliver conf details: %v", err) } // Tx 3 will be confirmed in block 10 and requires 2 confs. tx3 := wire.MsgTx{Version: 3} tx3.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn3 := chainntnfs.ConfNtfn{ - ConfID: 3, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx3.TxHash(), - PkScript: testScript, - }, - NumConfirmations: tx3NumConfs, - Event: chainntnfs.NewConfirmationEvent(tx3NumConfs, nil), - } - if _, _, err := n.RegisterConf(&ntfn3); err != nil { + tx3Hash := tx3.TxHash() + ntfn3, err := n.RegisterConf(&tx3Hash, testRawScript, tx3NumConfs, 1) + if err != nil { t.Fatalf("unable to register ntfn: %v", err) } - if err := n.UpdateConfDetails(ntfn3.ConfRequest, nil); err != nil { + err = n.UpdateConfDetails(ntfn3.HistoricalDispatch.ConfRequest, nil) + if err != nil { t.Fatalf("unable to deliver conf details: %v", err) } @@ -1397,7 +1311,7 @@ func TestTxNotifierConfReorg(t *testing.T) { }) block4 := btcutil.NewBlock(&wire.MsgBlock{}) - err := n.ConnectTip(block3.Hash(), 12, block3.Transactions()) + err = n.ConnectTip(block3.Hash(), 12, block3.Transactions()) if err != nil { t.Fatalf("Failed to connect block: %v", err) } @@ -1726,45 +1640,30 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // Create two test transactions and register them for notifications. tx1 := wire.MsgTx{Version: 1} tx1.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn1 := &chainntnfs.ConfNtfn{ - ConfID: 1, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx1.TxHash(), - PkScript: testScript, - }, - NumConfirmations: 1, - Event: chainntnfs.NewConfirmationEvent(1, nil), + tx1Hash := tx1.TxHash() + ntfn1, err := n.RegisterConf(&tx1Hash, testRawScript, 1, 1) + if err != nil { + t.Fatalf("unable to register tx1: %v", err) } tx2 := wire.MsgTx{Version: 2} tx2.AddTxOut(&wire.TxOut{PkScript: testRawScript}) - ntfn2 := &chainntnfs.ConfNtfn{ - ConfID: 2, - ConfRequest: chainntnfs.ConfRequest{ - TxID: tx2.TxHash(), - PkScript: testScript, - }, - NumConfirmations: 2, - Event: chainntnfs.NewConfirmationEvent(2, nil), - } - - if _, _, err := n.RegisterConf(ntfn1); err != nil { - t.Fatalf("unable to register tx1: %v", err) - } - if _, _, err := n.RegisterConf(ntfn2); err != nil { + tx2Hash := tx2.TxHash() + ntfn2, err := n.RegisterConf(&tx2Hash, testRawScript, 2, 1) + if err != nil { t.Fatalf("unable to register tx2: %v", err) } // Both transactions should not have a height hint set, as RegisterConf // should not alter the cache state. - _, err := hintCache.QueryConfirmHint(ntfn1.ConfRequest) + _, err = hintCache.QueryConfirmHint(ntfn1.HistoricalDispatch.ConfRequest) if err != chainntnfs.ErrConfirmHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "want: %v, got %v", chainntnfs.ErrConfirmHintNotFound, err) } - _, err = hintCache.QueryConfirmHint(ntfn2.ConfRequest) + _, err = hintCache.QueryConfirmHint(ntfn2.HistoricalDispatch.ConfRequest) if err != chainntnfs.ErrConfirmHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "want: %v, got %v", @@ -1790,14 +1689,14 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // the height hints should remain unchanged. This simulates blocks // confirming while the historical dispatch is processing the // registration. - hint, err := hintCache.QueryConfirmHint(ntfn1.ConfRequest) + hint, err := hintCache.QueryConfirmHint(ntfn1.HistoricalDispatch.ConfRequest) if err != chainntnfs.ErrConfirmHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "want: %v, got %v", chainntnfs.ErrConfirmHintNotFound, err) } - hint, err = hintCache.QueryConfirmHint(ntfn2.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn2.HistoricalDispatch.ConfRequest) if err != chainntnfs.ErrConfirmHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "want: %v, got %v", @@ -1806,10 +1705,12 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // Now, update the conf details reporting that the neither txn was found // in the historical dispatch. - if err := n.UpdateConfDetails(ntfn1.ConfRequest, nil); err != nil { + err = n.UpdateConfDetails(ntfn1.HistoricalDispatch.ConfRequest, nil) + if err != nil { t.Fatalf("unable to update conf details: %v", err) } - if err := n.UpdateConfDetails(ntfn2.ConfRequest, nil); err != nil { + err = n.UpdateConfDetails(ntfn2.HistoricalDispatch.ConfRequest, nil) + if err != nil { t.Fatalf("unable to update conf details: %v", err) } @@ -1830,7 +1731,7 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // Now that both notifications are waiting at tip for confirmations, // they should have their height hints updated to the latest block // height. - hint, err = hintCache.QueryConfirmHint(ntfn1.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn1.HistoricalDispatch.ConfRequest) if err != nil { t.Fatalf("unable to query for hint: %v", err) } @@ -1839,7 +1740,7 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { tx1Height, hint) } - hint, err = hintCache.QueryConfirmHint(ntfn2.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn2.HistoricalDispatch.ConfRequest) if err != nil { t.Fatalf("unable to query for hint: %v", err) } @@ -1863,7 +1764,7 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { } // The height hint for the first transaction should remain the same. - hint, err = hintCache.QueryConfirmHint(ntfn1.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn1.HistoricalDispatch.ConfRequest) if err != nil { t.Fatalf("unable to query for hint: %v", err) } @@ -1874,7 +1775,7 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // The height hint for the second transaction should now be updated to // reflect its confirmation. - hint, err = hintCache.QueryConfirmHint(ntfn2.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn2.HistoricalDispatch.ConfRequest) if err != nil { t.Fatalf("unable to query for hint: %v", err) } @@ -1891,7 +1792,7 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // This should update the second transaction's height hint within the // cache to the previous height. - hint, err = hintCache.QueryConfirmHint(ntfn2.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn2.HistoricalDispatch.ConfRequest) if err != nil { t.Fatalf("unable to query for hint: %v", err) } @@ -1902,7 +1803,7 @@ func TestTxNotifierConfirmHintCache(t *testing.T) { // The first transaction's height hint should remain at the original // confirmation height. - hint, err = hintCache.QueryConfirmHint(ntfn2.ConfRequest) + hint, err = hintCache.QueryConfirmHint(ntfn2.HistoricalDispatch.ConfRequest) if err != nil { t.Fatalf("unable to query for hint: %v", err) } @@ -2122,16 +2023,8 @@ func TestTxNotifierNtfnDone(t *testing.T) { // We'll start by creating two notification requests: one confirmation // and one spend. - confNtfn := &chainntnfs.ConfNtfn{ - ConfID: 1, - ConfRequest: chainntnfs.ConfRequest{ - TxID: chainntnfs.ZeroHash, - PkScript: testScript, - }, - NumConfirmations: 1, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - if _, _, err := n.RegisterConf(confNtfn); err != nil { + confNtfn, err := n.RegisterConf(&chainntnfs.ZeroHash, testRawScript, 1, 1) + if err != nil { t.Fatalf("unable to register conf ntfn: %v", err) } @@ -2160,7 +2053,7 @@ func TestTxNotifierNtfnDone(t *testing.T) { Transactions: []*wire.MsgTx{tx, spendTx}, }) - err := n.ConnectTip(block.Hash(), 11, block.Transactions()) + err = n.ConnectTip(block.Hash(), 11, block.Transactions()) if err != nil { t.Fatalf("unable to connect block: %v", err) } @@ -2268,13 +2161,8 @@ func TestTxNotifierTearDown(t *testing.T) { // To begin the test, we'll register for a confirmation and spend // notification. - confNtfn := &chainntnfs.ConfNtfn{ - ConfID: 1, - ConfRequest: chainntnfs.ConfRequest{TxID: chainntnfs.ZeroHash}, - NumConfirmations: 1, - Event: chainntnfs.NewConfirmationEvent(1, nil), - } - if _, _, err := n.RegisterConf(confNtfn); err != nil { + confNtfn, err := n.RegisterConf(&chainntnfs.ZeroHash, testRawScript, 1, 1) + if err != nil { t.Fatalf("unable to register conf ntfn: %v", err) } @@ -2320,7 +2208,8 @@ func TestTxNotifierTearDown(t *testing.T) { // Now that the notifier is torn down, we should no longer be able to // register notification requests. - if _, _, err := n.RegisterConf(confNtfn); err == nil { + _, err = n.RegisterConf(&chainntnfs.ZeroHash, testRawScript, 1, 1) + if err == nil { t.Fatal("expected confirmation registration to fail") } if _, _, err := n.RegisterSpend(spendNtfn); err == nil { From 7821eb6ffb745b27dbb3a7b3819f5a8f6b686920 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 16 Aug 2019 19:56:26 -0700 Subject: [PATCH 2/9] chainntnfs: refactor common registration code within RegisterSpendNtfn --- chainntnfs/bitcoindnotify/bitcoind.go | 59 +++---- chainntnfs/btcdnotify/btcd.go | 75 ++++---- chainntnfs/neutrinonotify/neutrino.go | 52 +++--- chainntnfs/txnotifier.go | 100 ++++++++--- chainntnfs/txnotifier_test.go | 241 +++++++++----------------- 5 files changed, 240 insertions(+), 287 deletions(-) diff --git a/chainntnfs/bitcoindnotify/bitcoind.go b/chainntnfs/bitcoindnotify/bitcoind.go index d5203de25..f9fbfaf65 100644 --- a/chainntnfs/bitcoindnotify/bitcoind.go +++ b/chainntnfs/bitcoindnotify/bitcoind.go @@ -9,6 +9,7 @@ import ( "github.com/btcsuite/btcd/btcjson" "github.com/btcsuite/btcd/chaincfg" "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcutil" "github.com/btcsuite/btcwallet/chain" @@ -39,7 +40,6 @@ type chainUpdate struct { // chain client. Multiple concurrent clients are supported. All notifications // are achieved via non-blocking sends on client channels. type BitcoindNotifier struct { - spendClientCounter uint64 // To be used atomically. epochClientCounter uint64 // To be used atomically. started int32 // To be used atomically. @@ -615,23 +615,11 @@ func (b *BitcoindNotifier) notifyBlockEpochClient(epochClient *blockEpochRegistr func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, error) { - // First, we'll construct a spend notification request and hand it off - // to the txNotifier. - spendID := atomic.AddUint64(&b.spendClientCounter, 1) - spendRequest, err := chainntnfs.NewSpendRequest(outpoint, pkScript) - if err != nil { - return nil, err - } - ntfn := &chainntnfs.SpendNtfn{ - SpendID: spendID, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(func() { - b.txNotifier.CancelSpend(spendRequest, spendID) - }), - HeightHint: heightHint, - } - - historicalDispatch, _, err := b.txNotifier.RegisterSpend(ntfn) + // Register the conf notification with the TxNotifier. A non-nil value + // for `dispatch` will be returned if we are required to perform a + // manual scan for the confirmation. Otherwise the notifier will begin + // watching at tip for the transaction to confirm. + ntfn, err := b.txNotifier.RegisterSpend(outpoint, pkScript, heightHint) if err != nil { return nil, err } @@ -640,17 +628,18 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // outpoint/output script as spent. // // TODO(wilmer): use LoadFilter API instead. - if spendRequest.OutPoint == chainntnfs.ZeroOutPoint { - addr, err := spendRequest.PkScript.Address(b.chainParams) + if outpoint == nil || *outpoint == chainntnfs.ZeroOutPoint { + _, addrs, _, err := txscript.ExtractPkScriptAddrs( + pkScript, b.chainParams, + ) if err != nil { - return nil, err + return nil, fmt.Errorf("unable to parse script: %v", err) } - addrs := []btcutil.Address{addr} if err := b.chainConn.NotifyReceived(addrs); err != nil { return nil, err } } else { - ops := []*wire.OutPoint{&spendRequest.OutPoint} + ops := []*wire.OutPoint{outpoint} if err := b.chainConn.NotifySpent(ops); err != nil { return nil, err } @@ -659,7 +648,7 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // If the txNotifier didn't return any details to perform a historical // scan of the chain, then we can return early as there's nothing left // for us to do. - if historicalDispatch == nil { + if ntfn.HistoricalDispatch == nil { return ntfn.Event, nil } @@ -669,9 +658,9 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // We'll short-circuit the path when dispatching the spend of a script, // rather than an outpoint, as there aren't any additional checks we can // make for scripts. - if spendRequest.OutPoint == chainntnfs.ZeroOutPoint { + if ntfn.HistoricalDispatch.OutPoint == chainntnfs.ZeroOutPoint { select { - case b.notificationRegistry <- historicalDispatch: + case b.notificationRegistry <- ntfn.HistoricalDispatch: case <-b.quit: return nil, chainntnfs.ErrChainNotifierShuttingDown } @@ -686,16 +675,16 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // We'll start by checking the backend's UTXO set to determine whether // the outpoint has been spent. If it hasn't, we can return to the // caller as well. - txOut, err := b.chainConn.GetTxOut( - &spendRequest.OutPoint.Hash, spendRequest.OutPoint.Index, true, - ) + txOut, err := b.chainConn.GetTxOut(&outpoint.Hash, outpoint.Index, true) if err != nil { return nil, err } if txOut != nil { // We'll let the txNotifier know the outpoint is still unspent // in order to begin updating its spend hint. - err := b.txNotifier.UpdateSpendDetails(spendRequest, nil) + err := b.txNotifier.UpdateSpendDetails( + ntfn.HistoricalDispatch.SpendRequest, nil, + ) if err != nil { return nil, err } @@ -710,14 +699,14 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // index (if enabled) to determine if we have a better rescan starting // height. We can do this as the GetRawTransaction call will return the // hash of the block it was included in within the chain. - tx, err := b.chainConn.GetRawTransactionVerbose(&spendRequest.OutPoint.Hash) + tx, err := b.chainConn.GetRawTransactionVerbose(&outpoint.Hash) if err != nil { // Avoid returning an error if the transaction was not found to // proceed with fallback methods. jsonErr, ok := err.(*btcjson.RPCError) if !ok || jsonErr.Code != btcjson.ErrRPCNoTxInfo { return nil, fmt.Errorf("unable to query for txid %v: %v", - spendRequest.OutPoint.Hash, err) + outpoint.Hash, err) } } @@ -740,15 +729,15 @@ func (b *BitcoindNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, return nil, err } - if uint32(blockHeight) > historicalDispatch.StartHeight { - historicalDispatch.StartHeight = uint32(blockHeight) + if uint32(blockHeight) > ntfn.HistoricalDispatch.StartHeight { + ntfn.HistoricalDispatch.StartHeight = uint32(blockHeight) } } // Now that we've determined the starting point of our rescan, we can // dispatch it and return. select { - case b.notificationRegistry <- historicalDispatch: + case b.notificationRegistry <- ntfn.HistoricalDispatch: case <-b.quit: return nil, chainntnfs.ErrChainNotifierShuttingDown } diff --git a/chainntnfs/btcdnotify/btcd.go b/chainntnfs/btcdnotify/btcd.go index c174f4c85..cdd1e7789 100644 --- a/chainntnfs/btcdnotify/btcd.go +++ b/chainntnfs/btcdnotify/btcd.go @@ -11,6 +11,7 @@ import ( "github.com/btcsuite/btcd/chaincfg" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/rpcclient" + "github.com/btcsuite/btcd/txscript" "github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcutil" "github.com/lightningnetwork/lnd/chainntnfs" @@ -50,7 +51,6 @@ type txUpdate struct { // notifications. Multiple concurrent clients are supported. All notifications // are achieved via non-blocking sends on client channels. type BtcdNotifier struct { - spendClientCounter uint64 // To be used atomically. epochClientCounter uint64 // To be used atomically. started int32 // To be used atomically. @@ -651,23 +651,11 @@ func (b *BtcdNotifier) notifyBlockEpochClient(epochClient *blockEpochRegistratio func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, error) { - // First, we'll construct a spend notification request and hand it off - // to the txNotifier. - spendID := atomic.AddUint64(&b.spendClientCounter, 1) - spendRequest, err := chainntnfs.NewSpendRequest(outpoint, pkScript) - if err != nil { - return nil, err - } - ntfn := &chainntnfs.SpendNtfn{ - SpendID: spendID, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(func() { - b.txNotifier.CancelSpend(spendRequest, spendID) - }), - HeightHint: heightHint, - } - - historicalDispatch, _, err := b.txNotifier.RegisterSpend(ntfn) + // Register the conf notification with the TxNotifier. A non-nil value + // for `dispatch` will be returned if we are required to perform a + // manual scan for the confirmation. Otherwise the notifier will begin + // watching at tip for the transaction to confirm. + ntfn, err := b.txNotifier.RegisterSpend(outpoint, pkScript, heightHint) if err != nil { return nil, err } @@ -676,17 +664,18 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // outpoint/output script as spent. // // TODO(wilmer): use LoadFilter API instead. - if spendRequest.OutPoint == chainntnfs.ZeroOutPoint { - addr, err := spendRequest.PkScript.Address(b.chainParams) + if outpoint == nil || *outpoint == chainntnfs.ZeroOutPoint { + _, addrs, _, err := txscript.ExtractPkScriptAddrs( + pkScript, b.chainParams, + ) if err != nil { - return nil, err + return nil, fmt.Errorf("unable to parse script: %v", err) } - addrs := []btcutil.Address{addr} if err := b.chainConn.NotifyReceived(addrs); err != nil { return nil, err } } else { - ops := []*wire.OutPoint{&spendRequest.OutPoint} + ops := []*wire.OutPoint{outpoint} if err := b.chainConn.NotifySpent(ops); err != nil { return nil, err } @@ -695,7 +684,7 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // If the txNotifier didn't return any details to perform a historical // scan of the chain, then we can return early as there's nothing left // for us to do. - if historicalDispatch == nil { + if ntfn.HistoricalDispatch == nil { return ntfn.Event, nil } @@ -705,26 +694,29 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // We'll short-circuit the path when dispatching the spend of a script, // rather than an outpoint, as there aren't any additional checks we can // make for scripts. - if spendRequest.OutPoint == chainntnfs.ZeroOutPoint { + if outpoint == nil || *outpoint == chainntnfs.ZeroOutPoint { startHash, err := b.chainConn.GetBlockHash( - int64(historicalDispatch.StartHeight), + int64(ntfn.HistoricalDispatch.StartHeight), ) if err != nil { return nil, err } // TODO(wilmer): add retry logic if rescan fails? - addr, err := spendRequest.PkScript.Address(b.chainParams) + _, addrs, _, err := txscript.ExtractPkScriptAddrs( + pkScript, b.chainParams, + ) if err != nil { - return nil, err + return nil, fmt.Errorf("unable to parse address: %v", err) } - addrs := []btcutil.Address{addr} + asyncResult := b.chainConn.RescanAsync(startHash, addrs, nil) go func() { if rescanErr := asyncResult.Receive(); rescanErr != nil { chainntnfs.Log.Errorf("Rescan to determine "+ "the spend details of %v failed: %v", - spendRequest, rescanErr) + ntfn.HistoricalDispatch.SpendRequest, + rescanErr) } }() @@ -738,16 +730,16 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // We'll start by checking the backend's UTXO set to determine whether // the outpoint has been spent. If it hasn't, we can return to the // caller as well. - txOut, err := b.chainConn.GetTxOut( - &spendRequest.OutPoint.Hash, spendRequest.OutPoint.Index, true, - ) + txOut, err := b.chainConn.GetTxOut(&outpoint.Hash, outpoint.Index, true) if err != nil { return nil, err } if txOut != nil { // We'll let the txNotifier know the outpoint is still unspent // in order to begin updating its spend hint. - err := b.txNotifier.UpdateSpendDetails(spendRequest, nil) + err := b.txNotifier.UpdateSpendDetails( + ntfn.HistoricalDispatch.SpendRequest, nil, + ) if err != nil { return nil, err } @@ -759,25 +751,25 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // set, we'll determine when it happened by scanning the chain. We'll // begin by fetching the block hash of our starting height. startHash, err := b.chainConn.GetBlockHash( - int64(historicalDispatch.StartHeight), + int64(ntfn.HistoricalDispatch.StartHeight), ) if err != nil { return nil, fmt.Errorf("unable to get block hash for height "+ - "%d: %v", historicalDispatch.StartHeight, err) + "%d: %v", ntfn.HistoricalDispatch.StartHeight, err) } // As a minimal optimization, we'll query the backend's transaction // index (if enabled) to determine if we have a better rescan starting // height. We can do this as the GetRawTransaction call will return the // hash of the block it was included in within the chain. - tx, err := b.chainConn.GetRawTransactionVerbose(&spendRequest.OutPoint.Hash) + tx, err := b.chainConn.GetRawTransactionVerbose(&outpoint.Hash) if err != nil { // Avoid returning an error if the transaction was not found to // proceed with fallback methods. jsonErr, ok := err.(*btcjson.RPCError) if !ok || jsonErr.Code != btcjson.ErrRPCNoTxInfo { return nil, fmt.Errorf("unable to query for txid %v: %v", - spendRequest.OutPoint.Hash, err) + outpoint.Hash, err) } } @@ -801,7 +793,7 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, "block %v: %v", blockHash, err) } - if uint32(blockHeader.Height) > historicalDispatch.StartHeight { + if uint32(blockHeader.Height) > ntfn.HistoricalDispatch.StartHeight { startHash, err = b.chainConn.GetBlockHash( int64(blockHeader.Height), ) @@ -824,13 +816,12 @@ func (b *BtcdNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // // TODO(wilmer): add retry logic if rescan fails? asyncResult := b.chainConn.RescanAsync( - startHash, nil, []*wire.OutPoint{&spendRequest.OutPoint}, + startHash, nil, []*wire.OutPoint{outpoint}, ) go func() { if rescanErr := asyncResult.Receive(); rescanErr != nil { chainntnfs.Log.Errorf("Rescan to determine the spend "+ - "details of %v failed: %v", spendRequest, - rescanErr) + "details of %v failed: %v", outpoint, rescanErr) } }() diff --git a/chainntnfs/neutrinonotify/neutrino.go b/chainntnfs/neutrinonotify/neutrino.go index 76f6dcbcd..ab9c51e9d 100644 --- a/chainntnfs/neutrinonotify/neutrino.go +++ b/chainntnfs/neutrinonotify/neutrino.go @@ -37,7 +37,6 @@ const ( // TODO(roasbeef): heavily consolidate with NeutrinoNotifier code // * maybe combine into single package? type NeutrinoNotifier struct { - spendClientCounter uint64 // To be used atomically. epochClientCounter uint64 // To be used atomically. started int32 // To be used atomically. @@ -662,23 +661,11 @@ func (n *NeutrinoNotifier) notifyBlockEpochClient(epochClient *blockEpochRegistr func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, error) { - // First, we'll construct a spend notification request and hand it off - // to the txNotifier. - spendID := atomic.AddUint64(&n.spendClientCounter, 1) - spendRequest, err := chainntnfs.NewSpendRequest(outpoint, pkScript) - if err != nil { - return nil, err - } - ntfn := &chainntnfs.SpendNtfn{ - SpendID: spendID, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(func() { - n.txNotifier.CancelSpend(spendRequest, spendID) - }), - HeightHint: heightHint, - } - - historicalDispatch, txNotifierTip, err := n.txNotifier.RegisterSpend(ntfn) + // Register the conf notification with the TxNotifier. A non-nil value + // for `dispatch` will be returned if we are required to perform a + // manual scan for the confirmation. Otherwise the notifier will begin + // watching at tip for the transaction to confirm. + ntfn, err := n.txNotifier.RegisterSpend(outpoint, pkScript, heightHint) if err != nil { return nil, err } @@ -690,9 +677,12 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // // We'll update our filter first to ensure we can immediately detect the // spend at tip. + if outpoint == nil { + outpoint = &chainntnfs.ZeroOutPoint + } inputToWatch := neutrino.InputWithScript{ - OutPoint: spendRequest.OutPoint, - PkScript: spendRequest.PkScript.Script(), + OutPoint: *outpoint, + PkScript: pkScript, } updateOptions := []neutrino.UpdateOption{ neutrino.AddInputs(inputToWatch), @@ -703,10 +693,9 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // update. In the case of an output script spend request, we'll check if // we should perform a historical rescan and start from there, as we // cannot do so with GetUtxo since it matches outpoints. - rewindHeight := txNotifierTip - if historicalDispatch != nil && - spendRequest.OutPoint == chainntnfs.ZeroOutPoint { - rewindHeight = historicalDispatch.StartHeight + rewindHeight := ntfn.Height + if ntfn.HistoricalDispatch != nil && *outpoint == chainntnfs.ZeroOutPoint { + rewindHeight = ntfn.HistoricalDispatch.StartHeight } updateOptions = append(updateOptions, neutrino.Rewind(rewindHeight)) @@ -733,8 +722,7 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // scan of the chain, or if we already performed one like in the case of // output script spend requests, then we can return early as there's // nothing left for us to do. - if historicalDispatch == nil || - spendRequest.OutPoint == chainntnfs.ZeroOutPoint { + if ntfn.HistoricalDispatch == nil || *outpoint == chainntnfs.ZeroOutPoint { return ntfn.Event, nil } @@ -752,7 +740,7 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, currentHeight := uint32(n.bestBlock.Height) n.bestBlockMtx.RUnlock() - if currentHeight >= historicalDispatch.StartHeight { + if currentHeight >= ntfn.HistoricalDispatch.StartHeight { break } @@ -766,10 +754,10 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, spendReport, err := n.p2pNode.GetUtxo( neutrino.WatchInputs(inputToWatch), neutrino.StartBlock(&waddrmgr.BlockStamp{ - Height: int32(historicalDispatch.StartHeight), + Height: int32(ntfn.HistoricalDispatch.StartHeight), }), neutrino.EndBlock(&waddrmgr.BlockStamp{ - Height: int32(historicalDispatch.EndHeight), + Height: int32(ntfn.HistoricalDispatch.EndHeight), }), neutrino.QuitChan(n.quit), ) @@ -784,7 +772,7 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, if spendReport != nil && spendReport.SpendingTx != nil { spendingTxHash := spendReport.SpendingTx.TxHash() spendDetails = &chainntnfs.SpendDetail{ - SpentOutPoint: &spendRequest.OutPoint, + SpentOutPoint: outpoint, SpenderTxHash: &spendingTxHash, SpendingTx: spendReport.SpendingTx, SpenderInputIndex: spendReport.SpendingInputIndex, @@ -796,7 +784,9 @@ func (n *NeutrinoNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint, // not, we'll mark our historical rescan as complete to ensure the // outpoint's spend hint gets updated upon connected/disconnected // blocks. - err = n.txNotifier.UpdateSpendDetails(spendRequest, spendDetails) + err = n.txNotifier.UpdateSpendDetails( + ntfn.HistoricalDispatch.SpendRequest, spendDetails, + ) if err != nil { chainntnfs.Log.Errorf("Failed to update spend details: %v", err) return diff --git a/chainntnfs/txnotifier.go b/chainntnfs/txnotifier.go index 5e503f217..82961ab89 100644 --- a/chainntnfs/txnotifier.go +++ b/chainntnfs/txnotifier.go @@ -415,13 +415,33 @@ type HistoricalSpendDispatch struct { EndHeight uint32 } +// SpendRegistration encompasses all of the information required for callers to +// retrieve details about a spend event. +type SpendRegistration struct { + // Event contains references to the channels that the notifications are + // to be sent over. + Event *SpendEvent + + // HistoricalDispatch, if non-nil, signals to the client who registered + // the notification that they are responsible for attempting to manually + // rescan blocks for the txid/output script between the start and end + // heights. + HistoricalDispatch *HistoricalSpendDispatch + + // Height is the height of the TxNotifier at the time the spend + // notification was registered. This can be used so that backends can + // request to be notified of spends from this point forwards. + Height uint32 +} + // TxNotifier is a struct responsible for delivering transaction notifications // to subscribers. These notifications can be of two different types: // transaction/output script confirmations and/or outpoint/output script spends. // The TxNotifier will watch the blockchain as new blocks come in, in order to // satisfy its client requests. type TxNotifier struct { - confClientCounter uint64 // To be used atomically. + confClientCounter uint64 // To be used atomically. + spendClientCounter uint64 // To be used atomically. // currentHeight is the height of the tracked blockchain. It is used to // determine the number of confirmations a tx has and ensure blocks are @@ -885,29 +905,50 @@ func (n *TxNotifier) dispatchConfDetails( return nil } +// newSpendNtfn validates all of the parameters required to successfully create +// and register a spend notification. +func (n *TxNotifier) newSpendNtfn(outpoint *wire.OutPoint, + pkScript []byte, heightHint uint32) (*SpendNtfn, error) { + + spendRequest, err := NewSpendRequest(outpoint, pkScript) + if err != nil { + return nil, err + } + + spendID := atomic.AddUint64(&n.spendClientCounter, 1) + return &SpendNtfn{ + SpendID: spendID, + SpendRequest: spendRequest, + Event: NewSpendEvent(func() { + n.CancelSpend(spendRequest, spendID) + }), + HeightHint: heightHint, + }, nil +} + // RegisterSpend handles a new spend notification request. The client will be // notified once the outpoint/output script is detected as spent within the // chain. // -// The registration succeeds if no error is returned. If the returned -// HistoricalSpendDisaptch is non-nil, the caller is responsible for attempting -// to determine whether the outpoint/output script has been spent between the -// start and end heights. The notifier's current height is also returned so that -// backends can request to be notified of spends from this point forwards. -// // NOTE: If the outpoint/output script has already been spent within the chain // before the notifier's current tip, the spend details must be provided with // the UpdateSpendDetails method, otherwise we will wait for the outpoint/output // script to be spent at tip, even though it already has. -func (n *TxNotifier) RegisterSpend(ntfn *SpendNtfn) (*HistoricalSpendDispatch, - uint32, error) { +func (n *TxNotifier) RegisterSpend(outpoint *wire.OutPoint, pkScript []byte, + heightHint uint32) (*SpendRegistration, error) { select { case <-n.quit: - return nil, 0, ErrTxNotifierExiting + return nil, ErrTxNotifierExiting default: } + // We'll start by performing a series of validation checks. + ntfn, err := n.newSpendNtfn(outpoint, pkScript, heightHint) + if err != nil { + return nil, err + } + // Before proceeding to register the notification, we'll query our spend // hint cache to determine whether a better one exists. startHeight := ntfn.HeightHint @@ -953,9 +994,16 @@ func (n *TxNotifier) RegisterSpend(ntfn *SpendNtfn) (*HistoricalSpendDispatch, "registration since rescan has finished", ntfn.SpendRequest) - return nil, n.currentHeight, n.dispatchSpendDetails( - ntfn, spendSet.details, - ) + err := n.dispatchSpendDetails(ntfn, spendSet.details) + if err != nil { + return nil, err + } + + return &SpendRegistration{ + Event: ntfn.Event, + HistoricalDispatch: nil, + Height: n.currentHeight, + }, nil // If there is an active rescan to determine whether the request has // been spent, then we won't trigger another one. @@ -963,7 +1011,11 @@ func (n *TxNotifier) RegisterSpend(ntfn *SpendNtfn) (*HistoricalSpendDispatch, Log.Debugf("Waiting for pending rescan to finish before "+ "notifying %v at tip", ntfn.SpendRequest) - return nil, n.currentHeight, nil + return &SpendRegistration{ + Event: ntfn.Event, + HistoricalDispatch: nil, + Height: n.currentHeight, + }, nil // Otherwise, we'll fall through and let the caller know that a rescan // should be dispatched to determine whether the request has already @@ -983,7 +1035,11 @@ func (n *TxNotifier) RegisterSpend(ntfn *SpendNtfn) (*HistoricalSpendDispatch, // spend hints for this request get updated upon // connected/disconnected blocks. spendSet.rescanStatus = rescanComplete - return nil, n.currentHeight, nil + return &SpendRegistration{ + Event: ntfn.Event, + HistoricalDispatch: nil, + Height: n.currentHeight, + }, nil } // We'll set the rescan status to pending to ensure subsequent @@ -993,11 +1049,15 @@ func (n *TxNotifier) RegisterSpend(ntfn *SpendNtfn) (*HistoricalSpendDispatch, Log.Debugf("Dispatching historical spend rescan for %v", ntfn.SpendRequest) - return &HistoricalSpendDispatch{ - SpendRequest: ntfn.SpendRequest, - StartHeight: startHeight, - EndHeight: n.currentHeight, - }, n.currentHeight, nil + return &SpendRegistration{ + Event: ntfn.Event, + HistoricalDispatch: &HistoricalSpendDispatch{ + SpendRequest: ntfn.SpendRequest, + StartHeight: startHeight, + EndHeight: n.currentHeight, + }, + Height: n.currentHeight, + }, nil } // CancelSpend cancels an existing request for a spend notification of an diff --git a/chainntnfs/txnotifier_test.go b/chainntnfs/txnotifier_test.go index ecfc327f7..f79fd59d9 100644 --- a/chainntnfs/txnotifier_test.go +++ b/chainntnfs/txnotifier_test.go @@ -479,14 +479,9 @@ func TestTxNotifierFutureSpendDispatch(t *testing.T) { // We'll start off by registering for a spend notification of an // outpoint. - ntfn := &chainntnfs.SpendNtfn{ - SpendRequest: chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 1}, - PkScript: testScript, - }, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(ntfn); err != nil { + op := wire.OutPoint{Index: 1} + ntfn, err := n.RegisterSpend(&op, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } @@ -503,14 +498,14 @@ func TestTxNotifierFutureSpendDispatch(t *testing.T) { // spend notification. spendTx := wire.NewMsgTx(2) spendTx.AddTxIn(&wire.TxIn{ - PreviousOutPoint: ntfn.OutPoint, + PreviousOutPoint: op, SignatureScript: testSigScript, }) spendTxHash := spendTx.TxHash() block := btcutil.NewBlock(&wire.MsgBlock{ Transactions: []*wire.MsgTx{spendTx}, }) - err := n.ConnectTip(block.Hash(), 11, block.Transactions()) + err = n.ConnectTip(block.Hash(), 11, block.Transactions()) if err != nil { t.Fatalf("unable to connect block: %v", err) } @@ -519,7 +514,7 @@ func TestTxNotifierFutureSpendDispatch(t *testing.T) { } expectedSpendDetails := &chainntnfs.SpendDetail{ - SpentOutPoint: &ntfn.OutPoint, + SpentOutPoint: &op, SpenderTxHash: &spendTxHash, SpendingTx: spendTx, SpenderInputIndex: 0, @@ -593,11 +588,8 @@ func TestTxNotifierHistoricalSpendDispatch(t *testing.T) { // We'll register for a spend notification of the outpoint and ensure // that a notification isn't dispatched. - ntfn := &chainntnfs.SpendNtfn{ - SpendRequest: chainntnfs.SpendRequest{OutPoint: spentOutpoint}, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(ntfn); err != nil { + ntfn, err := n.RegisterSpend(&spentOutpoint, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } @@ -611,7 +603,9 @@ func TestTxNotifierHistoricalSpendDispatch(t *testing.T) { // we'll hand off the spending details of the outpoint to the notifier // as it is not possible for it to view historical events in the chain. // By doing this, we replicate the functionality of the ChainNotifier. - err := n.UpdateSpendDetails(ntfn.SpendRequest, expectedSpendDetails) + err = n.UpdateSpendDetails( + ntfn.HistoricalDispatch.SpendRequest, expectedSpendDetails, + ) if err != nil { t.Fatalf("unable to update spend details: %v", err) } @@ -721,35 +715,23 @@ func TestTxNotifierMultipleHistoricalSpendRescans(t *testing.T) { // The first registration for an outpoint in the notifier should request // a historical spend rescan as it does not have a historical view of // the chain. - spendRequest := chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 1}, - } - ntfn1 := &chainntnfs.SpendNtfn{ - SpendID: 0, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(nil), - } - historicalDispatch1, _, err := n.RegisterSpend(ntfn1) + op := wire.OutPoint{Index: 1} + ntfn1, err := n.RegisterSpend(&op, testRawScript, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalDispatch1 == nil { + if ntfn1.HistoricalDispatch == nil { t.Fatal("expected to receive historical dispatch request") } // We'll register another spend notification for the same outpoint. This // should not request a historical spend rescan since the first one is // still pending. - ntfn2 := &chainntnfs.SpendNtfn{ - SpendID: 1, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(nil), - } - historicalDispatch2, _, err := n.RegisterSpend(ntfn2) + ntfn2, err := n.RegisterSpend(&op, testRawScript, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalDispatch2 != nil { + if ntfn2.HistoricalDispatch != nil { t.Fatal("received unexpected historical rescan request") } @@ -758,27 +740,24 @@ func TestTxNotifierMultipleHistoricalSpendRescans(t *testing.T) { // historical rescan request since the confirmation details should be // cached. spendDetails := &chainntnfs.SpendDetail{ - SpentOutPoint: &ntfn2.OutPoint, + SpentOutPoint: &op, SpenderTxHash: &chainntnfs.ZeroHash, SpendingTx: wire.NewMsgTx(2), SpenderInputIndex: 0, SpendingHeight: startingHeight - 1, } - err = n.UpdateSpendDetails(ntfn2.SpendRequest, spendDetails) + err = n.UpdateSpendDetails( + ntfn1.HistoricalDispatch.SpendRequest, spendDetails, + ) if err != nil { t.Fatalf("unable to update spend details: %v", err) } - ntfn3 := &chainntnfs.SpendNtfn{ - SpendID: 2, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(nil), - } - historicalDispatch3, _, err := n.RegisterSpend(ntfn3) + ntfn3, err := n.RegisterSpend(&op, testRawScript, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalDispatch3 != nil { + if ntfn3.HistoricalDispatch != nil { t.Fatal("received unexpected historical rescan request") } } @@ -872,19 +851,14 @@ func TestTxNotifierMultipleHistoricalNtfns(t *testing.T) { } // Similarly, we'll do the same thing but for spend notifications. - spendRequest := chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 1}, - } - spendNtfns := make([]*chainntnfs.SpendNtfn, numNtfns) + op := wire.OutPoint{Index: 1} + spendNtfns := make([]*chainntnfs.SpendRegistration, numNtfns) for i := uint64(0); i < numNtfns; i++ { - spendNtfns[i] = &chainntnfs.SpendNtfn{ - SpendID: i, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(spendNtfns[i]); err != nil { + ntfn, err := n.RegisterSpend(&op, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn #%d: %v", i, err) } + spendNtfns[i] = ntfn } // Ensure none of them have received the spend details. @@ -901,13 +875,15 @@ func TestTxNotifierMultipleHistoricalNtfns(t *testing.T) { // following spend details. We'll let the notifier know so that it can // stop watching at tip. expectedSpendDetails := &chainntnfs.SpendDetail{ - SpentOutPoint: &spendNtfns[0].OutPoint, + SpentOutPoint: &op, SpenderTxHash: &chainntnfs.ZeroHash, SpendingTx: wire.NewMsgTx(2), SpenderInputIndex: 0, SpendingHeight: startingHeight - 1, } - err = n.UpdateSpendDetails(spendNtfns[0].SpendRequest, expectedSpendDetails) + err = n.UpdateSpendDetails( + spendNtfns[0].HistoricalDispatch.SpendRequest, expectedSpendDetails, + ) if err != nil { t.Fatalf("unable to update spend details: %v", err) } @@ -928,16 +904,11 @@ func TestTxNotifierMultipleHistoricalNtfns(t *testing.T) { // cached, we'll register another client for the same outpoint. We // should not see a historical rescan request and the spend notification // should come through immediately. - extraSpendNtfn := &chainntnfs.SpendNtfn{ - SpendID: numNtfns + 1, - SpendRequest: spendRequest, - Event: chainntnfs.NewSpendEvent(nil), - } - historicalSpendRescan, _, err := n.RegisterSpend(extraSpendNtfn) + extraSpendNtfn, err := n.RegisterSpend(&op, testRawScript, 1) if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - if historicalSpendRescan != nil { + if extraSpendNtfn.HistoricalDispatch != nil { t.Fatal("received unexpected historical rescan request") } @@ -1035,26 +1006,15 @@ func TestTxNotifierCancelSpend(t *testing.T) { // We'll register two notification requests. Only the second one will be // canceled. - ntfn1 := &chainntnfs.SpendNtfn{ - SpendID: 0, - SpendRequest: chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 1}, - PkScript: testScript, - }, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(ntfn1); err != nil { + op1 := wire.OutPoint{Index: 1} + ntfn1, err := n.RegisterSpend(&op1, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - ntfn2 := &chainntnfs.SpendNtfn{ - SpendID: 1, - SpendRequest: chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 2}, - }, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(ntfn2); err != nil { + op2 := wire.OutPoint{Index: 2} + ntfn2, err := n.RegisterSpend(&op2, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } @@ -1062,12 +1022,12 @@ func TestTxNotifierCancelSpend(t *testing.T) { // block containing it. spendTx := wire.NewMsgTx(2) spendTx.AddTxIn(&wire.TxIn{ - PreviousOutPoint: ntfn1.OutPoint, + PreviousOutPoint: op1, SignatureScript: testSigScript, }) spendTxHash := spendTx.TxHash() expectedSpendDetails := &chainntnfs.SpendDetail{ - SpentOutPoint: &ntfn1.OutPoint, + SpentOutPoint: &op1, SpenderTxHash: &spendTxHash, SpendingTx: spendTx, SpenderInputIndex: 0, @@ -1080,9 +1040,9 @@ func TestTxNotifierCancelSpend(t *testing.T) { // Before extending the notifier's tip with the dummy block above, we'll // cancel the second request. - n.CancelSpend(ntfn2.SpendRequest, ntfn2.SpendID) + n.CancelSpend(ntfn2.HistoricalDispatch.SpendRequest, 2) - err := n.ConnectTip(block.Hash(), startingHeight+1, block.Transactions()) + err = n.ConnectTip(block.Hash(), startingHeight+1, block.Transactions()) if err != nil { t.Fatalf("unable to connect block: %v", err) } @@ -1404,35 +1364,29 @@ func TestTxNotifierSpendReorg(t *testing.T) { // We'll have two outpoints that will be spent throughout the test. The // first will be spent and will not experience a reorg, while the second // one will. - spendRequest1 := chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 1}, - PkScript: testScript, - } + op1 := wire.OutPoint{Index: 1} spendTx1 := wire.NewMsgTx(2) spendTx1.AddTxIn(&wire.TxIn{ - PreviousOutPoint: spendRequest1.OutPoint, + PreviousOutPoint: op1, SignatureScript: testSigScript, }) spendTxHash1 := spendTx1.TxHash() expectedSpendDetails1 := &chainntnfs.SpendDetail{ - SpentOutPoint: &spendRequest1.OutPoint, + SpentOutPoint: &op1, SpenderTxHash: &spendTxHash1, SpendingTx: spendTx1, SpenderInputIndex: 0, SpendingHeight: startingHeight + 1, } - spendRequest2 := chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 2}, - PkScript: testScript, - } + op2 := wire.OutPoint{Index: 2} spendTx2 := wire.NewMsgTx(2) spendTx2.AddTxIn(&wire.TxIn{ PreviousOutPoint: chainntnfs.ZeroOutPoint, SignatureScript: testSigScript, }) spendTx2.AddTxIn(&wire.TxIn{ - PreviousOutPoint: spendRequest2.OutPoint, + PreviousOutPoint: op2, SignatureScript: testSigScript, }) spendTxHash2 := spendTx2.TxHash() @@ -1441,7 +1395,7 @@ func TestTxNotifierSpendReorg(t *testing.T) { // different height, so we'll need to construct the spend details for // before and after the reorg. expectedSpendDetails2BeforeReorg := chainntnfs.SpendDetail{ - SpentOutPoint: &spendRequest2.OutPoint, + SpentOutPoint: &op2, SpenderTxHash: &spendTxHash2, SpendingTx: spendTx2, SpenderInputIndex: 1, @@ -1454,21 +1408,13 @@ func TestTxNotifierSpendReorg(t *testing.T) { expectedSpendDetails2AfterReorg.SpendingHeight++ // We'll register for a spend notification for each outpoint above. - ntfn1 := &chainntnfs.SpendNtfn{ - SpendID: 78, - SpendRequest: spendRequest1, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(ntfn1); err != nil { + ntfn1, err := n.RegisterSpend(&op1, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } - ntfn2 := &chainntnfs.SpendNtfn{ - SpendID: 21, - SpendRequest: spendRequest2, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(ntfn2); err != nil { + ntfn2, err := n.RegisterSpend(&op2, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } @@ -1477,7 +1423,7 @@ func TestTxNotifierSpendReorg(t *testing.T) { block1 := btcutil.NewBlock(&wire.MsgBlock{ Transactions: []*wire.MsgTx{spendTx1}, }) - err := n.ConnectTip(block1.Hash(), startingHeight+1, block1.Transactions()) + err = n.ConnectTip(block1.Hash(), startingHeight+1, block1.Transactions()) if err != nil { t.Fatalf("unable to connect block: %v", err) } @@ -1836,40 +1782,27 @@ func TestTxNotifierSpendHintCache(t *testing.T) { ) // Create two test outpoints and register them for spend notifications. - ntfn1 := &chainntnfs.SpendNtfn{ - SpendID: 1, - SpendRequest: chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 1}, - PkScript: testScript, - }, - Event: chainntnfs.NewSpendEvent(nil), - } - ntfn2 := &chainntnfs.SpendNtfn{ - SpendID: 2, - SpendRequest: chainntnfs.SpendRequest{ - OutPoint: wire.OutPoint{Index: 2}, - PkScript: testScript, - }, - Event: chainntnfs.NewSpendEvent(nil), - } - - if _, _, err := n.RegisterSpend(ntfn1); err != nil { + op1 := wire.OutPoint{Index: 1} + ntfn1, err := n.RegisterSpend(&op1, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend for op1: %v", err) } - if _, _, err := n.RegisterSpend(ntfn2); err != nil { + op2 := wire.OutPoint{Index: 2} + ntfn2, err := n.RegisterSpend(&op2, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend for op2: %v", err) } // Both outpoints should not have a spend hint set upon registration, as // we must first determine whether they have already been spent in the // chain. - _, err := hintCache.QuerySpendHint(ntfn1.SpendRequest) + _, err = hintCache.QuerySpendHint(ntfn1.HistoricalDispatch.SpendRequest) if err != chainntnfs.ErrSpendHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "expected: %v, got %v", chainntnfs.ErrSpendHintNotFound, err) } - _, err = hintCache.QuerySpendHint(ntfn2.SpendRequest) + _, err = hintCache.QuerySpendHint(ntfn2.HistoricalDispatch.SpendRequest) if err != chainntnfs.ErrSpendHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "expected: %v, got %v", chainntnfs.ErrSpendHintNotFound, @@ -1891,13 +1824,13 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // Since we haven't called UpdateSpendDetails on any of the test // outpoints, this implies that there is a still a pending historical // rescan for them, so their spend hints should not be created/updated. - _, err = hintCache.QuerySpendHint(ntfn1.SpendRequest) + _, err = hintCache.QuerySpendHint(ntfn1.HistoricalDispatch.SpendRequest) if err != chainntnfs.ErrSpendHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "expected: %v, got %v", chainntnfs.ErrSpendHintNotFound, err) } - _, err = hintCache.QuerySpendHint(ntfn2.SpendRequest) + _, err = hintCache.QuerySpendHint(ntfn2.HistoricalDispatch.SpendRequest) if err != chainntnfs.ErrSpendHintNotFound { t.Fatalf("unexpected error when querying for height hint "+ "expected: %v, got %v", chainntnfs.ErrSpendHintNotFound, @@ -1907,10 +1840,12 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // Now, we'll simulate that their historical rescans have finished by // calling UpdateSpendDetails. This should allow their spend hints to be // updated upon every block connected/disconnected. - if err := n.UpdateSpendDetails(ntfn1.SpendRequest, nil); err != nil { + err = n.UpdateSpendDetails(ntfn1.HistoricalDispatch.SpendRequest, nil) + if err != nil { t.Fatalf("unable to update spend details: %v", err) } - if err := n.UpdateSpendDetails(ntfn2.SpendRequest, nil); err != nil { + err = n.UpdateSpendDetails(ntfn2.HistoricalDispatch.SpendRequest, nil) + if err != nil { t.Fatalf("unable to update spend details: %v", err) } @@ -1918,7 +1853,7 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // of the first outpoint. spendTx1 := wire.NewMsgTx(2) spendTx1.AddTxIn(&wire.TxIn{ - PreviousOutPoint: ntfn1.OutPoint, + PreviousOutPoint: op1, SignatureScript: testSigScript, }) block1 := btcutil.NewBlock(&wire.MsgBlock{ @@ -1935,14 +1870,14 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // Both outpoints should have their spend hints reflect the height of // the new block being connected due to the first outpoint being spent // at this height, and the second outpoint still being unspent. - op1Hint, err := hintCache.QuerySpendHint(ntfn1.SpendRequest) + op1Hint, err := hintCache.QuerySpendHint(ntfn1.HistoricalDispatch.SpendRequest) if err != nil { t.Fatalf("unable to query for spend hint of op1: %v", err) } if op1Hint != op1Height { t.Fatalf("expected hint %d, got %d", op1Height, op1Hint) } - op2Hint, err := hintCache.QuerySpendHint(ntfn2.SpendRequest) + op2Hint, err := hintCache.QuerySpendHint(ntfn2.HistoricalDispatch.SpendRequest) if err != nil { t.Fatalf("unable to query for spend hint of op2: %v", err) } @@ -1953,7 +1888,7 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // Then, we'll create another block that spends the second outpoint. spendTx2 := wire.NewMsgTx(2) spendTx2.AddTxIn(&wire.TxIn{ - PreviousOutPoint: ntfn2.OutPoint, + PreviousOutPoint: op2, SignatureScript: testSigScript, }) block2 := btcutil.NewBlock(&wire.MsgBlock{ @@ -1970,14 +1905,14 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // Only the second outpoint should have its spend hint updated due to // being spent within the new block. The first outpoint's spend hint // should remain the same as it's already been spent before. - op1Hint, err = hintCache.QuerySpendHint(ntfn1.SpendRequest) + op1Hint, err = hintCache.QuerySpendHint(ntfn1.HistoricalDispatch.SpendRequest) if err != nil { t.Fatalf("unable to query for spend hint of op1: %v", err) } if op1Hint != op1Height { t.Fatalf("expected hint %d, got %d", op1Height, op1Hint) } - op2Hint, err = hintCache.QuerySpendHint(ntfn2.SpendRequest) + op2Hint, err = hintCache.QuerySpendHint(ntfn2.HistoricalDispatch.SpendRequest) if err != nil { t.Fatalf("unable to query for spend hint of op2: %v", err) } @@ -1995,14 +1930,14 @@ func TestTxNotifierSpendHintCache(t *testing.T) { // to the previous height, as that's where its spending transaction was // included in within the chain. The first outpoint's spend hint should // remain the same. - op1Hint, err = hintCache.QuerySpendHint(ntfn1.SpendRequest) + op1Hint, err = hintCache.QuerySpendHint(ntfn1.HistoricalDispatch.SpendRequest) if err != nil { t.Fatalf("unable to query for spend hint of op1: %v", err) } if op1Hint != op1Height { t.Fatalf("expected hint %d, got %d", op1Height, op1Hint) } - op2Hint, err = hintCache.QuerySpendHint(ntfn2.SpendRequest) + op2Hint, err = hintCache.QuerySpendHint(ntfn2.HistoricalDispatch.SpendRequest) if err != nil { t.Fatalf("unable to query for spend hint of op2: %v", err) } @@ -2027,16 +1962,8 @@ func TestTxNotifierNtfnDone(t *testing.T) { if err != nil { t.Fatalf("unable to register conf ntfn: %v", err) } - - spendNtfn := &chainntnfs.SpendNtfn{ - SpendID: 2, - SpendRequest: chainntnfs.SpendRequest{ - OutPoint: chainntnfs.ZeroOutPoint, - PkScript: testScript, - }, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(spendNtfn); err != nil { + spendNtfn, err := n.RegisterSpend(&chainntnfs.ZeroOutPoint, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend: %v", err) } @@ -2165,13 +2092,8 @@ func TestTxNotifierTearDown(t *testing.T) { if err != nil { t.Fatalf("unable to register conf ntfn: %v", err) } - - spendNtfn := &chainntnfs.SpendNtfn{ - SpendID: 1, - SpendRequest: chainntnfs.SpendRequest{OutPoint: chainntnfs.ZeroOutPoint}, - Event: chainntnfs.NewSpendEvent(nil), - } - if _, _, err := n.RegisterSpend(spendNtfn); err != nil { + spendNtfn, err := n.RegisterSpend(&chainntnfs.ZeroOutPoint, testRawScript, 1) + if err != nil { t.Fatalf("unable to register spend ntfn: %v", err) } @@ -2212,7 +2134,8 @@ func TestTxNotifierTearDown(t *testing.T) { if err == nil { t.Fatal("expected confirmation registration to fail") } - if _, _, err := n.RegisterSpend(spendNtfn); err == nil { + _, err = n.RegisterSpend(&chainntnfs.ZeroOutPoint, testRawScript, 1) + if err == nil { t.Fatal("expected spend registration to fail") } } From 17b6205f3a7b06b2db5f7807420629211ecb86ad Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 16 Aug 2019 19:57:26 -0700 Subject: [PATCH 3/9] chainntnfs: validate conf/spend ntfn registration parameters A height hint not being set would cause lnd to scan for the confirmation/spend of a txid/outpoint/address from genesis. The number of confirmations not being set within a confirmation request would cause the internal TxNotifier to deadlock when dispatching updates. --- chainntnfs/txnotifier.go | 54 +++++++++++++++++---- chainntnfs/txnotifier_test.go | 91 +++++++++++++++++++++++++++-------- 2 files changed, 116 insertions(+), 29 deletions(-) diff --git a/chainntnfs/txnotifier.go b/chainntnfs/txnotifier.go index 82961ab89..654b757ec 100644 --- a/chainntnfs/txnotifier.go +++ b/chainntnfs/txnotifier.go @@ -46,10 +46,22 @@ var ( // with the TxNotifier but it been shut down. ErrTxNotifierExiting = errors.New("TxNotifier is exiting") - // ErrTxMaxConfs signals that the user requested a number of - // confirmations beyond the reorg safety limit. - ErrTxMaxConfs = fmt.Errorf("too many confirmations requested, max is %d", - MaxNumConfs) + // ErrNoScript is an error returned when a confirmation/spend + // registration is attempted without providing an accompanying output + // script. + ErrNoScript = errors.New("an output script must be provided") + + // ErrNoHeightHint is an error returned when a confirmation/spend + // registration is attempted without providing an accompanying height + // hint. + ErrNoHeightHint = errors.New("a height hint greater than 0 must be " + + "provided") + + // ErrNumConfsOutOfRange is an error returned when a confirmation/spend + // registration is attempted and the number of confirmations provided is + // out of range. + ErrNumConfsOutOfRange = fmt.Errorf("number of confirmations must be "+ + "between %d and %d", 1, MaxNumConfs) ) // rescanState indicates the progression of a registration before the notifier @@ -531,15 +543,27 @@ func NewTxNotifier(startHeight uint32, reorgSafetyLimit uint32, func (n *TxNotifier) newConfNtfn(txid *chainhash.Hash, pkScript []byte, numConfs, heightHint uint32) (*ConfNtfn, error) { - confRequest, err := NewConfRequest(txid, pkScript) - if err != nil { - return nil, err + // An accompanying output script must always be provided. + if len(pkScript) == 0 { + return nil, ErrNoScript } // Enforce that we will not dispatch confirmations beyond the reorg // safety limit. - if numConfs > n.reorgSafetyLimit { - return nil, ErrTxMaxConfs + if numConfs == 0 || numConfs > n.reorgSafetyLimit { + return nil, ErrNumConfsOutOfRange + } + + // A height hint must be provided to prevent scanning from the genesis + // block. + if heightHint == 0 { + return nil, ErrNoHeightHint + } + + // Ensure the output script is of a supported type. + confRequest, err := NewConfRequest(txid, pkScript) + if err != nil { + return nil, err } confID := atomic.AddUint64(&n.confClientCounter, 1) @@ -910,6 +934,18 @@ func (n *TxNotifier) dispatchConfDetails( func (n *TxNotifier) newSpendNtfn(outpoint *wire.OutPoint, pkScript []byte, heightHint uint32) (*SpendNtfn, error) { + // An accompanying output script must always be provided. + if len(pkScript) == 0 { + return nil, ErrNoScript + } + + // A height hint must be provided to prevent scanning from the genesis + // block. + if heightHint == 0 { + return nil, ErrNoHeightHint + } + + // Ensure the output script is of a supported type. spendRequest, err := NewSpendRequest(outpoint, pkScript) if err != nil { return nil, err diff --git a/chainntnfs/txnotifier_test.go b/chainntnfs/txnotifier_test.go index f79fd59d9..79dc735a3 100644 --- a/chainntnfs/txnotifier_test.go +++ b/chainntnfs/txnotifier_test.go @@ -124,30 +124,81 @@ func newMockHintCache() *mockHintCache { } } -// TestTxNotifierMaxConfs ensures that we are not able to register for more -// confirmations on a transaction than the maximum supported. -func TestTxNotifierMaxConfs(t *testing.T) { +// TestTxNotifierRegistrationValidation ensures that we are not able to register +// requests with invalid parameters. +func TestTxNotifierRegistrationValidation(t *testing.T) { t.Parallel() - hintCache := newMockHintCache() - n := chainntnfs.NewTxNotifier( - 10, chainntnfs.ReorgSafetyLimit, hintCache, hintCache, - ) - - // Registering one confirmation above the maximum should fail with - // ErrTxMaxConfs. - _, err := n.RegisterConf( - &chainntnfs.ZeroHash, testRawScript, chainntnfs.MaxNumConfs+1, 1, - ) - if err != chainntnfs.ErrTxMaxConfs { - t.Fatalf("expected chainntnfs.ErrTxMaxConfs, got %v", err) + testCases := []struct { + name string + pkScript []byte + numConfs uint32 + heightHint uint32 + checkSpend bool + err error + }{ + { + name: "empty output script", + pkScript: nil, + numConfs: 1, + heightHint: 1, + checkSpend: true, + err: chainntnfs.ErrNoScript, + }, + { + name: "zero num confs", + pkScript: testRawScript, + numConfs: 0, + heightHint: 1, + err: chainntnfs.ErrNumConfsOutOfRange, + }, + { + name: "exceed max num confs", + pkScript: testRawScript, + numConfs: chainntnfs.MaxNumConfs + 1, + heightHint: 1, + err: chainntnfs.ErrNumConfsOutOfRange, + }, + { + name: "empty height hint", + pkScript: testRawScript, + numConfs: 1, + heightHint: 0, + checkSpend: true, + err: chainntnfs.ErrNoHeightHint, + }, } - _, err = n.RegisterConf( - &chainntnfs.ZeroHash, testRawScript, chainntnfs.MaxNumConfs, 1, - ) - if err != nil { - t.Fatalf("unable to register conf ntfn: %v", err) + for _, testCase := range testCases { + testCase := testCase + t.Run(testCase.name, func(t *testing.T) { + hintCache := newMockHintCache() + n := chainntnfs.NewTxNotifier( + 10, chainntnfs.ReorgSafetyLimit, hintCache, hintCache, + ) + + _, err := n.RegisterConf( + &chainntnfs.ZeroHash, testCase.pkScript, + testCase.numConfs, testCase.heightHint, + ) + if err != testCase.err { + t.Fatalf("conf registration expected error "+ + "\"%v\", got \"%v\"", testCase.err, err) + } + + if !testCase.checkSpend { + return + } + + _, err = n.RegisterSpend( + &chainntnfs.ZeroOutPoint, testCase.pkScript, + testCase.heightHint, + ) + if err != testCase.err { + t.Fatalf("spend registration expected error "+ + "\"%v\", got \"%v\"", testCase.err, err) + } + }) } } From cb1d6683d9ad415410053d3feef648bdfde606c7 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Fri, 16 Aug 2019 19:58:26 -0700 Subject: [PATCH 4/9] chainntnfs: acquire TxNotifier lock after closing quit chan This prevents a deadlock while tearing down the TxNotifier if it's currently blocked delivering a notification. By closing the quit chan first, we ensure blocked sends/reads can exit and allow the TxNotifier to proceed tearing down. --- chainntnfs/txnotifier.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/chainntnfs/txnotifier.go b/chainntnfs/txnotifier.go index 654b757ec..ad81aba04 100644 --- a/chainntnfs/txnotifier.go +++ b/chainntnfs/txnotifier.go @@ -1912,11 +1912,11 @@ func (n *TxNotifier) dispatchSpendReorg(ntfn *SpendNtfn) error { // closes the event channels of all registered notifications that have not been // dispatched yet. func (n *TxNotifier) TearDown() { + close(n.quit) + n.Lock() defer n.Unlock() - close(n.quit) - for _, confSet := range n.confNotifications { for _, ntfn := range confSet.ntfns { close(ntfn.Event.Confirmed) From a9efb617671cfda15b6b4e58194abfefb3dfde75 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Mon, 19 Aug 2019 14:23:43 -0700 Subject: [PATCH 5/9] lnwallet: remove irrelevant fields in Utxo struct These fields are only relevant for spent transaction outputs. --- lnwallet/interface.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/lnwallet/interface.go b/lnwallet/interface.go index f6badd075..e09fc2715 100644 --- a/lnwallet/interface.go +++ b/lnwallet/interface.go @@ -59,8 +59,6 @@ type Utxo struct { Value btcutil.Amount Confirmations int64 PkScript []byte - RedeemScript []byte - WitnessScript []byte wire.OutPoint } From c7bdfe149a603f4d89978b81514eb5c20cc537c0 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Mon, 19 Aug 2019 14:25:50 -0700 Subject: [PATCH 6/9] lnwallet/btcwallet: remove internal utxoCache The cache wasn't really serving a purpose as FetchInputInfo isn't known to be a hot path. Also, with a planned addition of returning the confirmation status of an output within FetchInputInfo in a later commit, caching won't be useful as we'll have to go to disk anyway to determine the confirmation status. --- lnwallet/btcwallet/btcwallet.go | 6 ------ lnwallet/btcwallet/signer.go | 15 +-------------- 2 files changed, 1 insertion(+), 20 deletions(-) diff --git a/lnwallet/btcwallet/btcwallet.go b/lnwallet/btcwallet/btcwallet.go index 420760e86..9e2cab9e6 100644 --- a/lnwallet/btcwallet/btcwallet.go +++ b/lnwallet/btcwallet/btcwallet.go @@ -59,11 +59,6 @@ type BtcWallet struct { netParams *chaincfg.Params chainKeyScope waddrmgr.KeyScope - - // utxoCache is a cache used to speed up repeated calls to - // FetchInputInfo. - utxoCache map[wire.OutPoint]*wire.TxOut - cacheMtx sync.RWMutex } // A compile time check to ensure that BtcWallet implements the @@ -130,7 +125,6 @@ func New(cfg Config) (*BtcWallet, error) { chain: cfg.ChainSource, netParams: cfg.NetParams, chainKeyScope: chainKeyScope, - utxoCache: make(map[wire.OutPoint]*wire.TxOut), }, nil } diff --git a/lnwallet/btcwallet/signer.go b/lnwallet/btcwallet/signer.go index ddc45f3bf..c32c64e3f 100644 --- a/lnwallet/btcwallet/signer.go +++ b/lnwallet/btcwallet/signer.go @@ -26,17 +26,7 @@ func (b *BtcWallet) FetchInputInfo(prevOut *wire.OutPoint) (*wire.TxOut, error) err error output *wire.TxOut ) - - // First check to see if the output is already within the utxo cache. - // If so we can return directly saving a disk access. - b.cacheMtx.RLock() - if output, ok := b.utxoCache[*prevOut]; ok { - b.cacheMtx.RUnlock() - return output, nil - } - b.cacheMtx.RUnlock() - - // Otherwise, we manually look up the output within the tx store. + // We manually look up the output within the tx store. txid := &prevOut.Hash txDetail, err := base.UnstableAPI(b.wallet).TxDetails(txid) if err != nil { @@ -54,9 +44,6 @@ func (b *BtcWallet) FetchInputInfo(prevOut *wire.OutPoint) (*wire.TxOut, error) return nil, err } - b.cacheMtx.Lock() - b.utxoCache[*prevOut] = output - b.cacheMtx.Unlock() return output, nil } From 8b398bdbd17d56c134fa712fd8cc78b75e5be1a0 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Mon, 19 Aug 2019 14:25:51 -0700 Subject: [PATCH 7/9] sweep: remove unnecessary use of FetchInputInfo within CraftSweepAllTx We already have all of the information required for the outputs from the ListUnspent method. --- sweep/walletsweep.go | 28 ++++++++-------------------- sweep/walletsweep_test.go | 28 +++++----------------------- 2 files changed, 13 insertions(+), 43 deletions(-) diff --git a/sweep/walletsweep.go b/sweep/walletsweep.go index 19ffd986c..e21457788 100644 --- a/sweep/walletsweep.go +++ b/sweep/walletsweep.go @@ -101,11 +101,6 @@ type UtxoSource interface { // ListUnspentWitness returns all UTXOs from the source that have // between minConfs and maxConfs number of confirmations. ListUnspentWitness(minConfs, maxConfs int32) ([]*lnwallet.Utxo, error) - - // FetchInputInfo returns the matching output for an outpoint. If the - // outpoint doesn't belong to this UTXO source, then an error should be - // returned. - FetchInputInfo(*wire.OutPoint) (*wire.TxOut, error) } // CoinSelectionLocker is an interface that allows the caller to perform an @@ -217,41 +212,34 @@ func CraftSweepAllTx(feeRate lnwallet.SatPerKWeight, blockHeight uint32, // sweeper to generate and sign a transaction for us. var inputsToSweep []input.Input for _, output := range allOutputs { - // We'll consult the utxoSource for information concerning this - // outpoint, we'll need to properly populate a signDescriptor - // for this output. - outputInfo, err := utxoSource.FetchInputInfo(&output.OutPoint) - if err != nil { - unlockOutputs() - - return nil, err - } - // As we'll be signing for outputs under control of the wallet, // we only need to populate the output value and output script. // The rest of the items will be populated internally within // the sweeper via the witness generation function. signDesc := &input.SignDescriptor{ - Output: outputInfo, + Output: &wire.TxOut{ + PkScript: output.PkScript, + Value: int64(output.Value), + }, HashType: txscript.SigHashAll, } - pkScript := outputInfo.PkScript + pkScript := output.PkScript // Based on the output type, we'll map it to the proper witness // type so we can generate the set of input scripts needed to // sweep the output. var witnessType input.WitnessType - switch { + switch output.AddressType { // If this is a p2wkh output, then we'll assume it's a witness // key hash witness type. - case txscript.IsPayToWitnessPubKeyHash(pkScript): + case lnwallet.WitnessPubKey: witnessType = input.WitnessKeyHash // If this is a p2sh output, then as since it's under control // of the wallet, we'll assume it's a nested p2sh output. - case txscript.IsPayToScriptHash(pkScript): + case lnwallet.NestedWitnessPubKey: witnessType = input.NestedWitnessKeyHash // All other output types we count as unknown and will fail to diff --git a/sweep/walletsweep_test.go b/sweep/walletsweep_test.go index f8f692dbe..0b96e00b9 100644 --- a/sweep/walletsweep_test.go +++ b/sweep/walletsweep_test.go @@ -108,25 +108,13 @@ func TestDetermineFeePerKw(t *testing.T) { } type mockUtxoSource struct { - outpoints map[wire.OutPoint]*wire.TxOut - outputs []*lnwallet.Utxo } func newMockUtxoSource(utxos []*lnwallet.Utxo) *mockUtxoSource { - m := &mockUtxoSource{ - outputs: utxos, - outpoints: make(map[wire.OutPoint]*wire.TxOut), + return &mockUtxoSource{ + outputs: utxos, } - - for _, utxo := range utxos { - m.outpoints[utxo.OutPoint] = &wire.TxOut{ - Value: int64(utxo.Value), - PkScript: utxo.PkScript, - } - } - - return m } func (m *mockUtxoSource) ListUnspentWitness(minConfs int32, @@ -135,15 +123,6 @@ func (m *mockUtxoSource) ListUnspentWitness(minConfs int32, return m.outputs, nil } -func (m *mockUtxoSource) FetchInputInfo(op *wire.OutPoint) (*wire.TxOut, error) { - txOut, ok := m.outpoints[*op] - if !ok { - return nil, fmt.Errorf("no output found") - } - - return txOut, nil -} - type mockCoinSelectionLocker struct { fail bool } @@ -202,6 +181,7 @@ var deliveryAddr = func() btcutil.Address { var testUtxos = []*lnwallet.Utxo{ { // A p2wkh output. + AddressType: lnwallet.WitnessPubKey, PkScript: []byte{ 0x0, 0x14, 0x64, 0x3d, 0x8b, 0x15, 0x69, 0x4a, 0x54, 0x7d, 0x57, 0x33, 0x6e, 0x51, 0xdf, 0xfd, 0x38, 0xe3, @@ -215,6 +195,7 @@ var testUtxos = []*lnwallet.Utxo{ { // A np2wkh output. + AddressType: lnwallet.NestedWitnessPubKey, PkScript: []byte{ 0xa9, 0x14, 0x97, 0x17, 0xf7, 0xd1, 0x5f, 0x6f, 0x8b, 0x7, 0xe3, 0x58, 0x43, 0x19, 0xb9, 0x7e, 0xa9, 0x20, @@ -228,6 +209,7 @@ var testUtxos = []*lnwallet.Utxo{ // A p2wsh output. { + AddressType: lnwallet.UnknownAddressType, PkScript: []byte{ 0x0, 0x20, 0x70, 0x1a, 0x8d, 0x40, 0x1c, 0x84, 0xfb, 0x13, 0xe6, 0xba, 0xf1, 0x69, 0xd5, 0x96, 0x84, 0xe2, 0x7a, 0xbd, From 9e8a1707cca78dd8a35d55a923574ccb2126e007 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Mon, 19 Aug 2019 14:26:36 -0700 Subject: [PATCH 8/9] lnwallet: modify FetchInputInfo to return additional information for utxos --- lnrpc/walletrpc/walletkit_server.go | 13 +++++---- lnwallet/btcwallet/signer.go | 41 +++++++++++++++++++++++------ lnwallet/interface.go | 2 +- lnwallet/wallet.go | 5 +++- mock.go | 13 +++++---- 5 files changed, 54 insertions(+), 20 deletions(-) diff --git a/lnrpc/walletrpc/walletkit_server.go b/lnrpc/walletrpc/walletkit_server.go index 4b3a26c60..374983f42 100644 --- a/lnrpc/walletrpc/walletkit_server.go +++ b/lnrpc/walletrpc/walletkit_server.go @@ -496,23 +496,26 @@ func (w *WalletKit) BumpFee(ctx context.Context, // // We'll gather all of the information required by the UtxoSweeper in // order to sweep the output. - txOut, err := w.cfg.Wallet.FetchInputInfo(op) + utxo, err := w.cfg.Wallet.FetchInputInfo(op) if err != nil { return nil, err } var witnessType input.WitnessType - switch { - case txscript.IsPayToWitnessPubKeyHash(txOut.PkScript): + switch utxo.AddressType { + case lnwallet.WitnessPubKey: witnessType = input.WitnessKeyHash - case txscript.IsPayToScriptHash(txOut.PkScript): + case lnwallet.NestedWitnessPubKey: witnessType = input.NestedWitnessKeyHash default: return nil, fmt.Errorf("unknown input witness %v", op) } signDesc := &input.SignDescriptor{ - Output: txOut, + Output: &wire.TxOut{ + PkScript: utxo.PkScript, + Value: int64(utxo.Value), + }, HashType: txscript.SigHashAll, } diff --git a/lnwallet/btcwallet/signer.go b/lnwallet/btcwallet/signer.go index c32c64e3f..35e158c02 100644 --- a/lnwallet/btcwallet/signer.go +++ b/lnwallet/btcwallet/signer.go @@ -1,6 +1,8 @@ package btcwallet import ( + "fmt" + "github.com/btcsuite/btcd/btcec" "github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/txscript" @@ -21,11 +23,7 @@ import ( // of ErrNotMine should be returned instead. // // This is a part of the WalletController interface. -func (b *BtcWallet) FetchInputInfo(prevOut *wire.OutPoint) (*wire.TxOut, error) { - var ( - err error - output *wire.TxOut - ) +func (b *BtcWallet) FetchInputInfo(prevOut *wire.OutPoint) (*lnwallet.Utxo, error) { // We manually look up the output within the tx store. txid := &prevOut.Hash txDetail, err := base.UnstableAPI(b.wallet).TxDetails(txid) @@ -39,13 +37,40 @@ func (b *BtcWallet) FetchInputInfo(prevOut *wire.OutPoint) (*wire.TxOut, error) // we actually have control of this output. We do this because the check // above only guarantees that the transaction is somehow relevant to us, // like in the event of us being the sender of the transaction. - output = txDetail.TxRecord.MsgTx.TxOut[prevOut.Index] - if _, err := b.fetchOutputAddr(output.PkScript); err != nil { + pkScript := txDetail.TxRecord.MsgTx.TxOut[prevOut.Index].PkScript + if _, err := b.fetchOutputAddr(pkScript); err != nil { return nil, err } + // Then, we'll populate all of the information required by the struct. + addressType := lnwallet.UnknownAddressType + switch { + case txscript.IsPayToWitnessPubKeyHash(pkScript): + addressType = lnwallet.WitnessPubKey + case txscript.IsPayToScriptHash(pkScript): + addressType = lnwallet.NestedWitnessPubKey + } - return output, nil + // Determine the number of confirmations the output currently has. + _, currentHeight, err := b.GetBestBlock() + if err != nil { + return nil, fmt.Errorf("unable to retrieve current height: %v", + err) + } + confs := int64(0) + if txDetail.Block.Height != -1 { + confs = int64(currentHeight - txDetail.Block.Height) + } + + return &lnwallet.Utxo{ + AddressType: addressType, + Value: btcutil.Amount( + txDetail.TxRecord.MsgTx.TxOut[prevOut.Index].Value, + ), + PkScript: pkScript, + Confirmations: confs, + OutPoint: *prevOut, + }, nil } // fetchOutputAddr attempts to fetch the managed address corresponding to the diff --git a/lnwallet/interface.go b/lnwallet/interface.go index e09fc2715..bf6c255da 100644 --- a/lnwallet/interface.go +++ b/lnwallet/interface.go @@ -135,7 +135,7 @@ type WalletController interface { // passed outpoint. If the base wallet determines this output is under // its control, then the original txout should be returned. Otherwise, // a non-nil error value of ErrNotMine should be returned instead. - FetchInputInfo(prevOut *wire.OutPoint) (*wire.TxOut, error) + FetchInputInfo(prevOut *wire.OutPoint) (*Utxo, error) // ConfirmedBalance returns the sum of all the wallet's unspent outputs // that have at least confs confirmations. If confs is set to zero, diff --git a/lnwallet/wallet.go b/lnwallet/wallet.go index 3a2a01606..13c738516 100644 --- a/lnwallet/wallet.go +++ b/lnwallet/wallet.go @@ -774,7 +774,10 @@ func (l *LightningWallet) handleContributionMsg(req *addContributionMsg) { return } - signDesc.Output = info + signDesc.Output = &wire.TxOut{ + PkScript: info.PkScript, + Value: int64(info.Value), + } signDesc.InputIndex = i inputScript, err := l.Cfg.Signer.ComputeInputScript( diff --git a/mock.go b/mock.go index cb4892ed7..0b41bf835 100644 --- a/mock.go +++ b/mock.go @@ -242,12 +242,15 @@ func (*mockWalletController) BackEnd() string { // FetchInputInfo will be called to get info about the inputs to the funding // transaction. func (*mockWalletController) FetchInputInfo( - prevOut *wire.OutPoint) (*wire.TxOut, error) { - txOut := &wire.TxOut{ - Value: int64(10 * btcutil.SatoshiPerBitcoin), - PkScript: []byte("dummy"), + prevOut *wire.OutPoint) (*lnwallet.Utxo, error) { + utxo := &lnwallet.Utxo{ + AddressType: lnwallet.WitnessPubKey, + Value: 10 * btcutil.SatoshiPerBitcoin, + PkScript: []byte("dummy"), + Confirmations: 1, + OutPoint: *prevOut, } - return txOut, nil + return utxo, nil } func (*mockWalletController) ConfirmedBalance(confs int32) (btcutil.Amount, error) { return 0, nil From 2cc5891c8c6c367e9e83ffca41a8340234267a31 Mon Sep 17 00:00:00 2001 From: Wilmer Paulino Date: Wed, 21 Aug 2019 17:24:07 -0700 Subject: [PATCH 9/9] lnrpc/walletrpc: use current height as height hint for cpfp In this commit, we address an issue that would cause us to scan from the genesis block for the spend of an output that we wish to use to raise the fee of a transaction through CPFP. This was due to setting a 0 height hint when constructing the input required by the sweeper and was discovered due to the recently added validation checks at the chain notifier level. We'll now use the current height as the height hint instead as the sweeper will end up creating a new transaction that spends the input. --- lnrpc/walletrpc/config_active.go | 4 ++++ lnrpc/walletrpc/driver.go | 8 ++++++++ lnrpc/walletrpc/walletkit_server.go | 17 ++++++++++++++++- subrpcserver_config.go | 3 +++ 4 files changed, 31 insertions(+), 1 deletion(-) diff --git a/lnrpc/walletrpc/config_active.go b/lnrpc/walletrpc/config_active.go index 77dc0e574..9804bc119 100644 --- a/lnrpc/walletrpc/config_active.go +++ b/lnrpc/walletrpc/config_active.go @@ -43,4 +43,8 @@ type Config struct { // Sweeper is the central batching engine of lnd. It is responsible for // sweeping inputs in batches back into the wallet. Sweeper *sweep.UtxoSweeper + + // Chain is an interface that the WalletKit will use to determine state + // about the backing chain of the wallet. + Chain lnwallet.BlockChainIO } diff --git a/lnrpc/walletrpc/driver.go b/lnrpc/walletrpc/driver.go index 4135d3aad..deb02a6e9 100644 --- a/lnrpc/walletrpc/driver.go +++ b/lnrpc/walletrpc/driver.go @@ -51,6 +51,14 @@ func createNewSubServer(configRegistry lnrpc.SubServerConfigDispatcher) (lnrpc.S case config.KeyRing == nil: return nil, nil, fmt.Errorf("KeyRing must be set to create " + "WalletKit RPC server") + + case config.Sweeper == nil: + return nil, nil, fmt.Errorf("Sweeper must be set to create " + + "WalletKit RPC server") + + case config.Chain == nil: + return nil, nil, fmt.Errorf("Chain must be set to create " + + "WalletKit RPC server") } return New(config) diff --git a/lnrpc/walletrpc/walletkit_server.go b/lnrpc/walletrpc/walletkit_server.go index 374983f42..b463774ed 100644 --- a/lnrpc/walletrpc/walletkit_server.go +++ b/lnrpc/walletrpc/walletkit_server.go @@ -4,6 +4,7 @@ package walletrpc import ( "bytes" + "errors" "fmt" "io/ioutil" "os" @@ -501,6 +502,12 @@ func (w *WalletKit) BumpFee(ctx context.Context, return nil, err } + // We're only able to bump the fee of unconfirmed transactions. + if utxo.Confirmations > 0 { + return nil, errors.New("unable to bump fee of a confirmed " + + "transaction") + } + var witnessType input.WitnessType switch utxo.AddressType { case lnwallet.WitnessPubKey: @@ -519,7 +526,15 @@ func (w *WalletKit) BumpFee(ctx context.Context, HashType: txscript.SigHashAll, } - input := input.NewBaseInput(op, witnessType, signDesc, 0) + // We'll use the current height as the height hint since we're dealing + // with an unconfirmed transaction. + _, currentHeight, err := w.cfg.Chain.GetBestBlock() + if err != nil { + return nil, fmt.Errorf("unable to retrieve current height: %v", + err) + } + + input := input.NewBaseInput(op, witnessType, signDesc, uint32(currentHeight)) if _, err = w.cfg.Sweeper.SweepInput(input, feePreference); err != nil { return nil, err } diff --git a/subrpcserver_config.go b/subrpcserver_config.go index 06b9dfc5b..d00e3a43d 100644 --- a/subrpcserver_config.go +++ b/subrpcserver_config.go @@ -152,6 +152,9 @@ func (s *subRPCServerConfigs) PopulateDependencies(cc *chainControl, subCfgValue.FieldByName("Sweeper").Set( reflect.ValueOf(sweeper), ) + subCfgValue.FieldByName("Chain").Set( + reflect.ValueOf(cc.chainIO), + ) case *autopilotrpc.Config: subCfgValue := extractReflectValue(subCfg)