akovalenko/lnd.xprv
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

chainntnfs: rename TxConfNotifier -> TxNotifier

Browse Source
This commit is contained in:
Wilmer Paulino 2018-10-05 02:07:55 -07:00
parent f8789e9db0
commit 82f6fd7a91
No known key found for this signature in database
GPG Key ID: 6DF57B9F9514972F
9 changed files with 231 additions and 233 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
chainntnfs/bitcoindnotify/bitcoind.go
View File

@ -71,7 +71,7 @@ type BitcoindNotifier struct {
spendNotifications map[wire.OutPoint]map[uint64]*spendNotification spendNotifications map[wire.OutPoint]map[uint64]*spendNotification
txConfNotifier *chainntnfs.TxConfNotifier txNotifier *chainntnfs.TxNotifier
blockEpochClients map[uint64]*blockEpochRegistration blockEpochClients map[uint64]*blockEpochRegistration
@ -142,7 +142,7 @@ func (b *BitcoindNotifier) Start() error {
return err return err
} }
b.txConfNotifier = chainntnfs.NewTxConfNotifier( b.txNotifier = chainntnfs.NewTxNotifier(
uint32(currentHeight), reorgSafetyLimit, b.confirmHintCache, uint32(currentHeight), reorgSafetyLimit, b.confirmHintCache,
) )
@ -184,7 +184,7 @@ func (b *BitcoindNotifier) Stop() error {
close(epochClient.epochChan) close(epochClient.epochChan)
} }
b.txConfNotifier.TearDown() b.txNotifier.TearDown()
return nil return nil
} }
@ -278,7 +278,7 @@ out:
// begin safely updating the height hint // begin safely updating the height hint
// cache at tip, since any pending // cache at tip, since any pending
// rescans have now completed. // rescans have now completed.
err = b.txConfNotifier.UpdateConfDetails( err = b.txNotifier.UpdateConfDetails(
*msg.TxID, confDetails, *msg.TxID, confDetails,
) )
if err != nil { if err != nil {
@ -330,7 +330,7 @@ out:
newBestBlock, missedBlocks, err := newBestBlock, missedBlocks, err :=
chainntnfs.HandleMissedBlocks( chainntnfs.HandleMissedBlocks(
b.chainConn, b.chainConn,
b.txConfNotifier, b.txNotifier,
b.bestBlock, item.Height, b.bestBlock, item.Height,
true, true,
) )
@ -369,7 +369,7 @@ out:
} }
newBestBlock, err := chainntnfs.RewindChain( newBestBlock, err := chainntnfs.RewindChain(
b.chainConn, b.txConfNotifier, b.chainConn, b.txNotifier,
b.bestBlock, item.Height-1, b.bestBlock, item.Height-1,
) )
if err != nil { if err != nil {
@ -621,7 +621,7 @@ func (b *BitcoindNotifier) handleBlockConnected(block chainntnfs.BlockEpoch) err
} }
txns := btcutil.NewBlock(rawBlock).Transactions() txns := btcutil.NewBlock(rawBlock).Transactions()
err = b.txConfNotifier.ConnectTip( err = b.txNotifier.ConnectTip(
block.Hash, uint32(block.Height), txns) block.Hash, uint32(block.Height), txns)
if err != nil { if err != nil {
return fmt.Errorf("unable to connect tip: %v", err) return fmt.Errorf("unable to connect tip: %v", err)
@ -959,11 +959,11 @@ func (b *BitcoindNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash,
chainntnfs.Log.Infof("New confirmation subscription: "+ chainntnfs.Log.Infof("New confirmation subscription: "+
"txid=%v, numconfs=%v", txid, numConfs) "txid=%v, numconfs=%v", txid, numConfs)
// Register the conf notification with txconfnotifier. A non-nil value // Register the conf notification with the TxNotifier. A non-nil value
// for `dispatch` will be returned if we are required to perform a // for `dispatch` will be returned if we are required to perform a
// manual scan for the confirmation. Otherwise the notifier will begin // manual scan for the confirmation. Otherwise the notifier will begin
// watching at tip for the transaction to confirm. // watching at tip for the transaction to confirm.
dispatch, err := b.txConfNotifier.Register(ntfn) dispatch, err := b.txNotifier.Register(ntfn)
if err != nil { if err != nil {
return nil, err return nil, err
} }

13
chainntnfs/bitcoindnotify/bitcoind_dev.go
View File

@ -12,12 +12,11 @@ import (
) )
// UnsafeStart starts the notifier with a specified best height and optional // UnsafeStart starts the notifier with a specified best height and optional
// best hash. Its bestBlock and txConfNotifier are initialized with // best hash. Its bestBlock and txNotifier are initialized with bestHeight and
// bestHeight and optionally bestHash. The parameter generateBlocks is // optionally bestHash. The parameter generateBlocks is necessary for the
// necessary for the bitcoind notifier to ensure we drain all notifications up // bitcoind notifier to ensure we drain all notifications up to syncHeight,
// to syncHeight, since if they are generated ahead of UnsafeStart the chainConn // since if they are generated ahead of UnsafeStart the chainConn may start up
// may start up with an outdated best block and miss sending ntfns. Used for // with an outdated best block and miss sending ntfns. Used for testing.
// testing.
func (b *BitcoindNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Hash, func (b *BitcoindNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Hash,
syncHeight int32, generateBlocks func() error) error { syncHeight int32, generateBlocks func() error) error {
@ -30,7 +29,7 @@ func (b *BitcoindNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Has
return err return err
} }
b.txConfNotifier = chainntnfs.NewTxConfNotifier( b.txNotifier = chainntnfs.NewTxNotifier(
uint32(bestHeight), reorgSafetyLimit, b.confirmHintCache, uint32(bestHeight), reorgSafetyLimit, b.confirmHintCache,
) )

18
chainntnfs/btcdnotify/btcd.go
View File

@ -76,7 +76,7 @@ type BtcdNotifier struct {
spendNotifications map[wire.OutPoint]map[uint64]*spendNotification spendNotifications map[wire.OutPoint]map[uint64]*spendNotification
txConfNotifier *chainntnfs.TxConfNotifier txNotifier *chainntnfs.TxNotifier
blockEpochClients map[uint64]*blockEpochRegistration blockEpochClients map[uint64]*blockEpochRegistration
@ -166,7 +166,7 @@ func (b *BtcdNotifier) Start() error {
return err return err
} }
b.txConfNotifier = chainntnfs.NewTxConfNotifier( b.txNotifier = chainntnfs.NewTxNotifier(
uint32(currentHeight), reorgSafetyLimit, b.confirmHintCache, uint32(currentHeight), reorgSafetyLimit, b.confirmHintCache,
) )
@ -214,7 +214,7 @@ func (b *BtcdNotifier) Stop() error {
close(epochClient.epochChan) close(epochClient.epochChan)
} }
b.txConfNotifier.TearDown() b.txNotifier.TearDown()
return nil return nil
} }
@ -348,7 +348,7 @@ out:
// begin safely updating the height hint // begin safely updating the height hint
// cache at tip, since any pending // cache at tip, since any pending
// rescans have now completed. // rescans have now completed.
err = b.txConfNotifier.UpdateConfDetails( err = b.txNotifier.UpdateConfDetails(
*msg.TxID, confDetails, *msg.TxID, confDetails,
) )
if err != nil { if err != nil {
@ -398,7 +398,7 @@ out:
newBestBlock, missedBlocks, err := newBestBlock, missedBlocks, err :=
chainntnfs.HandleMissedBlocks( chainntnfs.HandleMissedBlocks(
b.chainConn, b.chainConn,
b.txConfNotifier, b.txNotifier,
b.bestBlock, b.bestBlock,
update.blockHeight, update.blockHeight,
true, true,
@ -436,7 +436,7 @@ out:
} }
newBestBlock, err := chainntnfs.RewindChain( newBestBlock, err := chainntnfs.RewindChain(
b.chainConn, b.txConfNotifier, b.bestBlock, b.chainConn, b.txNotifier, b.bestBlock,
update.blockHeight-1, update.blockHeight-1,
) )
if err != nil { if err != nil {
@ -703,7 +703,7 @@ func (b *BtcdNotifier) handleBlockConnected(epoch chainntnfs.BlockEpoch) error {
connect: true, connect: true,
} }
err = b.txConfNotifier.ConnectTip( err = b.txNotifier.ConnectTip(
&newBlock.hash, newBlock.height, newBlock.txns, &newBlock.hash, newBlock.height, newBlock.txns,
) )
if err != nil { if err != nil {
@ -1019,11 +1019,11 @@ func (b *BtcdNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash, _ []byte,
chainntnfs.Log.Infof("New confirmation subscription: "+ chainntnfs.Log.Infof("New confirmation subscription: "+
"txid=%v, numconfs=%v", txid, numConfs) "txid=%v, numconfs=%v", txid, numConfs)
// Register the conf notification with txconfnotifier. A non-nil value // Register the conf notification with the TxNotifier. A non-nil value
// for `dispatch` will be returned if we are required to perform a // for `dispatch` will be returned if we are required to perform a
// manual scan for the confirmation. Otherwise the notifier will begin // manual scan for the confirmation. Otherwise the notifier will begin
// watching at tip for the transaction to confirm. // watching at tip for the transaction to confirm.
dispatch, err := b.txConfNotifier.Register(ntfn) dispatch, err := b.txNotifier.Register(ntfn)
if err != nil { if err != nil {
return nil, err return nil, err
} }

13
chainntnfs/btcdnotify/btcd_dev.go
View File

@ -11,12 +11,11 @@ import (
) )
// UnsafeStart starts the notifier with a specified best height and optional // UnsafeStart starts the notifier with a specified best height and optional
// best hash. Its bestBlock and txConfNotifier are initialized with // best hash. Its bestBlock and txNotifier are initialized with bestHeight and
// bestHeight and optionally bestHash. The parameter generateBlocks is // optionally bestHash. The parameter generateBlocks is necessary for the
// necessary for the bitcoind notifier to ensure we drain all notifications up // bitcoind notifier to ensure we drain all notifications up to syncHeight,
// to syncHeight, since if they are generated ahead of UnsafeStart the chainConn // since if they are generated ahead of UnsafeStart the chainConn may start up
// may start up with an outdated best block and miss sending ntfns. Used for // with an outdated best block and miss sending ntfns. Used for testing.
// testing.
func (b *BtcdNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Hash, func (b *BtcdNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Hash,
syncHeight int32, generateBlocks func() error) error { syncHeight int32, generateBlocks func() error) error {
@ -29,7 +28,7 @@ func (b *BtcdNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Hash,
return err return err
} }
b.txConfNotifier = chainntnfs.NewTxConfNotifier( b.txNotifier = chainntnfs.NewTxNotifier(
uint32(bestHeight), reorgSafetyLimit, b.confirmHintCache, uint32(bestHeight), reorgSafetyLimit, b.confirmHintCache,
) )

12
chainntnfs/interface.go
View File

@ -392,10 +392,10 @@ func GetClientMissedBlocks(chainConn ChainConn, clientBestBlock *BlockEpoch,
return missedBlocks, nil return missedBlocks, nil
} }
// RewindChain handles internal state updates for the notifier's TxConfNotifier // RewindChain handles internal state updates for the notifier's TxNotifier It
// It has no effect if given a height greater than or equal to our current best // has no effect if given a height greater than or equal to our current best
// known height. It returns the new best block for the notifier. // known height. It returns the new best block for the notifier.
func RewindChain(chainConn ChainConn, txConfNotifier *TxConfNotifier, func RewindChain(chainConn ChainConn, txNotifier *TxNotifier,
currBestBlock BlockEpoch, targetHeight int32) (BlockEpoch, error) { currBestBlock BlockEpoch, targetHeight int32) (BlockEpoch, error) {
newBestBlock := BlockEpoch{ newBestBlock := BlockEpoch{
@ -414,7 +414,7 @@ func RewindChain(chainConn ChainConn, txConfNotifier *TxConfNotifier,
Log.Infof("Block disconnected from main chain: "+ Log.Infof("Block disconnected from main chain: "+
"height=%v, sha=%v", height, newBestBlock.Hash) "height=%v, sha=%v", height, newBestBlock.Hash)
err = txConfNotifier.DisconnectTip(uint32(height)) err = txNotifier.DisconnectTip(uint32(height))
if err != nil { if err != nil {
return newBestBlock, fmt.Errorf("unable to "+ return newBestBlock, fmt.Errorf("unable to "+
" disconnect tip for height=%d: %v", " disconnect tip for height=%d: %v",
@ -436,7 +436,7 @@ func RewindChain(chainConn ChainConn, txConfNotifier *TxConfNotifier,
// returned in case a chain rewind occurs and partially completes before // returned in case a chain rewind occurs and partially completes before
// erroring. In the case where there is no rewind, the notifier's // erroring. In the case where there is no rewind, the notifier's
// current best block is returned. // current best block is returned.
func HandleMissedBlocks(chainConn ChainConn, txConfNotifier *TxConfNotifier, func HandleMissedBlocks(chainConn ChainConn, txNotifier *TxNotifier,
currBestBlock BlockEpoch, newHeight int32, currBestBlock BlockEpoch, newHeight int32,
backendStoresReorgs bool) (BlockEpoch, []BlockEpoch, error) { backendStoresReorgs bool) (BlockEpoch, []BlockEpoch, error) {
@ -462,7 +462,7 @@ func HandleMissedBlocks(chainConn ChainConn, txConfNotifier *TxConfNotifier,
"common ancestor: %v", err) "common ancestor: %v", err)
} }
currBestBlock, err = RewindChain(chainConn, txConfNotifier, currBestBlock, err = RewindChain(chainConn, txNotifier,
currBestBlock, startingHeight) currBestBlock, startingHeight)
if err != nil { if err != nil {
return currBestBlock, nil, fmt.Errorf("unable to "+ return currBestBlock, nil, fmt.Errorf("unable to "+

18
chainntnfs/neutrinonotify/neutrino.go
View File

@ -70,7 +70,7 @@ type NeutrinoNotifier struct {
spendNotifications map[wire.OutPoint]map[uint64]*spendNotification spendNotifications map[wire.OutPoint]map[uint64]*spendNotification
txConfNotifier *chainntnfs.TxConfNotifier txNotifier *chainntnfs.TxNotifier
blockEpochClients map[uint64]*blockEpochRegistration blockEpochClients map[uint64]*blockEpochRegistration
@ -162,7 +162,7 @@ func (n *NeutrinoNotifier) Start() error {
neutrino.WatchInputs(zeroInput), neutrino.WatchInputs(zeroInput),
} }
n.txConfNotifier = chainntnfs.NewTxConfNotifier( n.txNotifier = chainntnfs.NewTxNotifier(
n.bestHeight, reorgSafetyLimit, n.confirmHintCache, n.bestHeight, reorgSafetyLimit, n.confirmHintCache,
) )
@ -206,7 +206,7 @@ func (n *NeutrinoNotifier) Stop() error {
close(epochClient.epochChan) close(epochClient.epochChan)
} }
n.txConfNotifier.TearDown() n.txNotifier.TearDown()
return nil return nil
} }
@ -350,7 +350,7 @@ out:
// begin safely updating the height hint // begin safely updating the height hint
// cache at tip, since any pending // cache at tip, since any pending
// rescans have now completed. // rescans have now completed.
err = n.txConfNotifier.UpdateConfDetails( err = n.txNotifier.UpdateConfDetails(
*msg.TxID, confDetails, *msg.TxID, confDetails,
) )
if err != nil { if err != nil {
@ -426,7 +426,7 @@ out:
_, missedBlocks, err := _, missedBlocks, err :=
chainntnfs.HandleMissedBlocks( chainntnfs.HandleMissedBlocks(
n.chainConn, n.chainConn,
n.txConfNotifier, n.txNotifier,
bestBlock, bestBlock,
int32(update.height), int32(update.height),
false, false,
@ -482,7 +482,7 @@ out:
Hash: hash, Hash: hash,
} }
newBestBlock, err := chainntnfs.RewindChain( newBestBlock, err := chainntnfs.RewindChain(
n.chainConn, n.txConfNotifier, notifierBestBlock, n.chainConn, n.txNotifier, notifierBestBlock,
int32(update.height-1), int32(update.height-1),
) )
if err != nil { if err != nil {
@ -593,7 +593,7 @@ func (n *NeutrinoNotifier) handleBlockConnected(newBlock *filteredBlock) error {
// First process the block for our internal state. A new block has // First process the block for our internal state. A new block has
// been connected to the main chain. Send out any N confirmation // been connected to the main chain. Send out any N confirmation
// notifications which may have been triggered by this new block. // notifications which may have been triggered by this new block.
err := n.txConfNotifier.ConnectTip( err := n.txNotifier.ConnectTip(
&newBlock.hash, newBlock.height, newBlock.txns, &newBlock.hash, newBlock.height, newBlock.txns,
) )
if err != nil { if err != nil {
@ -928,11 +928,11 @@ func (n *NeutrinoNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash,
chainntnfs.Log.Infof("New confirmation subscription: "+ chainntnfs.Log.Infof("New confirmation subscription: "+
"txid=%v, numconfs=%v", txid, numConfs) "txid=%v, numconfs=%v", txid, numConfs)
// Register the conf notification with txconfnotifier. A non-nil value // Register the conf notification with the TxNotifier. A non-nil value
// for `dispatch` will be returned if we are required to perform a // for `dispatch` will be returned if we are required to perform a
// manual scan for the confirmation. Otherwise the notifier will begin // manual scan for the confirmation. Otherwise the notifier will begin
// watching at tip for the transaction to confirm. // watching at tip for the transaction to confirm.
dispatch, err := n.txConfNotifier.Register(ntfn) dispatch, err := n.txNotifier.Register(ntfn)
if err != nil { if err != nil {
return nil, err return nil, err
} }

17
chainntnfs/neutrinonotify/neutrino_dev.go
View File

@ -13,13 +13,14 @@ import (
) )
// UnsafeStart starts the notifier with a specified best height and optional // UnsafeStart starts the notifier with a specified best height and optional
// best hash. Its bestHeight, txConfNotifier and neutrino node are initialized // best hash. Its bestHeight, txNotifier and neutrino node are initialized with
// with bestHeight. The parameter generateBlocks is necessary for the // bestHeight. The parameter generateBlocks is necessary for the bitcoind
// bitcoind notifier to ensure we drain all notifications up to syncHeight, // notifier to ensure we drain all notifications up to syncHeight, since if they
// since if they are generated ahead of UnsafeStart the chainConn may start // are generated ahead of UnsafeStart the chainConn may start up with an
// up with an outdated best block and miss sending ntfns. Used for testing. // outdated best block and miss sending ntfns. Used for testing.
func (n *NeutrinoNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Hash, func (n *NeutrinoNotifier) UnsafeStart(bestHeight int32,
syncHeight int32, generateBlocks func() error) error { bestHash *chainhash.Hash, syncHeight int32,
generateBlocks func() error) error {
// We'll obtain the latest block height of the p2p node. We'll // We'll obtain the latest block height of the p2p node. We'll
// start the auto-rescan from this point. Once a caller actually wishes // start the auto-rescan from this point. Once a caller actually wishes
@ -47,7 +48,7 @@ func (n *NeutrinoNotifier) UnsafeStart(bestHeight int32, bestHash *chainhash.Has
neutrino.WatchInputs(zeroInput), neutrino.WatchInputs(zeroInput),
} }
n.txConfNotifier = chainntnfs.NewTxConfNotifier( n.txNotifier = chainntnfs.NewTxNotifier(
uint32(bestHeight), reorgSafetyLimit, n.confirmHintCache, uint32(bestHeight), reorgSafetyLimit, n.confirmHintCache,
) )

232
chainntnfs/txnotifier.go
View File

@ -10,9 +10,9 @@ import (
) )
var ( var (
// ErrTxConfNotifierExiting is an error returned when attempting to // ErrTxNotifierExiting is an error returned when attempting to interact
// interact with the TxConfNotifier but it been shut down. // with the TxNotifier but it been shut down.
ErrTxConfNotifierExiting = errors.New("TxConfNotifier is exiting") ErrTxNotifierExiting = errors.New("TxNotifier is exiting")
// ErrTxMaxConfs signals that the user requested a number of // ErrTxMaxConfs signals that the user requested a number of
// confirmations beyond the reorg safety limit. // confirmations beyond the reorg safety limit.
@ -87,12 +87,12 @@ func NewConfirmationEvent(numConfs uint32) *ConfirmationEvent {
} }
} }
// TxConfNotifier is used to register transaction confirmation notifications and // TxNotifier is used to register transaction confirmation notifications and
// dispatch them as the transactions confirm. A client can request to be // dispatch them as the transactions confirm. A client can request to be
// notified when a particular transaction has sufficient on-chain confirmations // notified when a particular transaction has sufficient on-chain confirmations
// (or be notified immediately if the tx already does), and the TxConfNotifier // (or be notified immediately if the tx already does), and the TxNotifier
// will watch changes to the blockchain in order to satisfy these requests. // will watch changes to the blockchain in order to satisfy these requests.
type TxConfNotifier struct { type TxNotifier struct {
// currentHeight is the height of the tracked blockchain. It is used to // currentHeight is the height of the tracked blockchain. It is used to
// determine the number of confirmations a tx has and ensure blocks are // determine the number of confirmations a tx has and ensure blocks are
// connected and disconnected in order. // connected and disconnected in order.
@ -176,12 +176,12 @@ func newConfNtfnSet() *confNtfnSet {
} }
} }
// NewTxConfNotifier creates a TxConfNotifier. The current height of the // NewTxNotifier creates a TxNotifier. The current height of the blockchain is
// blockchain is accepted as a parameter. // accepted as a parameter.
func NewTxConfNotifier(startHeight uint32, reorgSafetyLimit uint32, func NewTxNotifier(startHeight uint32, reorgSafetyLimit uint32,
hintCache ConfirmHintCache) *TxConfNotifier { hintCache ConfirmHintCache) *TxNotifier {
return &TxConfNotifier{ return &TxNotifier{
currentHeight: startHeight, currentHeight: startHeight,
reorgSafetyLimit: reorgSafetyLimit, reorgSafetyLimit: reorgSafetyLimit,
confNotifications: make(map[chainhash.Hash]*confNtfnSet), confNotifications: make(map[chainhash.Hash]*confNtfnSet),
@ -202,18 +202,18 @@ func NewTxConfNotifier(startHeight uint32, reorgSafetyLimit uint32,
// the confirmation details must be provided with the UpdateConfDetails method, // the confirmation details must be provided with the UpdateConfDetails method,
// otherwise we will wait for the transaction to confirm even though it already // otherwise we will wait for the transaction to confirm even though it already
// has. // has.
func (tcn *TxConfNotifier) Register( func (n *TxNotifier) Register(
ntfn *ConfNtfn) (*HistoricalConfDispatch, error) { ntfn *ConfNtfn) (*HistoricalConfDispatch, error) {
select { select {
case <-tcn.quit: case <-n.quit:
return nil, ErrTxConfNotifierExiting return nil, ErrTxNotifierExiting
default: default:
} }
// Enforce that we will not dispatch confirmations beyond the reorg // Enforce that we will not dispatch confirmations beyond the reorg
// safety limit. // safety limit.
if ntfn.NumConfirmations > tcn.reorgSafetyLimit { if ntfn.NumConfirmations > n.reorgSafetyLimit {
return nil, ErrTxMaxConfs return nil, ErrTxMaxConfs
} }
@ -222,7 +222,7 @@ func (tcn *TxConfNotifier) Register(
// //
// TODO(conner): verify that all submitted height hints are identical. // TODO(conner): verify that all submitted height hints are identical.
startHeight := ntfn.HeightHint startHeight := ntfn.HeightHint
hint, err := tcn.hintCache.QueryConfirmHint(*ntfn.TxID) hint, err := n.hintCache.QueryConfirmHint(*ntfn.TxID)
if err == nil { if err == nil {
if hint > startHeight { if hint > startHeight {
Log.Debugf("Using height hint %d retrieved "+ Log.Debugf("Using height hint %d retrieved "+
@ -234,15 +234,15 @@ func (tcn *TxConfNotifier) Register(
*ntfn.TxID, err) *ntfn.TxID, err)
} }
tcn.Lock() n.Lock()
defer tcn.Unlock() defer n.Unlock()
confSet, ok := tcn.confNotifications[*ntfn.TxID] confSet, ok := n.confNotifications[*ntfn.TxID]
if !ok { if !ok {
// If this is the first registration for this txid, construct a // If this is the first registration for this txid, construct a
// confSet to coalesce all notifications for the same txid. // confSet to coalesce all notifications for the same txid.
confSet = newConfNtfnSet() confSet = newConfNtfnSet()
tcn.confNotifications[*ntfn.TxID] = confSet n.confNotifications[*ntfn.TxID] = confSet
} }
confSet.ntfns[ntfn.ConfID] = ntfn confSet.ntfns[ntfn.ConfID] = ntfn
@ -257,7 +257,7 @@ func (tcn *TxConfNotifier) Register(
// client. // client.
Log.Debugf("Attempting to dispatch conf for txid=%v "+ Log.Debugf("Attempting to dispatch conf for txid=%v "+
"on registration since rescan has finished", ntfn.TxID) "on registration since rescan has finished", ntfn.TxID)
return nil, tcn.dispatchConfDetails(ntfn, confSet.details) return nil, n.dispatchConfDetails(ntfn, confSet.details)
// A rescan is already in progress, return here to prevent dispatching // A rescan is already in progress, return here to prevent dispatching
// another. When the scan returns, this notifications details will be // another. When the scan returns, this notifications details will be
@ -274,7 +274,7 @@ func (tcn *TxConfNotifier) Register(
// If the provided or cached height hint indicates that the transaction // If the provided or cached height hint indicates that the transaction
// is to be confirmed at a height greater than the conf notifier's // is to be confirmed at a height greater than the conf notifier's
// current height, we'll refrain from spawning a historical dispatch. // current height, we'll refrain from spawning a historical dispatch.
if startHeight > tcn.currentHeight { if startHeight > n.currentHeight {
Log.Debugf("Height hint is above current height, not dispatching "+ Log.Debugf("Height hint is above current height, not dispatching "+
"historical rescan for txid=%v ", ntfn.TxID) "historical rescan for txid=%v ", ntfn.TxID)
// Set the rescan status to complete, which will allow the conf // Set the rescan status to complete, which will allow the conf
@ -294,7 +294,7 @@ func (tcn *TxConfNotifier) Register(
TxID: ntfn.TxID, TxID: ntfn.TxID,
PkScript: ntfn.PkScript, PkScript: ntfn.PkScript,
StartHeight: startHeight, StartHeight: startHeight,
EndHeight: tcn.currentHeight, EndHeight: n.currentHeight,
} }
// Set this confSet's status to pending, ensuring subsequent // Set this confSet's status to pending, ensuring subsequent
@ -310,23 +310,23 @@ func (tcn *TxConfNotifier) Register(
// //
// NOTE: The notification should be registered first to ensure notifications are // NOTE: The notification should be registered first to ensure notifications are
// dispatched correctly. // dispatched correctly.
func (tcn *TxConfNotifier) UpdateConfDetails(txid chainhash.Hash, func (n *TxNotifier) UpdateConfDetails(txid chainhash.Hash,
details *TxConfirmation) error { details *TxConfirmation) error {
select { select {
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
default: default:
} }
// Ensure we hold the lock throughout handling the notification to // Ensure we hold the lock throughout handling the notification to
// prevent the notifier from advancing its height underneath us. // prevent the notifier from advancing its height underneath us.
tcn.Lock() n.Lock()
defer tcn.Unlock() defer n.Unlock()
// First, we'll determine whether we have an active notification for // First, we'll determine whether we have an active notification for
// this transaction with the given ID. // this transaction with the given ID.
confSet, ok := tcn.confNotifications[txid] confSet, ok := n.confNotifications[txid]
if !ok { if !ok {
return fmt.Errorf("no notification found with TxID %v", txid) return fmt.Errorf("no notification found with TxID %v", txid)
} }
@ -354,7 +354,7 @@ func (tcn *TxConfNotifier) UpdateConfDetails(txid chainhash.Hash,
return nil return nil
} }
if details.BlockHeight > tcn.currentHeight { if details.BlockHeight > n.currentHeight {
Log.Debugf("Conf details for txid=%v found above current "+ Log.Debugf("Conf details for txid=%v found above current "+
"height, waiting to dispatch at tip", txid) "height, waiting to dispatch at tip", txid)
return nil return nil
@ -362,7 +362,7 @@ func (tcn *TxConfNotifier) UpdateConfDetails(txid chainhash.Hash,
Log.Debugf("Updating conf details for txid=%v details", txid) Log.Debugf("Updating conf details for txid=%v details", txid)
err := tcn.hintCache.CommitConfirmHint(details.BlockHeight, txid) err := n.hintCache.CommitConfirmHint(details.BlockHeight, txid)
if err != nil { if err != nil {
// The error is not fatal, so we should not return an error to // The error is not fatal, so we should not return an error to
// the caller. // the caller.
@ -374,7 +374,7 @@ func (tcn *TxConfNotifier) UpdateConfDetails(txid chainhash.Hash,
// notifications that have not yet been delivered. // notifications that have not yet been delivered.
confSet.details = details confSet.details = details
for _, ntfn := range confSet.ntfns { for _, ntfn := range confSet.ntfns {
err = tcn.dispatchConfDetails(ntfn, details) err = n.dispatchConfDetails(ntfn, details)
if err != nil { if err != nil {
return err return err
} }
@ -386,7 +386,7 @@ func (tcn *TxConfNotifier) UpdateConfDetails(txid chainhash.Hash,
// dispatchConfDetails attempts to cache and dispatch details to a particular // dispatchConfDetails attempts to cache and dispatch details to a particular
// client if the transaction has sufficiently confirmed. If the provided details // client if the transaction has sufficiently confirmed. If the provided details
// are nil, this method will be a no-op. // are nil, this method will be a no-op.
func (tcn *TxConfNotifier) dispatchConfDetails( func (n *TxNotifier) dispatchConfDetails(
ntfn *ConfNtfn, details *TxConfirmation) error { ntfn *ConfNtfn, details *TxConfirmation) error {
// If no details are provided, return early as we can't dispatch. // If no details are provided, return early as we can't dispatch.
@ -401,7 +401,7 @@ func (tcn *TxConfNotifier) dispatchConfDetails(
// confirmations. If it has, we'll dispatch a confirmation // confirmations. If it has, we'll dispatch a confirmation
// notification to the caller. // notification to the caller.
confHeight := details.BlockHeight + ntfn.NumConfirmations - 1 confHeight := details.BlockHeight + ntfn.NumConfirmations - 1
if confHeight <= tcn.currentHeight { if confHeight <= n.currentHeight {
Log.Infof("Dispatching %v conf notification for %v", Log.Infof("Dispatching %v conf notification for %v",
ntfn.NumConfirmations, ntfn.TxID) ntfn.NumConfirmations, ntfn.TxID)
@ -410,15 +410,15 @@ func (tcn *TxConfNotifier) dispatchConfDetails(
// confirmed. // confirmed.
select { select {
case ntfn.Event.Updates <- 0: case ntfn.Event.Updates <- 0:
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
} }
select { select {
case ntfn.Event.Confirmed <- details: case ntfn.Event.Confirmed <- details:
ntfn.dispatched = true ntfn.dispatched = true
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
} }
} else { } else {
Log.Debugf("Queueing %v conf notification for %v at tip ", Log.Debugf("Queueing %v conf notification for %v at tip ",
@ -427,33 +427,33 @@ func (tcn *TxConfNotifier) dispatchConfDetails(
// Otherwise, we'll keep track of the notification // Otherwise, we'll keep track of the notification
// request by the height at which we should dispatch the // request by the height at which we should dispatch the
// confirmation notification. // confirmation notification.
ntfnSet, exists := tcn.ntfnsByConfirmHeight[confHeight] ntfnSet, exists := n.ntfnsByConfirmHeight[confHeight]
if !exists { if !exists {
ntfnSet = make(map[*ConfNtfn]struct{}) ntfnSet = make(map[*ConfNtfn]struct{})
tcn.ntfnsByConfirmHeight[confHeight] = ntfnSet n.ntfnsByConfirmHeight[confHeight] = ntfnSet
} }
ntfnSet[ntfn] = struct{}{} ntfnSet[ntfn] = struct{}{}
// We'll also send an update to the client of how many // We'll also send an update to the client of how many
// confirmations are left for the transaction to be // confirmations are left for the transaction to be
// confirmed. // confirmed.
numConfsLeft := confHeight - tcn.currentHeight numConfsLeft := confHeight - n.currentHeight
select { select {
case ntfn.Event.Updates <- numConfsLeft: case ntfn.Event.Updates <- numConfsLeft:
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
} }
} }
// As a final check, we'll also watch the transaction if it's // As a final check, we'll also watch the transaction if it's
// still possible for it to get reorged out of the chain. // still possible for it to get reorged out of the chain.
blockHeight := details.BlockHeight blockHeight := details.BlockHeight
reorgSafeHeight := blockHeight + tcn.reorgSafetyLimit reorgSafeHeight := blockHeight + n.reorgSafetyLimit
if reorgSafeHeight > tcn.currentHeight { if reorgSafeHeight > n.currentHeight {
txSet, exists := tcn.txsByInitialHeight[blockHeight] txSet, exists := n.txsByInitialHeight[blockHeight]
if !exists { if !exists {
txSet = make(map[chainhash.Hash]struct{}) txSet = make(map[chainhash.Hash]struct{})
tcn.txsByInitialHeight[blockHeight] = txSet n.txsByInitialHeight[blockHeight] = txSet
} }
txSet[*ntfn.TxID] = struct{}{} txSet[*ntfn.TxID] = struct{}{}
} }
@ -466,25 +466,25 @@ func (tcn *TxConfNotifier) dispatchConfDetails(
// Also, if any watched transactions now have the required number of // Also, if any watched transactions now have the required number of
// confirmations as a result of this block being connected, this dispatches // confirmations as a result of this block being connected, this dispatches
// notifications. // notifications.
func (tcn *TxConfNotifier) ConnectTip(blockHash *chainhash.Hash, func (n *TxNotifier) ConnectTip(blockHash *chainhash.Hash,
blockHeight uint32, txns []*btcutil.Tx) error { blockHeight uint32, txns []*btcutil.Tx) error {
select { select {
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
default: default:
} }
tcn.Lock() n.Lock()
defer tcn.Unlock() defer n.Unlock()
if blockHeight != tcn.currentHeight+1 { if blockHeight != n.currentHeight+1 {
return fmt.Errorf("Received blocks out of order: "+ return fmt.Errorf("Received blocks out of order: "+
"current height=%d, new height=%d", "current height=%d, new height=%d",
tcn.currentHeight, blockHeight) n.currentHeight, blockHeight)
} }
tcn.currentHeight++ n.currentHeight++
tcn.reorgDepth = 0 n.reorgDepth = 0
// Record any newly confirmed transactions by their confirmed height so // Record any newly confirmed transactions by their confirmed height so
// that notifications get dispatched when the transactions reach their // that notifications get dispatched when the transactions reach their
@ -496,7 +496,7 @@ func (tcn *TxConfNotifier) ConnectTip(blockHash *chainhash.Hash,
// Check if we have any pending notifications for this txid. If // Check if we have any pending notifications for this txid. If
// none are found, we can proceed to the next transaction. // none are found, we can proceed to the next transaction.
confSet, ok := tcn.confNotifications[*txHash] confSet, ok := n.confNotifications[*txHash]
if !ok { if !ok {
continue continue
} }
@ -529,20 +529,20 @@ func (tcn *TxConfNotifier) ConnectTip(blockHash *chainhash.Hash,
// confirmations so that we can notify them when // confirmations so that we can notify them when
// expected. // expected.
confHeight := blockHeight + ntfn.NumConfirmations - 1 confHeight := blockHeight + ntfn.NumConfirmations - 1
ntfnSet, exists := tcn.ntfnsByConfirmHeight[confHeight] ntfnSet, exists := n.ntfnsByConfirmHeight[confHeight]
if !exists { if !exists {
ntfnSet = make(map[*ConfNtfn]struct{}) ntfnSet = make(map[*ConfNtfn]struct{})
tcn.ntfnsByConfirmHeight[confHeight] = ntfnSet n.ntfnsByConfirmHeight[confHeight] = ntfnSet
} }
ntfnSet[ntfn] = struct{}{} ntfnSet[ntfn] = struct{}{}
// We'll also note the initial confirmation height in // We'll also note the initial confirmation height in
// order to correctly handle dispatching notifications // order to correctly handle dispatching notifications
// when the transaction gets reorged out of the chain. // when the transaction gets reorged out of the chain.
txSet, exists := tcn.txsByInitialHeight[blockHeight] txSet, exists := n.txsByInitialHeight[blockHeight]
if !exists { if !exists {
txSet = make(map[chainhash.Hash]struct{}) txSet = make(map[chainhash.Hash]struct{})
tcn.txsByInitialHeight[blockHeight] = txSet n.txsByInitialHeight[blockHeight] = txSet
} }
txSet[*txHash] = struct{}{} txSet[*txHash] = struct{}{}
} }
@ -556,11 +556,11 @@ func (tcn *TxConfNotifier) ConnectTip(blockHash *chainhash.Hash,
// this map doesn't tell us whether the transaction has confirmed or // this map doesn't tell us whether the transaction has confirmed or
// not, we'll need to look at txsByInitialHeight to determine so. // not, we'll need to look at txsByInitialHeight to determine so.
var txsToUpdateHints []chainhash.Hash var txsToUpdateHints []chainhash.Hash
for confirmedTx := range tcn.txsByInitialHeight[tcn.currentHeight] { for confirmedTx := range n.txsByInitialHeight[n.currentHeight] {
txsToUpdateHints = append(txsToUpdateHints, confirmedTx) txsToUpdateHints = append(txsToUpdateHints, confirmedTx)
} }
out: out:
for maybeUnconfirmedTx, confSet := range tcn.confNotifications { for maybeUnconfirmedTx, confSet := range n.confNotifications {
// We shouldn't update the confirm hints if we still have a // We shouldn't update the confirm hints if we still have a
// pending rescan in progress. We'll skip writing any for // pending rescan in progress. We'll skip writing any for
// notification sets that haven't reached rescanComplete. // notification sets that haven't reached rescanComplete.
@ -568,7 +568,7 @@ out:
continue continue
} }
for height, confirmedTxs := range tcn.txsByInitialHeight { for height, confirmedTxs := range n.txsByInitialHeight {
// Skip the transactions that confirmed at the new block // Skip the transactions that confirmed at the new block
// height as those have already been added. // height as those have already been added.
if height == blockHeight { if height == blockHeight {
@ -585,14 +585,14 @@ out:
} }
if len(txsToUpdateHints) > 0 { if len(txsToUpdateHints) > 0 {
err := tcn.hintCache.CommitConfirmHint( err := n.hintCache.CommitConfirmHint(
tcn.currentHeight, txsToUpdateHints..., n.currentHeight, txsToUpdateHints...,
) )
if err != nil { if err != nil {
// The error is not fatal, so we should not return an // The error is not fatal, so we should not return an
// error to the caller. // error to the caller.
Log.Errorf("Unable to update confirm hint to %d for "+ Log.Errorf("Unable to update confirm hint to %d for "+
"%v: %v", tcn.currentHeight, txsToUpdateHints, "%v: %v", n.currentHeight, txsToUpdateHints,
err) err)
} }
} }
@ -600,9 +600,9 @@ out:
// Next, we'll dispatch an update to all of the notification clients for // Next, we'll dispatch an update to all of the notification clients for
// our watched transactions with the number of confirmations left at // our watched transactions with the number of confirmations left at
// this new height. // this new height.
for _, txHashes := range tcn.txsByInitialHeight { for _, txHashes := range n.txsByInitialHeight {
for txHash := range txHashes { for txHash := range txHashes {
confSet := tcn.confNotifications[txHash] confSet := n.confNotifications[txHash]
for _, ntfn := range confSet.ntfns { for _, ntfn := range confSet.ntfns {
txConfHeight := confSet.details.BlockHeight + txConfHeight := confSet.details.BlockHeight +
ntfn.NumConfirmations - 1 ntfn.NumConfirmations - 1
@ -619,8 +619,8 @@ out:
select { select {
case ntfn.Event.Updates <- numConfsLeft: case ntfn.Event.Updates <- numConfsLeft:
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
} }
} }
} }
@ -628,8 +628,8 @@ out:
// Then, we'll dispatch notifications for all the transactions that have // Then, we'll dispatch notifications for all the transactions that have
// become confirmed at this new block height. // become confirmed at this new block height.
for ntfn := range tcn.ntfnsByConfirmHeight[blockHeight] { for ntfn := range n.ntfnsByConfirmHeight[blockHeight] {
confSet := tcn.confNotifications[*ntfn.TxID] confSet := n.confNotifications[*ntfn.TxID]
Log.Infof("Dispatching %v conf notification for %v", Log.Infof("Dispatching %v conf notification for %v",
ntfn.NumConfirmations, ntfn.TxID) ntfn.NumConfirmations, ntfn.TxID)
@ -637,21 +637,21 @@ out:
select { select {
case ntfn.Event.Confirmed <- confSet.details: case ntfn.Event.Confirmed <- confSet.details:
ntfn.dispatched = true ntfn.dispatched = true
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
} }
} }
delete(tcn.ntfnsByConfirmHeight, tcn.currentHeight) delete(n.ntfnsByConfirmHeight, n.currentHeight)
// Clear entries from confNotifications and confTxsByInitialHeight. We // Clear entries from confNotifications and confTxsByInitialHeight. We
// assume that reorgs deeper than the reorg safety limit do not happen, // assume that reorgs deeper than the reorg safety limit do not happen,
// so we can clear out entries for the block that is now mature. // so we can clear out entries for the block that is now mature.
if tcn.currentHeight >= tcn.reorgSafetyLimit { if n.currentHeight >= n.reorgSafetyLimit {
matureBlockHeight := tcn.currentHeight - tcn.reorgSafetyLimit matureBlockHeight := n.currentHeight - n.reorgSafetyLimit
for txHash := range tcn.txsByInitialHeight[matureBlockHeight] { for txHash := range n.txsByInitialHeight[matureBlockHeight] {
delete(tcn.confNotifications, txHash) delete(n.confNotifications, txHash)
} }
delete(tcn.txsByInitialHeight, matureBlockHeight) delete(n.txsByInitialHeight, matureBlockHeight)
} }
return nil return nil
@ -661,47 +661,47 @@ out:
// a chain reorganization. If any watched transactions were included in this // a chain reorganization. If any watched transactions were included in this
// block, internal structures are updated to ensure a confirmation notification // block, internal structures are updated to ensure a confirmation notification
// is not sent unless the transaction is included in the new chain. // is not sent unless the transaction is included in the new chain.
func (tcn *TxConfNotifier) DisconnectTip(blockHeight uint32) error { func (n *TxNotifier) DisconnectTip(blockHeight uint32) error {
select { select {
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
default: default:
} }
tcn.Lock() n.Lock()
defer tcn.Unlock() defer n.Unlock()
if blockHeight != tcn.currentHeight { if blockHeight != n.currentHeight {
return fmt.Errorf("Received blocks out of order: "+ return fmt.Errorf("Received blocks out of order: "+
"current height=%d, disconnected height=%d", "current height=%d, disconnected height=%d",
tcn.currentHeight, blockHeight) n.currentHeight, blockHeight)
} }
tcn.currentHeight-- n.currentHeight--
tcn.reorgDepth++ n.reorgDepth++
// Rewind the height hint for all watched transactions. // Rewind the height hint for all watched transactions.
var txs []chainhash.Hash var txs []chainhash.Hash
for tx := range tcn.confNotifications { for tx := range n.confNotifications {
txs = append(txs, tx) txs = append(txs, tx)
} }
err := tcn.hintCache.CommitConfirmHint(tcn.currentHeight, txs...) err := n.hintCache.CommitConfirmHint(n.currentHeight, txs...)
if err != nil { if err != nil {
Log.Errorf("Unable to update confirm hint to %d for %v: %v", Log.Errorf("Unable to update confirm hint to %d for %v: %v",
tcn.currentHeight, txs, err) n.currentHeight, txs, err)
return err return err
} }
// We'll go through all of our watched transactions and attempt to drain // We'll go through all of our watched transactions and attempt to drain
// their notification channels to ensure sending notifications to the // their notification channels to ensure sending notifications to the
// clients is always non-blocking. // clients is always non-blocking.
for initialHeight, txHashes := range tcn.txsByInitialHeight { for initialHeight, txHashes := range n.txsByInitialHeight {
for txHash := range txHashes { for txHash := range txHashes {
// If the transaction has been reorged out of the chain, // If the transaction has been reorged out of the chain,
// we'll make sure to remove the cached confirmation // we'll make sure to remove the cached confirmation
// details to prevent notifying clients with old // details to prevent notifying clients with old
// information. // information.
confSet := tcn.confNotifications[txHash] confSet := n.confNotifications[txHash]
if initialHeight == blockHeight { if initialHeight == blockHeight {
confSet.details = nil confSet.details = nil
} }
@ -712,8 +712,8 @@ func (tcn *TxConfNotifier) DisconnectTip(blockHeight uint32) error {
// Updates channel are always non-blocking. // Updates channel are always non-blocking.
select { select {
case <-ntfn.Event.Updates: case <-ntfn.Event.Updates:
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
default: default:
} }
@ -722,7 +722,7 @@ func (tcn *TxConfNotifier) DisconnectTip(blockHeight uint32) error {
// disconnected. If it was, we'll need to // disconnected. If it was, we'll need to
// dispatch a reorg notification to the client. // dispatch a reorg notification to the client.
if initialHeight == blockHeight { if initialHeight == blockHeight {
err := tcn.dispatchConfReorg( err := n.dispatchConfReorg(
ntfn, blockHeight, ntfn, blockHeight,
) )
if err != nil { if err != nil {
@ -735,7 +735,7 @@ func (tcn *TxConfNotifier) DisconnectTip(blockHeight uint32) error {
// Finally, we can remove the transactions we're currently watching that // Finally, we can remove the transactions we're currently watching that
// were included in this block height. // were included in this block height.
delete(tcn.txsByInitialHeight, blockHeight) delete(n.txsByInitialHeight, blockHeight)
return nil return nil
} }
@ -744,16 +744,16 @@ func (tcn *TxConfNotifier) DisconnectTip(blockHeight uint32) error {
// confirmation notification was already delivered. // confirmation notification was already delivered.
// //
// NOTE: This must be called with the TxNotifier's lock held. // NOTE: This must be called with the TxNotifier's lock held.
func (tcn *TxConfNotifier) dispatchConfReorg( func (n *TxNotifier) dispatchConfReorg(ntfn *ConfNtfn,
ntfn *ConfNtfn, heightDisconnected uint32) error { heightDisconnected uint32) error {
// If the transaction's confirmation notification has yet to be // If the transaction's confirmation notification has yet to be
// dispatched, we'll need to clear its entry within the // dispatched, we'll need to clear its entry within the
// ntfnsByConfirmHeight index to prevent from notifiying the client once // ntfnsByConfirmHeight index to prevent from notifying the client once
// the notifier reaches the confirmation height. // the notifier reaches the confirmation height.
if !ntfn.dispatched { if !ntfn.dispatched {
confHeight := heightDisconnected + ntfn.NumConfirmations - 1 confHeight := heightDisconnected + ntfn.NumConfirmations - 1
ntfnSet, exists := tcn.ntfnsByConfirmHeight[confHeight] ntfnSet, exists := n.ntfnsByConfirmHeight[confHeight]
if exists { if exists {
delete(ntfnSet, ntfn) delete(ntfnSet, ntfn)
} }
@ -765,8 +765,8 @@ func (tcn *TxConfNotifier) dispatchConfReorg(
// ensure sends to the Confirmed channel are always non-blocking. // ensure sends to the Confirmed channel are always non-blocking.
select { select {
case <-ntfn.Event.Confirmed: case <-ntfn.Event.Confirmed:
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
default: default:
} }
@ -775,24 +775,24 @@ func (tcn *TxConfNotifier) dispatchConfReorg(
// Send a negative confirmation notification to the client indicating // Send a negative confirmation notification to the client indicating
// how many blocks have been disconnected successively. // how many blocks have been disconnected successively.
select { select {
case ntfn.Event.NegativeConf <- int32(tcn.reorgDepth): case ntfn.Event.NegativeConf <- int32(n.reorgDepth):
case <-tcn.quit: case <-n.quit:
return ErrTxConfNotifierExiting return ErrTxNotifierExiting
} }
return nil return nil
} }
// TearDown is to be called when the owner of the TxConfNotifier is exiting. // TearDown is to be called when the owner of the TxNotifier is exiting. This
// This closes the event channels of all registered notifications that have // closes the event channels of all registered notifications that have not been
// not been dispatched yet. // dispatched yet.
func (tcn *TxConfNotifier) TearDown() { func (n *TxNotifier) TearDown() {
tcn.Lock() n.Lock()
defer tcn.Unlock() defer n.Unlock()
close(tcn.quit) close(n.quit)
for _, confSet := range tcn.confNotifications { for _, confSet := range n.confNotifications {
for _, ntfn := range confSet.ntfns { for _, ntfn := range confSet.ntfns {
if ntfn.dispatched { if ntfn.dispatched {
continue continue

123
chainntnfs/txnotifier_test.go
View File

@ -96,8 +96,8 @@ func newMockHintCache() *mockHintCache {
} }
} }
// TestTxConfFutureDispatch tests that the TxConfNotifier dispatches // TestTxConfFutureDispatch tests that the TxNotifier dispatches registered
// registered notifications when the transaction confirms after registration. // notifications when the transaction confirms after registration.
func TestTxConfFutureDispatch(t *testing.T) { func TestTxConfFutureDispatch(t *testing.T) {
t.Parallel() t.Parallel()
@ -113,10 +113,10 @@ func TestTxConfFutureDispatch(t *testing.T) {
) )
hintCache := newMockHintCache() hintCache := newMockHintCache()
tcn := chainntnfs.NewTxConfNotifier(10, 100, hintCache) n := chainntnfs.NewTxNotifier(10, 100, hintCache)
// Create the test transactions and register them with the // Create the test transactions and register them with the TxNotifier
// TxConfNotifier before including them in a block to receive future // before including them in a block to receive future
// notifications. // notifications.
tx1Hash := tx1.TxHash() tx1Hash := tx1.TxHash()
ntfn1 := chainntnfs.ConfNtfn{ ntfn1 := chainntnfs.ConfNtfn{
@ -124,7 +124,7 @@ func TestTxConfFutureDispatch(t *testing.T) {
NumConfirmations: tx1NumConfs, NumConfirmations: tx1NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx1NumConfs), Event: chainntnfs.NewConfirmationEvent(tx1NumConfs),
} }
if _, err := tcn.Register(&ntfn1); err != nil { if _, err := n.Register(&ntfn1); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
@ -134,7 +134,7 @@ func TestTxConfFutureDispatch(t *testing.T) {
NumConfirmations: tx2NumConfs, NumConfirmations: tx2NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx2NumConfs), Event: chainntnfs.NewConfirmationEvent(tx2NumConfs),
} }
if _, err := tcn.Register(&ntfn2); err != nil { if _, err := n.Register(&ntfn2); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
@ -156,13 +156,13 @@ func TestTxConfFutureDispatch(t *testing.T) {
default: default:
} }
// Include the transactions in a block and add it to the TxConfNotifier. // Include the transactions in a block and add it to the TxNotifier.
// This should confirm tx1, but not tx2. // This should confirm tx1, but not tx2.
block1 := btcutil.NewBlock(&wire.MsgBlock{ block1 := btcutil.NewBlock(&wire.MsgBlock{
Transactions: []*wire.MsgTx{&tx1, &tx2, &tx3}, Transactions: []*wire.MsgTx{&tx1, &tx2, &tx3},
}) })
err := tcn.ConnectTip( err := n.ConnectTip(
block1.Hash(), 11, block1.Transactions(), block1.Hash(), 11, block1.Transactions(),
) )
if err != nil { if err != nil {
@ -219,13 +219,13 @@ func TestTxConfFutureDispatch(t *testing.T) {
default: default:
} }
// Create a new block and add it to the TxConfNotifier at the next // Create a new block and add it to the TxNotifier at the next height.
// height. This should confirm tx2. // This should confirm tx2.
block2 := btcutil.NewBlock(&wire.MsgBlock{ block2 := btcutil.NewBlock(&wire.MsgBlock{
Transactions: []*wire.MsgTx{&tx3}, Transactions: []*wire.MsgTx{&tx3},
}) })
err = tcn.ConnectTip(block2.Hash(), 12, block2.Transactions()) err = n.ConnectTip(block2.Hash(), 12, block2.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -269,9 +269,8 @@ func TestTxConfFutureDispatch(t *testing.T) {
} }
} }
// TestTxConfHistoricalDispatch tests that the TxConfNotifier dispatches // TestTxConfHistoricalDispatch tests that the TxNotifier dispatches registered
// registered notifications when the transaction is confirmed before // notifications when the transaction is confirmed before registration.
// registration.
func TestTxConfHistoricalDispatch(t *testing.T) { func TestTxConfHistoricalDispatch(t *testing.T) {
t.Parallel() t.Parallel()
@ -287,9 +286,9 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
) )
hintCache := newMockHintCache() hintCache := newMockHintCache()
tcn := chainntnfs.NewTxConfNotifier(10, 100, hintCache) n := chainntnfs.NewTxNotifier(10, 100, hintCache)
// Create the test transactions at a height before the TxConfNotifier's // Create the test transactions at a height before the TxNotifier's
// starting height so that they are confirmed once registering them. // starting height so that they are confirmed once registering them.
tx1Hash := tx1.TxHash() tx1Hash := tx1.TxHash()
ntfn1 := chainntnfs.ConfNtfn{ ntfn1 := chainntnfs.ConfNtfn{
@ -298,7 +297,7 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
NumConfirmations: tx1NumConfs, NumConfirmations: tx1NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx1NumConfs), Event: chainntnfs.NewConfirmationEvent(tx1NumConfs),
} }
if _, err := tcn.Register(&ntfn1); err != nil { if _, err := n.Register(&ntfn1); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
@ -309,7 +308,7 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
NumConfirmations: tx2NumConfs, NumConfirmations: tx2NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx2NumConfs), Event: chainntnfs.NewConfirmationEvent(tx2NumConfs),
} }
if _, err := tcn.Register(&ntfn2); err != nil { if _, err := n.Register(&ntfn2); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
@ -320,7 +319,7 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
BlockHeight: 9, BlockHeight: 9,
TxIndex: 1, TxIndex: 1,
} }
err := tcn.UpdateConfDetails(tx1Hash, &txConf1) err := n.UpdateConfDetails(tx1Hash, &txConf1)
if err != nil { if err != nil {
t.Fatalf("unable to update conf details: %v", err) t.Fatalf("unable to update conf details: %v", err)
} }
@ -353,7 +352,7 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
BlockHeight: 9, BlockHeight: 9,
TxIndex: 2, TxIndex: 2,
} }
err = tcn.UpdateConfDetails(tx2Hash, &txConf2) err = n.UpdateConfDetails(tx2Hash, &txConf2)
if err != nil { if err != nil {
t.Fatalf("unable to update conf details: %v", err) t.Fatalf("unable to update conf details: %v", err)
} }
@ -375,13 +374,13 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
default: default:
} }
// Create a new block and add it to the TxConfNotifier at the next // Create a new block and add it to the TxNotifier at the next height.
// height. This should confirm tx2. // This should confirm tx2.
block := btcutil.NewBlock(&wire.MsgBlock{ block := btcutil.NewBlock(&wire.MsgBlock{
Transactions: []*wire.MsgTx{&tx3}, Transactions: []*wire.MsgTx{&tx3},
}) })
err = tcn.ConnectTip(block.Hash(), 11, block.Transactions()) err = n.ConnectTip(block.Hash(), 11, block.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -420,7 +419,7 @@ func TestTxConfHistoricalDispatch(t *testing.T) {
} }
} }
// TestTxConfChainReorg tests that TxConfNotifier dispatches Confirmed and // TestTxConfChainReorg tests that TxNotifier dispatches Confirmed and
// NegativeConf notifications appropriately when there is a chain // NegativeConf notifications appropriately when there is a chain
// reorganization. // reorganization.
func TestTxConfChainReorg(t *testing.T) { func TestTxConfChainReorg(t *testing.T) {
@ -439,7 +438,7 @@ func TestTxConfChainReorg(t *testing.T) {
) )
hintCache := newMockHintCache() hintCache := newMockHintCache()
tcn := chainntnfs.NewTxConfNotifier(7, 100, hintCache) n := chainntnfs.NewTxNotifier(7, 100, hintCache)
// Tx 1 will be confirmed in block 9 and requires 2 confs. // Tx 1 will be confirmed in block 9 and requires 2 confs.
tx1Hash := tx1.TxHash() tx1Hash := tx1.TxHash()
@ -448,11 +447,11 @@ func TestTxConfChainReorg(t *testing.T) {
NumConfirmations: tx1NumConfs, NumConfirmations: tx1NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx1NumConfs), Event: chainntnfs.NewConfirmationEvent(tx1NumConfs),
} }
if _, err := tcn.Register(&ntfn1); err != nil { if _, err := n.Register(&ntfn1); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
if err := tcn.UpdateConfDetails(*ntfn1.TxID, nil); err != nil { if err := n.UpdateConfDetails(*ntfn1.TxID, nil); err != nil {
t.Fatalf("unable to deliver conf details: %v", err) t.Fatalf("unable to deliver conf details: %v", err)
} }
@ -463,11 +462,11 @@ func TestTxConfChainReorg(t *testing.T) {
NumConfirmations: tx2NumConfs, NumConfirmations: tx2NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx2NumConfs), Event: chainntnfs.NewConfirmationEvent(tx2NumConfs),
} }
if _, err := tcn.Register(&ntfn2); err != nil { if _, err := n.Register(&ntfn2); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
if err := tcn.UpdateConfDetails(*ntfn2.TxID, nil); err != nil { if err := n.UpdateConfDetails(*ntfn2.TxID, nil); err != nil {
t.Fatalf("unable to deliver conf details: %v", err) t.Fatalf("unable to deliver conf details: %v", err)
} }
@ -478,11 +477,11 @@ func TestTxConfChainReorg(t *testing.T) {
NumConfirmations: tx3NumConfs, NumConfirmations: tx3NumConfs,
Event: chainntnfs.NewConfirmationEvent(tx3NumConfs), Event: chainntnfs.NewConfirmationEvent(tx3NumConfs),
} }
if _, err := tcn.Register(&ntfn3); err != nil { if _, err := n.Register(&ntfn3); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
if err := tcn.UpdateConfDetails(*ntfn3.TxID, nil); err != nil { if err := n.UpdateConfDetails(*ntfn3.TxID, nil); err != nil {
t.Fatalf("unable to deliver conf details: %v", err) t.Fatalf("unable to deliver conf details: %v", err)
} }
@ -490,11 +489,11 @@ func TestTxConfChainReorg(t *testing.T) {
block1 := btcutil.NewBlock(&wire.MsgBlock{ block1 := btcutil.NewBlock(&wire.MsgBlock{
Transactions: []*wire.MsgTx{&tx1}, Transactions: []*wire.MsgTx{&tx1},
}) })
err := tcn.ConnectTip(nil, 8, block1.Transactions()) err := n.ConnectTip(nil, 8, block1.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
err = tcn.ConnectTip(nil, 9, nil) err = n.ConnectTip(nil, 9, nil)
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -502,7 +501,7 @@ func TestTxConfChainReorg(t *testing.T) {
block2 := btcutil.NewBlock(&wire.MsgBlock{ block2 := btcutil.NewBlock(&wire.MsgBlock{
Transactions: []*wire.MsgTx{&tx2, &tx3}, Transactions: []*wire.MsgTx{&tx2, &tx3},
}) })
err = tcn.ConnectTip(nil, 10, block2.Transactions()) err = n.ConnectTip(nil, 10, block2.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -559,17 +558,17 @@ func TestTxConfChainReorg(t *testing.T) {
// The block that included tx2 and tx3 is disconnected and two next // The block that included tx2 and tx3 is disconnected and two next
// blocks without them are connected. // blocks without them are connected.
err = tcn.DisconnectTip(10) err = n.DisconnectTip(10)
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
err = tcn.ConnectTip(nil, 10, nil) err = n.ConnectTip(nil, 10, nil)
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
err = tcn.ConnectTip(nil, 11, nil) err = n.ConnectTip(nil, 11, nil)
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -617,12 +616,12 @@ func TestTxConfChainReorg(t *testing.T) {
}) })
block4 := btcutil.NewBlock(&wire.MsgBlock{}) block4 := btcutil.NewBlock(&wire.MsgBlock{})
err = tcn.ConnectTip(block3.Hash(), 12, block3.Transactions()) err = n.ConnectTip(block3.Hash(), 12, block3.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
err = tcn.ConnectTip(block4.Hash(), 13, block4.Transactions()) err = n.ConnectTip(block4.Hash(), 13, block4.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -701,9 +700,9 @@ func TestTxConfHeightHintCache(t *testing.T) {
tx2Height = 203 tx2Height = 203
) )
// Initialize our TxConfNotifier instance backed by a height hint cache. // Initialize our TxNotifier instance backed by a height hint cache.
hintCache := newMockHintCache() hintCache := newMockHintCache()
tcn := chainntnfs.NewTxConfNotifier( n := chainntnfs.NewTxNotifier(
startingHeight, 100, hintCache, startingHeight, 100, hintCache,
) )
@ -724,10 +723,10 @@ func TestTxConfHeightHintCache(t *testing.T) {
Event: chainntnfs.NewConfirmationEvent(2), Event: chainntnfs.NewConfirmationEvent(2),
} }
if _, err := tcn.Register(ntfn1); err != nil { if _, err := n.Register(ntfn1); err != nil {
t.Fatalf("unable to register tx1: %v", err) t.Fatalf("unable to register tx1: %v", err)
} }
if _, err := tcn.Register(ntfn2); err != nil { if _, err := n.Register(ntfn2); err != nil {
t.Fatalf("unable to register tx2: %v", err) t.Fatalf("unable to register tx2: %v", err)
} }
@ -754,7 +753,7 @@ func TestTxConfHeightHintCache(t *testing.T) {
Transactions: []*wire.MsgTx{&txDummy}, Transactions: []*wire.MsgTx{&txDummy},
}) })
err = tcn.ConnectTip( err = n.ConnectTip(
block1.Hash(), txDummyHeight, block1.Transactions(), block1.Hash(), txDummyHeight, block1.Transactions(),
) )
if err != nil { if err != nil {
@ -781,10 +780,10 @@ func TestTxConfHeightHintCache(t *testing.T) {
// Now, update the conf details reporting that the neither txn was found // Now, update the conf details reporting that the neither txn was found
// in the historical dispatch. // in the historical dispatch.
if err := tcn.UpdateConfDetails(tx1Hash, nil); err != nil { if err := n.UpdateConfDetails(tx1Hash, nil); err != nil {
t.Fatalf("unable to update conf details: %v", err) t.Fatalf("unable to update conf details: %v", err)
} }
if err := tcn.UpdateConfDetails(tx2Hash, nil); err != nil { if err := n.UpdateConfDetails(tx2Hash, nil); err != nil {
t.Fatalf("unable to update conf details: %v", err) t.Fatalf("unable to update conf details: %v", err)
} }
@ -794,7 +793,7 @@ func TestTxConfHeightHintCache(t *testing.T) {
Transactions: []*wire.MsgTx{&tx1}, Transactions: []*wire.MsgTx{&tx1},
}) })
err = tcn.ConnectTip( err = n.ConnectTip(
block2.Hash(), tx1Height, block2.Transactions(), block2.Hash(), tx1Height, block2.Transactions(),
) )
if err != nil { if err != nil {
@ -828,7 +827,7 @@ func TestTxConfHeightHintCache(t *testing.T) {
Transactions: []*wire.MsgTx{&tx2}, Transactions: []*wire.MsgTx{&tx2},
}) })
err = tcn.ConnectTip( err = n.ConnectTip(
block3.Hash(), tx2Height, block3.Transactions(), block3.Hash(), tx2Height, block3.Transactions(),
) )
if err != nil { if err != nil {
@ -858,7 +857,7 @@ func TestTxConfHeightHintCache(t *testing.T) {
// Finally, we'll attempt do disconnect the last block in order to // Finally, we'll attempt do disconnect the last block in order to
// simulate a chain reorg. // simulate a chain reorg.
if err := tcn.DisconnectTip(tx2Height); err != nil { if err := n.DisconnectTip(tx2Height); err != nil {
t.Fatalf("Failed to disconnect block: %v", err) t.Fatalf("Failed to disconnect block: %v", err)
} }
@ -894,20 +893,20 @@ func TestTxConfTearDown(t *testing.T) {
) )
hintCache := newMockHintCache() hintCache := newMockHintCache()
tcn := chainntnfs.NewTxConfNotifier(10, 100, hintCache) n := chainntnfs.NewTxNotifier(10, 100, hintCache)
// Create the test transactions and register them with the // Create the test transactions and register them with the TxNotifier to
// TxConfNotifier to receive notifications. // receive notifications.
tx1Hash := tx1.TxHash() tx1Hash := tx1.TxHash()
ntfn1 := chainntnfs.ConfNtfn{ ntfn1 := chainntnfs.ConfNtfn{
TxID: &tx1Hash, TxID: &tx1Hash,
NumConfirmations: 1, NumConfirmations: 1,
Event: chainntnfs.NewConfirmationEvent(1), Event: chainntnfs.NewConfirmationEvent(1),
} }
if _, err := tcn.Register(&ntfn1); err != nil { if _, err := n.Register(&ntfn1); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
if err := tcn.UpdateConfDetails(*ntfn1.TxID, nil); err != nil { if err := n.UpdateConfDetails(*ntfn1.TxID, nil); err != nil {
t.Fatalf("unable to update conf details: %v", err) t.Fatalf("unable to update conf details: %v", err)
} }
@ -917,20 +916,20 @@ func TestTxConfTearDown(t *testing.T) {
NumConfirmations: 2, NumConfirmations: 2,
Event: chainntnfs.NewConfirmationEvent(2), Event: chainntnfs.NewConfirmationEvent(2),
} }
if _, err := tcn.Register(&ntfn2); err != nil { if _, err := n.Register(&ntfn2); err != nil {
t.Fatalf("unable to register ntfn: %v", err) t.Fatalf("unable to register ntfn: %v", err)
} }
if err := tcn.UpdateConfDetails(*ntfn2.TxID, nil); err != nil { if err := n.UpdateConfDetails(*ntfn2.TxID, nil); err != nil {
t.Fatalf("unable to update conf details: %v", err) t.Fatalf("unable to update conf details: %v", err)
} }
// Include the transactions in a block and add it to the TxConfNotifier. // Include the transactions in a block and add it to the TxNotifier.
// This should confirm tx1, but not tx2. // This should confirm tx1, but not tx2.
block := btcutil.NewBlock(&wire.MsgBlock{ block := btcutil.NewBlock(&wire.MsgBlock{
Transactions: []*wire.MsgTx{&tx1, &tx2}, Transactions: []*wire.MsgTx{&tx1, &tx2},
}) })
err := tcn.ConnectTip(block.Hash(), 11, block.Transactions()) err := n.ConnectTip(block.Hash(), 11, block.Transactions())
if err != nil { if err != nil {
t.Fatalf("Failed to connect block: %v", err) t.Fatalf("Failed to connect block: %v", err)
} }
@ -966,10 +965,10 @@ func TestTxConfTearDown(t *testing.T) {
// The notification channels should be closed for notifications that // The notification channels should be closed for notifications that
// have not been dispatched yet, so we should not expect to receive any // have not been dispatched yet, so we should not expect to receive any
// more updates. // more updates.
tcn.TearDown() n.TearDown()
// tx1 should not receive any more updates because it has already been // tx1 should not receive any more updates because it has already been
// confirmed and the TxConfNotifier has been shut down. // confirmed and the TxNotifier has been shut down.
select { select {
case <-ntfn1.Event.Updates: case <-ntfn1.Event.Updates:
t.Fatal("Received unexpected confirmation update for tx1") t.Fatal("Received unexpected confirmation update for tx1")
@ -979,7 +978,7 @@ func TestTxConfTearDown(t *testing.T) {
} }
// tx2 should not receive any more updates after the notifications // tx2 should not receive any more updates after the notifications
// channels have been closed and the TxConfNotifier shut down. // channels have been closed and the TxNotifier shut down.
select { select {
case _, more := <-ntfn2.Event.Updates: case _, more := <-ntfn2.Event.Updates:
if more { if more {