17685638204

Committed 12 Sep 2025 08:45PM UTC coverage: 66.624% (-0.03%) from 66.651%

Build # 17685638204

Build Type

Pull #10015

github

Committed by

web-flow

Commit Message

Merge 0f62b1831 into 5082566ed

Pull Request Pull Request #10015: graph/db: add zombie channels cleanup routine

Run Details

59 of 67 new or added lines in 2 files covered. (88.06%)

126 existing lines in 24 files now uncovered.

136275 of 204543 relevant lines covered (66.62%)

21395.28 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.31

/chainntnfs/txnotifier.go

package chainntnfs

import (
        "bytes"
        "errors"
        "fmt"
        "sync"
        "sync/atomic"

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/chaincfg/chainhash"
        "github.com/btcsuite/btcd/txscript"
        "github.com/btcsuite/btcd/wire"
)

const (
        // ReorgSafetyLimit is the chain depth beyond which it is assumed a
        // block will not be reorganized out of the chain. This is used to
        // determine when to prune old confirmation requests so that reorgs are
        // handled correctly. The average number of blocks in a day is a
        // reasonable value to use.
        ReorgSafetyLimit = 144

        // MaxNumConfs is the maximum number of confirmations that can be
        // requested on a transaction.
        MaxNumConfs = ReorgSafetyLimit
)

var (
        // ZeroHash is the value that should be used as the txid when
        // registering for the confirmation of a script on-chain. This allows
        // the notifier to match _and_ dispatch upon the inclusion of the script
        // on-chain, rather than the txid.
        ZeroHash chainhash.Hash

        // ZeroOutPoint is the value that should be used as the outpoint when
        // registering for the spend of a script on-chain. This allows the
        // notifier to match _and_ dispatch upon detecting the spend of the
        // script on-chain, rather than the outpoint.
        ZeroOutPoint wire.OutPoint

        // zeroV1KeyPush is a pkScript that pushes an all-zero 32-byte Taproot
        // SegWit v1 key to the stack.
        zeroV1KeyPush = [34]byte{
                txscript.OP_1, txscript.OP_DATA_32, // 32 byte of zeroes here
        }

        // ZeroTaprootPkScript is the parsed txscript.PkScript of an empty
        // Taproot SegWit v1 key being pushed to the stack. This allows the
        // notifier to match _and_ dispatch upon detecting the spend of the
        // outpoint on-chain, rather than the pkScript (which cannot be derived
        // from the witness alone in the SegWit v1 case).
        ZeroTaprootPkScript, _ = txscript.ParsePkScript(zeroV1KeyPush[:])
)

var (
        // ErrTxNotifierExiting is an error returned when attempting to interact
        // with the TxNotifier but it been shut down.
        ErrTxNotifierExiting = errors.New("TxNotifier is exiting")

        // 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)

        // ErrEmptyWitnessStack is returned when a spending transaction has an
        // empty witness stack. More details in,
        // - https://github.com/bitcoin/bitcoin/issues/28730
        ErrEmptyWitnessStack = errors.New("witness stack is empty")
)

// rescanState indicates the progression of a registration before the notifier
// can begin dispatching confirmations at tip.
type rescanState byte

const (
        // rescanNotStarted is the initial state, denoting that a historical
        // dispatch may be required.
        rescanNotStarted rescanState = iota

        // rescanPending indicates that a dispatch has already been made, and we
        // are waiting for its completion. No other rescans should be dispatched
        // while in this state.
        rescanPending

        // rescanComplete signals either that a rescan was dispatched and has
        // completed, or that we began watching at tip immediately. In either
        // case, the notifier can only dispatch notifications from tip when in
        // this state.
        rescanComplete
)

// confNtfnSet holds all known, registered confirmation notifications for a
// txid/output script. If duplicates notifications are requested, only one
// historical dispatch will be spawned to ensure redundant scans are not
// permitted. A single conf detail will be constructed and dispatched to all
// interested
// clients.
type confNtfnSet struct {
        // ntfns keeps tracks of all the active client notification requests for
        // a transaction/output script
        ntfns map[uint64]*ConfNtfn

        // rescanStatus represents the current rescan state for the
        // transaction/output script.
        rescanStatus rescanState

        // details serves as a cache of the confirmation details of a
        // transaction that we'll use to determine if a transaction/output
        // script has already confirmed at the time of registration.
        // details is also used to make sure that in case of an address reuse
        // (funds sent to a previously confirmed script) no additional
        // notification is registered which would lead to an inconsistent state.
        details *TxConfirmation
}

// newConfNtfnSet constructs a fresh confNtfnSet for a group of clients
// interested in a notification for a particular txid.
func newConfNtfnSet() *confNtfnSet {
        return &confNtfnSet{
                ntfns:        make(map[uint64]*ConfNtfn),
                rescanStatus: rescanNotStarted,
        }
}

// spendNtfnSet holds all known, registered spend notifications for a spend
// request (outpoint/output script). If duplicate notifications are requested,
// only one historical dispatch will be spawned to ensure redundant scans are
// not permitted.
type spendNtfnSet struct {
        // ntfns keeps tracks of all the active client notification requests for
        // an outpoint/output script.
        ntfns map[uint64]*SpendNtfn

        // rescanStatus represents the current rescan state for the spend
        // request (outpoint/output script).
        rescanStatus rescanState

        // details serves as a cache of the spend details for an outpoint/output
        // script that we'll use to determine if it has already been spent at
        // the time of registration.
        details *SpendDetail
}

// newSpendNtfnSet constructs a new spend notification set.
func newSpendNtfnSet() *spendNtfnSet {
        return &spendNtfnSet{
                ntfns:        make(map[uint64]*SpendNtfn),
                rescanStatus: rescanNotStarted,
        }
}

// ConfRequest encapsulates a request for a confirmation notification of either
// a txid or output script.
type ConfRequest struct {
        // TxID is the hash of the transaction for which confirmation
        // notifications are requested. If set to a zero hash, then a
        // confirmation notification will be dispatched upon inclusion of the
        // _script_, rather than the txid.
        TxID chainhash.Hash

        // PkScript is the public key script of an outpoint created in this
        // transaction.
        PkScript txscript.PkScript
}

// NewConfRequest creates a request for a confirmation notification of either a
// txid or output script. A nil txid or an allocated ZeroHash can be used to
// dispatch the confirmation notification on the script.
func NewConfRequest(txid *chainhash.Hash, pkScript []byte) (ConfRequest, error) {
        var r ConfRequest
        outputScript, err := txscript.ParsePkScript(pkScript)
        if err != nil {
                return r, err
        }

        // We'll only set a txid for which we'll dispatch a confirmation
        // notification on this request if one was provided. Otherwise, we'll
        // default to dispatching on the confirmation of the script instead.
        if txid != nil {
                r.TxID = *txid
        }
        r.PkScript = outputScript

        return r, nil
}

// String returns the string representation of the ConfRequest.
func (r ConfRequest) String() string {
        if r.TxID != ZeroHash {
                return fmt.Sprintf("txid=%v", r.TxID)
        }
        return fmt.Sprintf("script=%v", r.PkScript)
}

// MatchesTx determines whether the given transaction satisfies the confirmation
// request. If the confirmation request is for a script, then we'll check all of
// the outputs of the transaction to determine if it matches. Otherwise, we'll
// match on the txid.
func (r ConfRequest) MatchesTx(tx *wire.MsgTx) bool {
        scriptMatches := func() bool {
                pkScript := r.PkScript.Script()
                for _, txOut := range tx.TxOut {
                        if bytes.Equal(txOut.PkScript, pkScript) {
                                return true
                        }
                }

                return false
        }

        if r.TxID != ZeroHash {
                return r.TxID == tx.TxHash() && scriptMatches()
        }

        return scriptMatches()
}

// ConfNtfn represents a notifier client's request to receive a notification
// once the target transaction/output script gets sufficient confirmations. The
// client is asynchronously notified via the ConfirmationEvent channels.
type ConfNtfn struct {
        // ConfID uniquely identifies the confirmation notification request for
        // the specified transaction/output script.
        ConfID uint64

        // ConfRequest represents either the txid or script we should detect
        // inclusion of within the chain.
        ConfRequest

        // NumConfirmations is the number of confirmations after which the
        // notification is to be sent.
        NumConfirmations uint32

        // Event contains references to the channels that the notifications are
        // to be sent over.
        Event *ConfirmationEvent

        // HeightHint is the minimum height in the chain that we expect to find
        // this txid.
        HeightHint uint32

        // dispatched is false if the confirmed notification has not been sent
        // yet.
        dispatched bool

        // includeBlock is true if the dispatched notification should also have
        // the block included with it.
        includeBlock bool

        // numConfsLeft is the number of confirmations left to be sent to the
        // subscriber.
        numConfsLeft uint32
}

// HistoricalConfDispatch parametrizes a manual rescan for a particular
// transaction/output script. The parameters include the start and end block
// heights specifying the range of blocks to scan.
type HistoricalConfDispatch struct {
        // ConfRequest represents either the txid or script we should detect
        // inclusion of within the chain.
        ConfRequest

        // StartHeight specifies the block height at which to begin the
        // historical rescan.
        StartHeight uint32

        // EndHeight specifies the last block height (inclusive) that the
        // historical scan should consider.
        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 {
        // OutPoint is the outpoint for which a client has requested a spend
        // notification for. If set to a zero outpoint, then a spend
        // notification will be dispatched upon detecting the spend of the
        // _script_, rather than the outpoint.
        OutPoint wire.OutPoint

        // PkScript is the script of the outpoint. If a zero outpoint is set,
        // then this can be an arbitrary script.
        PkScript txscript.PkScript
}

// NewSpendRequest creates a request for a spend notification of either an
// outpoint or output script. A nil outpoint or an allocated ZeroOutPoint can be
// used to dispatch the confirmation notification on the script.
func NewSpendRequest(op *wire.OutPoint, pkScript []byte) (SpendRequest, error) {
        var r SpendRequest
        outputScript, err := txscript.ParsePkScript(pkScript)
        if err != nil {
                return r, err
        }

        // We'll only set an outpoint for which we'll dispatch a spend
        // notification on this request if one was provided. Otherwise, we'll
        // default to dispatching on the spend of the script instead.
        if op != nil {
                r.OutPoint = *op
        }
        r.PkScript = outputScript

        // For Taproot spends we have the main problem that for the key spend
        // path we cannot derive the pkScript from only looking at the input's
        // witness. So we need to rely on the outpoint information alone.
        //
        // TODO(guggero): For script path spends we can derive the pkScript from
        // the witness, since we have the full control block and the spent
        // script available.
        if outputScript.Class() == txscript.WitnessV1TaprootTy {
                if op == nil {
                        return r, fmt.Errorf("cannot register witness v1 " +
                                "spend request without outpoint")
                }

                // We have an outpoint, so we can set the pkScript to an all
                // zero Taproot key that we'll compare this spend request to.
                r.PkScript = ZeroTaprootPkScript
        }

        return r, nil
}

// String returns the string representation of the SpendRequest.
func (r SpendRequest) String() string {
        if r.OutPoint != ZeroOutPoint {
                return fmt.Sprintf("outpoint=%v, script=%v", r.OutPoint,
                        r.PkScript)
        }
        return fmt.Sprintf("outpoint=<zero>, script=%v", r.PkScript)
}

// MatchesTx determines whether the given transaction satisfies the spend
// request. If the spend request is for an outpoint, then we'll check all of
// the outputs being spent by the inputs of the transaction to determine if it
// matches. Otherwise, we'll need to match on the output script being spent, so
// we'll recompute it for each input of the transaction to determine if it
// matches.
func (r SpendRequest) MatchesTx(tx *wire.MsgTx) (bool, uint32, error) {
        if r.OutPoint != ZeroOutPoint {
                for i, txIn := range tx.TxIn {
                        if txIn.PreviousOutPoint == r.OutPoint {
                                return true, uint32(i), nil
                        }
                }

                return false, 0, nil
        }

        for i, txIn := range tx.TxIn {
                pkScript, err := txscript.ComputePkScript(
                        txIn.SignatureScript, txIn.Witness,
                )
                if err == txscript.ErrUnsupportedScriptType {
                        continue
                }
                if err != nil {
                        return false, 0, err
                }

                if bytes.Equal(pkScript.Script(), r.PkScript.Script()) {
                        return true, uint32(i), nil
                }
        }

        return false, 0, nil
}

// SpendNtfn represents a client's request to receive a notification once an
// outpoint/output script has been spent on-chain. The client is asynchronously
// notified via the SpendEvent channels.
type SpendNtfn struct {
        // SpendID uniquely identies the spend notification request for the
        // specified outpoint/output script.
        SpendID uint64

        // SpendRequest represents either the outpoint or script we should
        // detect the spend of.
        SpendRequest

        // Event contains references to the channels that the notifications are
        // to be sent over.
        Event *SpendEvent

        // HeightHint is the earliest height in the chain that we expect to find
        // the spending transaction of the specified outpoint/output script.
        // This value will be overridden by the spend hint cache if it contains
        // an entry for it.
        HeightHint uint32

        // dispatched signals whether a spend notification has been dispatched
        // to the client.
        dispatched bool
}

// HistoricalSpendDispatch parametrizes a manual rescan to determine the
// spending details (if any) of an outpoint/output script. The parameters
// include the start and end block heights specifying the range of blocks to
// scan.
type HistoricalSpendDispatch struct {
        // SpendRequest represents either the outpoint or script we should
        // detect the spend of.
        SpendRequest

        // StartHeight specified the block height at which to begin the
        // historical rescan.
        StartHeight uint32

        // EndHeight specifies the last block height (inclusive) that the
        // historical rescan should consider.
        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.
        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
        // connected and disconnected in order.
        currentHeight uint32

        // reorgSafetyLimit is the chain depth beyond which it is assumed a
        // block will not be reorganized out of the chain. This is used to
        // determine when to prune old notification requests so that reorgs are
        // handled correctly. The coinbase maturity period is a reasonable value
        // to use.
        reorgSafetyLimit uint32

        // reorgDepth is the depth of a chain organization that this system is
        // being informed of. This is incremented as long as a sequence of
        // blocks are disconnected without being interrupted by a new block.
        reorgDepth uint32

        // confNotifications is an index of confirmation notification requests
        // by transaction hash/output script.
        confNotifications map[ConfRequest]*confNtfnSet

        // confsByInitialHeight is an index of watched transactions/output
        // scripts by the height that they are included at in the chain. This
        // is tracked so that incorrect notifications are not sent if a
        // transaction/output script is reorged out of the chain and so that
        // negative confirmations can be recognized.
        confsByInitialHeight map[uint32]map[ConfRequest]struct{}

        // ntfnsByConfirmHeight is an index of notification requests by the
        // height at which the transaction/output script will have sufficient
        // confirmations.
        ntfnsByConfirmHeight map[uint32]map[*ConfNtfn]struct{}

        // spendNotifications is an index of all active notification requests
        // per outpoint/output script.
        spendNotifications map[SpendRequest]*spendNtfnSet

        // spendsByHeight is an index that keeps tracks of the spending height
        // of outpoints/output scripts we are currently tracking notifications
        // for. This is used in order to recover from spending transactions
        // being reorged out of the chain.
        spendsByHeight map[uint32]map[SpendRequest]struct{}

        // confirmHintCache is a cache used to maintain the latest height hints
        // for transactions/output scripts. Each height hint represents the
        // earliest height at which they scripts could have been confirmed
        // within the chain.
        confirmHintCache ConfirmHintCache

        // spendHintCache is a cache used to maintain the latest height hints
        // for outpoints/output scripts. Each height hint represents the
        // earliest height at which they could have been spent within the chain.
        spendHintCache SpendHintCache

        // quit is closed in order to signal that the notifier is gracefully
        // exiting.
        quit chan struct{}

        sync.Mutex
}

// NewTxNotifier creates a TxNotifier. The current height of the blockchain is
// accepted as a parameter. The different hint caches (confirm and spend) are
// used as an optimization in order to retrieve a better starting point when
// dispatching a rescan for a historical event in the chain.
func NewTxNotifier(startHeight uint32, reorgSafetyLimit uint32,
        confirmHintCache ConfirmHintCache,
        spendHintCache SpendHintCache) *TxNotifier {

        return &TxNotifier{
                currentHeight:        startHeight,
                reorgSafetyLimit:     reorgSafetyLimit,
                confNotifications:    make(map[ConfRequest]*confNtfnSet),
                confsByInitialHeight: make(map[uint32]map[ConfRequest]struct{}),
                ntfnsByConfirmHeight: make(map[uint32]map[*ConfNtfn]struct{}),
                spendNotifications:   make(map[SpendRequest]*spendNtfnSet),
                spendsByHeight:       make(map[uint32]map[SpendRequest]struct{}),
                confirmHintCache:     confirmHintCache,
                spendHintCache:       spendHintCache,
                quit:                 make(chan struct{}),
        }
}

// 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,
        opts *NotifierOptions) (*ConfNtfn, error) {

        // 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 == 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)
        return &ConfNtfn{
                ConfID:           confID,
                ConfRequest:      confRequest,
                NumConfirmations: numConfs,
                Event: NewConfirmationEvent(numConfs, func() {
                        n.CancelConf(confRequest, confID)
                }),
                HeightHint:   heightHint,
                includeBlock: opts.IncludeBlock,
                numConfsLeft: numConfs,
        }, 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.
//
// 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(txid *chainhash.Hash, pkScript []byte,
        numConfs, heightHint uint32,
        optFuncs ...NotifierOption) (*ConfRegistration, error) {

        select {
        case <-n.quit:
                return nil, ErrTxNotifierExiting
        default:
        }

        opts := DefaultNotifierOptions()
        for _, optFunc := range optFuncs {
                optFunc(opts)
        }

        // We'll start by performing a series of validation checks.
        ntfn, err := n.newConfNtfn(txid, pkScript, numConfs, heightHint, opts)
        if err != nil {
                return nil, err
        }

        // Before proceeding to register the notification, we'll query our
        // height hint cache to determine whether a better one exists.
        //
        // TODO(conner): verify that all submitted height hints are identical.
        startHeight := ntfn.HeightHint
        hint, err := n.confirmHintCache.QueryConfirmHint(ntfn.ConfRequest)
        if err == nil {
                if hint > startHeight {
                        Log.Debugf("Using height hint %d retrieved from cache "+
                                "for %v instead of %d for conf subscription",
                                hint, ntfn.ConfRequest, startHeight)
                        startHeight = hint
                }
        } else if err != ErrConfirmHintNotFound {
                Log.Errorf("Unable to query confirm hint for %v: %v",
                        ntfn.ConfRequest, err)
        }

        Log.Infof("New confirmation subscription: conf_id=%d, %v, "+
                "num_confs=%v height_hint=%d", ntfn.ConfID, ntfn.ConfRequest,
                numConfs, startHeight)

        n.Lock()
        defer n.Unlock()

        confSet, ok := n.confNotifications[ntfn.ConfRequest]
        if !ok {
                // If this is the first registration for this request, construct
                // a confSet to coalesce all notifications for the same request.
                confSet = newConfNtfnSet()
                n.confNotifications[ntfn.ConfRequest] = confSet
        }
        confSet.ntfns[ntfn.ConfID] = ntfn

        switch confSet.rescanStatus {

        // A prior rescan has already completed and we are actively watching at
        // tip for this request.
        case rescanComplete:
                // If the confirmation details for this set of notifications has
                // already been found, we'll attempt to deliver them immediately
                // to this client.
                Log.Debugf("Attempting to dispatch confirmation for %v on "+
                        "registration since rescan has finished, conf_id=%v",
                        ntfn.ConfRequest, ntfn.ConfID)

                // The default notification we assigned above includes the
                // block along with the rest of the details. However not all
                // clients want the block, so we make a copy here w/o the block
                // if needed so we can give clients only what they ask for.
                confDetails := confSet.details
                if !ntfn.includeBlock && confDetails != nil {
                        confDetailsCopy := *confDetails
                        confDetailsCopy.Block = nil

                        confDetails = &confDetailsCopy
                }

                // Deliver the details to the whole conf set where this ntfn
                // lives in.
                for _, subscriber := range confSet.ntfns {
                        err := n.dispatchConfDetails(subscriber, confDetails)
                        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
        // updated as well.
        case rescanPending:
                Log.Debugf("Waiting for pending rescan to finish before "+
                        "notifying %v at tip", ntfn.ConfRequest)

                return &ConfRegistration{
                        Event:              ntfn.Event,
                        HistoricalDispatch: nil,
                        Height:             n.currentHeight,
                }, nil

        // If no rescan has been dispatched, attempt to do so now.
        case rescanNotStarted:
        }

        // If the provided or cached height hint indicates that the
        // transaction with the given txid/output script is to be confirmed at a
        // height greater than the notifier's current height, we'll refrain from
        // spawning a historical dispatch.
        if startHeight > n.currentHeight {
                Log.Debugf("Height hint is above current height, not "+
                        "dispatching historical confirmation rescan for %v",
                        ntfn.ConfRequest)

                // Set the rescan status to complete, which will allow the
                // notifier to start delivering messages for this set
                // immediately.
                confSet.rescanStatus = rescanComplete
                return &ConfRegistration{
                        Event:              ntfn.Event,
                        HistoricalDispatch: nil,
                        Height:             n.currentHeight,
                }, nil
        }

        Log.Debugf("Dispatching historical confirmation rescan for %v",
                ntfn.ConfRequest)

        // Construct the parameters for historical dispatch, scanning the range
        // of blocks between our best known height hint and the notifier's
        // current height. The notifier will begin also watching for
        // confirmations at tip starting with the next block.
        dispatch := &HistoricalConfDispatch{
                ConfRequest: ntfn.ConfRequest,
                StartHeight: startHeight,
                EndHeight:   n.currentHeight,
        }

        // Set this confSet's status to pending, ensuring subsequent
        // registrations don't also attempt a dispatch.
        confSet.rescanStatus = rescanPending

        return &ConfRegistration{
                Event:              ntfn.Event,
                HistoricalDispatch: dispatch,
                Height:             n.currentHeight,
        }, nil
}

// CancelConf cancels an existing request for a spend notification of an
// outpoint/output script. The request is identified by its spend ID.
func (n *TxNotifier) CancelConf(confRequest ConfRequest, confID uint64) {
        select {
        case <-n.quit:
                return
        default:
        }

        n.Lock()
        defer n.Unlock()

        confSet, ok := n.confNotifications[confRequest]
        if !ok {
                return
        }
        ntfn, ok := confSet.ntfns[confID]
        if !ok {
                return
        }

        Log.Debugf("Canceling confirmation notification: conf_id=%d, %v",
                confID, confRequest)

        // We'll close all the notification channels to let the client know
        // their cancel request has been fulfilled.
        close(ntfn.Event.Confirmed)
        close(ntfn.Event.Updates)
        close(ntfn.Event.NegativeConf)

        // Finally, we'll clean up any lingering references to this
        // notification.
        delete(confSet.ntfns, confID)

        // Remove the queued confirmation notification if the transaction has
        // already confirmed, but hasn't met its required number of
        // confirmations.
        if confSet.details != nil {
                confHeight := confSet.details.BlockHeight +
                        ntfn.NumConfirmations - 1
                delete(n.ntfnsByConfirmHeight[confHeight], ntfn)
        }
}

// UpdateConfDetails attempts to update the confirmation details for an active
// notification within the notifier. This should only be used in the case of a
// transaction/output script that has confirmed before the notifier's current
// height.
//
// NOTE: The notification should be registered first to ensure notifications are
// dispatched correctly.
func (n *TxNotifier) UpdateConfDetails(confRequest ConfRequest,
        details *TxConfirmation) error {

        select {
        case <-n.quit:
                return ErrTxNotifierExiting
        default:
        }

        // Ensure we hold the lock throughout handling the notification to
        // prevent the notifier from advancing its height underneath us.
        n.Lock()
        defer n.Unlock()

        // First, we'll determine whether we have an active confirmation
        // notification for the given txid/script.
        confSet, ok := n.confNotifications[confRequest]
        if !ok {
                return fmt.Errorf("confirmation notification for %v not found",
                        confRequest)
        }

        // If the confirmation details were already found at tip, all existing
        // notifications will have been dispatched or queued for dispatch. We
        // can exit early to avoid sending too many notifications on the
        // buffered channels.
        if confSet.details != nil {
                return nil
        }

        // The historical dispatch has been completed for this confSet. We'll
        // update the rescan status and cache any details that were found. If
        // the details are nil, that implies we did not find them and will
        // continue to watch for them at tip.
        confSet.rescanStatus = rescanComplete

        // The notifier has yet to reach the height at which the
        // transaction/output script was included in a block, so we should defer
        // until handling it then within ConnectTip.
        if details == nil {
                Log.Debugf("Confirmation details for %v not found during "+
                        "historical dispatch, waiting to dispatch at tip",
                        confRequest)

                // We'll commit the current height as the confirm hint to
                // prevent another potentially long rescan if we restart before
                // a new block comes in.
                err := n.confirmHintCache.CommitConfirmHint(
                        n.currentHeight, confRequest,
                )
                if err != nil {
                        // The error is not fatal as this is an optimistic
                        // optimization, so we'll avoid returning an error.
                        Log.Debugf("Unable to update confirm hint to %d for "+
                                "%v: %v", n.currentHeight, confRequest, err)
                }

                return nil
        }

        if details.BlockHeight > n.currentHeight {
                Log.Debugf("Confirmation details for %v found above current "+
                        "height, waiting to dispatch at tip", confRequest)

                return nil
        }

        Log.Debugf("Updating confirmation details for %v", confRequest)

        err := n.confirmHintCache.CommitConfirmHint(
                details.BlockHeight, confRequest,
        )
        if err != nil {
                // The error is not fatal, so we should not return an error to
                // the caller.
                Log.Errorf("Unable to update confirm hint to %d for %v: %v",
                        details.BlockHeight, confRequest, err)
        }

        // Cache the details found in the rescan and attempt to dispatch any
        // notifications that have not yet been delivered.
        confSet.details = details
        for _, ntfn := range confSet.ntfns {
                // The default notification we assigned above includes the
                // block along with the rest of the details. However not all
                // clients want the block, so we make a copy here w/o the block
                // if needed so we can give clients only what they ask for.
                confDetails := *details
                if !ntfn.includeBlock {
                        confDetails.Block = nil
                }

                err = n.dispatchConfDetails(ntfn, &confDetails)
                if err != nil {
                        return err
                }
        }

        return nil
}

// dispatchConfDetails attempts to cache and dispatch details to a particular
// client if the transaction/output script has sufficiently confirmed. If the
// provided details are nil, this method will be a no-op.
func (n *TxNotifier) dispatchConfDetails(
        ntfn *ConfNtfn, details *TxConfirmation) error {

        // If there are no conf details to dispatch or if the notification has
        // already been dispatched, then we can skip dispatching to this
        // client.
        if details == nil {
                Log.Debugf("Skipped dispatching nil conf details for request "+
                        "%v, conf_id=%v", ntfn.ConfRequest, ntfn.ConfID)

                return nil
        }

        if ntfn.dispatched {
                Log.Debugf("Skipped dispatched conf details for request %v "+
                        "conf_id=%v", ntfn.ConfRequest, ntfn.ConfID)

                return nil
        }

        // Now, we'll examine whether the transaction/output script of this
        // request has reached its required number of confirmations. If it has,
        // we'll dispatch a confirmation notification to the caller.
        confHeight := details.BlockHeight + ntfn.NumConfirmations - 1
        if confHeight <= n.currentHeight {
                Log.Debugf("Dispatching %v confirmation notification for "+
                        "conf_id=%v, %v", ntfn.NumConfirmations, ntfn.ConfID,
                        ntfn.ConfRequest)

                // We'll send a 0 value to the Updates channel,
                // indicating that the transaction/output script has already
                // been confirmed, and include the block height at which the
                // transaction was included.
                err := n.notifyNumConfsLeft(ntfn, TxUpdateInfo{
                        NumConfsLeft: 0,
                        BlockHeight:  details.BlockHeight,
                })
                if err != nil {
                        return err
                }

                select {
                case ntfn.Event.Confirmed <- details:
                        ntfn.dispatched = true
                case <-n.quit:
                        return ErrTxNotifierExiting
                }
        } else {
                Log.Debugf("Queueing %v confirmation notification for %v at "+
                        "tip", ntfn.NumConfirmations, ntfn.ConfRequest)

                // Otherwise, we'll keep track of the notification
                // request by the height at which we should dispatch the
                // confirmation notification.
                ntfnSet, exists := n.ntfnsByConfirmHeight[confHeight]
                if !exists {
                        ntfnSet = make(map[*ConfNtfn]struct{})
                        n.ntfnsByConfirmHeight[confHeight] = ntfnSet
                }
                ntfnSet[ntfn] = struct{}{}

                // We'll also send an update to the client of how many
                // confirmations are left for the transaction/output script to
                // be confirmed.
                numConfsLeft := confHeight - n.currentHeight
                err := n.notifyNumConfsLeft(ntfn, TxUpdateInfo{
                        NumConfsLeft: numConfsLeft,
                        BlockHeight:  details.BlockHeight,
                })
                if err != nil {
                        return err
                }
        }

        // As a final check, we'll also watch the transaction/output script if
        // it's still possible for it to get reorged out of the chain.
        reorgSafeHeight := details.BlockHeight + n.reorgSafetyLimit
        if reorgSafeHeight > n.currentHeight {
                txSet, exists := n.confsByInitialHeight[details.BlockHeight]
                if !exists {
                        txSet = make(map[ConfRequest]struct{})
                        n.confsByInitialHeight[details.BlockHeight] = txSet
                }
                txSet[ntfn.ConfRequest] = struct{}{}
        }

        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) {

        // 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
        }

        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.
//
// 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(outpoint *wire.OutPoint, pkScript []byte,
        heightHint uint32) (*SpendRegistration, error) {

        select {
        case <-n.quit:
                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
        hint, err := n.spendHintCache.QuerySpendHint(ntfn.SpendRequest)
        if err == nil {
                if hint > startHeight {
                        Log.Debugf("Using height hint %d retrieved from cache "+
                                "for %v instead of %d for spend subscription",
                                hint, ntfn.SpendRequest, startHeight)
                        startHeight = hint
                }
        } else if err != ErrSpendHintNotFound {
                Log.Errorf("Unable to query spend hint for %v: %v",
                        ntfn.SpendRequest, err)
        }

        n.Lock()
        defer n.Unlock()

        Log.Debugf("New spend subscription: spend_id=%d, %v, height_hint=%d",
                ntfn.SpendID, ntfn.SpendRequest, startHeight)

        // Keep track of the notification request so that we can properly
        // dispatch a spend notification later on.
        spendSet, ok := n.spendNotifications[ntfn.SpendRequest]
        if !ok {
                // If this is the first registration for the request, we'll
                // construct a spendNtfnSet to coalesce all notifications.
                spendSet = newSpendNtfnSet()
                n.spendNotifications[ntfn.SpendRequest] = spendSet
        }
        spendSet.ntfns[ntfn.SpendID] = ntfn

        // We'll now let the caller know whether a historical rescan is needed
        // depending on the current rescan status.
        switch spendSet.rescanStatus {

        // If the spending details for this request have already been determined
        // and cached, then we can use them to immediately dispatch the spend
        // notification to the client.
        case rescanComplete:
                Log.Debugf("Attempting to dispatch spend for %v on "+
                        "registration since rescan has finished",
                        ntfn.SpendRequest)

                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.
        case rescanPending:
                Log.Debugf("Waiting for pending rescan to finish before "+
                        "notifying %v at tip", ntfn.SpendRequest)

                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
        // been spent.
        case rescanNotStarted:
        }

        // However, if the spend hint, either provided by the caller or
        // retrieved from the cache, is found to be at a later height than the
        // TxNotifier is aware of, then we'll refrain from dispatching a
        // historical rescan and wait for the spend to come in at tip.
        if startHeight > n.currentHeight {
                Log.Debugf("Spend hint of %d for %v is above current height %d",
                        startHeight, ntfn.SpendRequest, n.currentHeight)

                // We'll also set the rescan status as complete to ensure that
                // spend hints for this request get updated upon
                // connected/disconnected blocks.
                spendSet.rescanStatus = rescanComplete
                return &SpendRegistration{
                        Event:              ntfn.Event,
                        HistoricalDispatch: nil,
                        Height:             n.currentHeight,
                }, nil
        }

        // We'll set the rescan status to pending to ensure subsequent
        // notifications don't also attempt a historical dispatch.
        spendSet.rescanStatus = rescanPending

        Log.Debugf("Dispatching historical spend rescan for %v, start=%d, "+
                "end=%d", ntfn.SpendRequest, startHeight, n.currentHeight)

        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
// outpoint/output script. The request is identified by its spend ID.
func (n *TxNotifier) CancelSpend(spendRequest SpendRequest, spendID uint64) {
        select {
        case <-n.quit:
                return
        default:
        }

        n.Lock()
        defer n.Unlock()

        spendSet, ok := n.spendNotifications[spendRequest]
        if !ok {
                return
        }
        ntfn, ok := spendSet.ntfns[spendID]
        if !ok {
                return
        }

        Log.Debugf("Canceling spend notification: spend_id=%d, %v", spendID,
                spendRequest)

        // We'll close all the notification channels to let the client know
        // their cancel request has been fulfilled.
        close(ntfn.Event.Spend)
        close(ntfn.Event.Reorg)
        close(ntfn.Event.Done)
        delete(spendSet.ntfns, spendID)
}

// ProcessRelevantSpendTx processes a transaction provided externally. This will
// check whether the transaction is relevant to the notifier if it spends any
// outpoints/output scripts for which we currently have registered notifications
// for. If it is relevant, spend notifications will be dispatched to the caller.
func (n *TxNotifier) ProcessRelevantSpendTx(tx *btcutil.Tx,
        blockHeight uint32) error {

        select {
        case <-n.quit:
                return ErrTxNotifierExiting
        default:
        }

        // Ensure we hold the lock throughout handling the notification to
        // prevent the notifier from advancing its height underneath us.
        n.Lock()
        defer n.Unlock()

        // We'll use a channel to coalesce all the spend requests that this
        // transaction fulfills.
        type spend struct {
                request *SpendRequest
                details *SpendDetail
        }

        // We'll set up the onSpend filter callback to gather all the fulfilled
        // spends requests within this transaction.
        var spends []spend
        onSpend := func(request SpendRequest, details *SpendDetail) {
                spends = append(spends, spend{&request, details})
        }
        n.filterTx(nil, tx, blockHeight, nil, onSpend)

        // After the transaction has been filtered, we can finally dispatch
        // notifications for each request.
        for _, spend := range spends {
                err := n.updateSpendDetails(*spend.request, spend.details)
                if err != nil {
                        return err
                }
        }

        return nil
}

// UpdateSpendDetails attempts to update the spend details for all active spend
// notification requests for an outpoint/output script. This method should be
// used once a historical scan of the chain has finished. If the historical scan
// did not find a spending transaction for it, the spend details may be nil.
//
// NOTE: A notification request for the outpoint/output script must be
// registered first to ensure notifications are delivered.
func (n *TxNotifier) UpdateSpendDetails(spendRequest SpendRequest,
        details *SpendDetail) error {

        select {
        case <-n.quit:
                return ErrTxNotifierExiting
        default:
        }

        // Ensure we hold the lock throughout handling the notification to
        // prevent the notifier from advancing its height underneath us.
        n.Lock()
        defer n.Unlock()

        return n.updateSpendDetails(spendRequest, details)
}

// updateSpendDetails attempts to update the spend details for all active spend
// notification requests for an outpoint/output script. This method should be
// used once a historical scan of the chain has finished. If the historical scan
// did not find a spending transaction for it, the spend details may be nil.
//
// NOTE: This method must be called with the TxNotifier's lock held.
func (n *TxNotifier) updateSpendDetails(spendRequest SpendRequest,
        details *SpendDetail) error {

        // Mark the ongoing historical rescan for this request as finished. This
        // will allow us to update the spend hints for it at tip.
        spendSet, ok := n.spendNotifications[spendRequest]
        if !ok {
                return fmt.Errorf("spend notification for %v not found",
                        spendRequest)
        }

        // If the spend details have already been found either at tip, then the
        // notifications should have already been dispatched, so we can exit
        // early to prevent sending duplicate notifications.
        if spendSet.details != nil {
                return nil
        }

        // Since the historical rescan has completed for this request, we'll
        // mark its rescan status as complete in order to ensure that the
        // TxNotifier can properly update its spend hints upon
        // connected/disconnected blocks.
        spendSet.rescanStatus = rescanComplete

        // If the historical rescan was not able to find a spending transaction
        // for this request, then we can track the spend at tip.
        if details == nil {
                // We'll commit the current height as the spend hint to prevent
                // another potentially long rescan if we restart before a new
                // block comes in.
                err := n.spendHintCache.CommitSpendHint(
                        n.currentHeight, spendRequest,
                )
                if err != nil {
                        // The error is not fatal as this is an optimistic
                        // optimization, so we'll avoid returning an error.
                        Log.Debugf("Unable to update spend hint to %d for %v: %v",
                                n.currentHeight, spendRequest, err)
                }

                Log.Debugf("Updated spend hint to height=%v for unconfirmed "+
                        "spend request %v", n.currentHeight, spendRequest)
                return nil
        }

        // Return an error if the witness data is not present in the spending
        // transaction.
        //
        // NOTE: if the witness stack is empty, we will do a critical log which
        // shuts down the node.
        if !details.HasSpenderWitness() {
                Log.Criticalf("Found spending tx for outpoint=%v, but the "+
                        "transaction %v does not have witness",
                        spendRequest.OutPoint, details.SpendingTx.TxHash())

                return ErrEmptyWitnessStack
        }

        // If the historical rescan found the spending transaction for this
        // request, but it's at a later height than the notifier (this can
        // happen due to latency with the backend during a reorg), then we'll
        // defer handling the notification until the notifier has caught up to
        // such height.
        if uint32(details.SpendingHeight) > n.currentHeight {
                return nil
        }

        // Now that we've determined the request has been spent, we'll commit
        // its spending height as its hint in the cache and dispatch
        // notifications to all of its respective clients.
        err := n.spendHintCache.CommitSpendHint(
                uint32(details.SpendingHeight), spendRequest,
        )
        if err != nil {
                // The error is not fatal as this is an optimistic optimization,
                // so we'll avoid returning an error.
                Log.Debugf("Unable to update spend hint to %d for %v: %v",
                        details.SpendingHeight, spendRequest, err)
        }

        Log.Debugf("Updated spend hint to height=%v for confirmed spend "+
                "request %v", details.SpendingHeight, spendRequest)

        spendSet.details = details
        for _, ntfn := range spendSet.ntfns {
                err := n.dispatchSpendDetails(ntfn, spendSet.details)
                if err != nil {
                        return err
                }
        }

        return nil
}

// dispatchSpendDetails dispatches a spend notification to the client.
//
// NOTE: This must be called with the TxNotifier's lock held.
func (n *TxNotifier) dispatchSpendDetails(ntfn *SpendNtfn, details *SpendDetail) error {
        // If there are no spend details to dispatch or if the notification has
        // already been dispatched, then we can skip dispatching to this client.
        if details == nil || ntfn.dispatched {
                Log.Debugf("Skipping dispatch of spend details(%v) for "+
                        "request %v, dispatched=%v", details, ntfn.SpendRequest,
                        ntfn.dispatched)
                return nil
        }

        Log.Debugf("Dispatching confirmed spend notification for %v at "+
                "current height=%d: %v", ntfn.SpendRequest, n.currentHeight,
                details)

        select {
        case ntfn.Event.Spend <- details:
                ntfn.dispatched = true
        case <-n.quit:
                return ErrTxNotifierExiting
        }

        spendHeight := uint32(details.SpendingHeight)

        // We also add to spendsByHeight to notify on chain reorgs.
        reorgSafeHeight := spendHeight + n.reorgSafetyLimit
        if reorgSafeHeight > n.currentHeight {
                txSet, exists := n.spendsByHeight[spendHeight]
                if !exists {
                        txSet = make(map[SpendRequest]struct{})
                        n.spendsByHeight[spendHeight] = txSet
                }
                txSet[ntfn.SpendRequest] = struct{}{}
        }

        return nil
}

// ConnectTip handles a new block extending the current chain. It will go
// through every transaction and determine if it is relevant to any of its
// clients. A transaction can be relevant in either of the following two ways:
//
//  1. One of the inputs in the transaction spends an outpoint/output script
//     for which we currently have an active spend registration for.
//
//  2. The transaction has a txid or output script for which we currently have
//     an active confirmation registration for.
//
// In the event that the transaction is relevant, a confirmation/spend
// notification will be queued for dispatch to the relevant clients.
// Confirmation notifications will only be dispatched for transactions/output
// scripts that have met the required number of confirmations required by the
// client.
//
// NOTE: In order to actually dispatch the relevant transaction notifications to
// clients, NotifyHeight must be called with the same block height in order to
// maintain correctness.
func (n *TxNotifier) ConnectTip(block *btcutil.Block,
        blockHeight uint32) error {

        select {
        case <-n.quit:
                return ErrTxNotifierExiting
        default:
        }

        n.Lock()
        defer n.Unlock()

        if blockHeight != n.currentHeight+1 {
                return fmt.Errorf("received blocks out of order: "+
                        "current height=%d, new height=%d",
                        n.currentHeight, blockHeight)
        }
        n.currentHeight++
        n.reorgDepth = 0

        // First, we'll iterate over all the transactions found in this block to
        // determine if it includes any relevant transactions to the TxNotifier.
        if block != nil {
                Log.Debugf("Filtering %d txns for %d spend requests at "+
                        "height %d", len(block.Transactions()),
                        len(n.spendNotifications), blockHeight)

                for _, tx := range block.Transactions() {
                        n.filterTx(
                                block, tx, blockHeight,
                                n.handleConfDetailsAtTip,
                                n.handleSpendDetailsAtTip,
                        )
                }
        }

        // Now that we've determined which requests were confirmed and spent
        // within the new block, we can update their entries in their respective
        // caches, along with all of our unconfirmed and unspent requests.
        n.updateHints(blockHeight)

        // Finally, we'll clear the entries from our set of notifications for
        // requests that are no longer under the risk of being reorged out of
        // the chain.
        if blockHeight >= n.reorgSafetyLimit {
                matureBlockHeight := blockHeight - n.reorgSafetyLimit
                for confRequest := range n.confsByInitialHeight[matureBlockHeight] {
                        confSet := n.confNotifications[confRequest]
                        for _, ntfn := range confSet.ntfns {
                                select {
                                case ntfn.Event.Done <- struct{}{}:
                                case <-n.quit:
                                        return ErrTxNotifierExiting
                                }
                        }

                        delete(n.confNotifications, confRequest)
                }
                delete(n.confsByInitialHeight, matureBlockHeight)

                for spendRequest := range n.spendsByHeight[matureBlockHeight] {
                        spendSet := n.spendNotifications[spendRequest]
                        for _, ntfn := range spendSet.ntfns {
                                select {
                                case ntfn.Event.Done <- struct{}{}:
                                case <-n.quit:
                                        return ErrTxNotifierExiting
                                }
                        }

                        Log.Debugf("Deleting mature spend request %v at "+
                                "height=%d", spendRequest, blockHeight)
                        delete(n.spendNotifications, spendRequest)
                }
                delete(n.spendsByHeight, matureBlockHeight)
        }

        return nil
}

// filterTx determines whether the transaction spends or confirms any
// outstanding pending requests. The onConf and onSpend callbacks can be used to
// retrieve all the requests fulfilled by this transaction as they occur.
func (n *TxNotifier) filterTx(block *btcutil.Block, tx *btcutil.Tx,
        blockHeight uint32, onConf func(ConfRequest, *TxConfirmation),
        onSpend func(SpendRequest, *SpendDetail)) {

        // In order to determine if this transaction is relevant to the
        // notifier, we'll check its inputs for any outstanding spend
        // requests.
        txHash := tx.Hash()
        if onSpend != nil {
                // notifyDetails is a helper closure that will construct the
                // spend details of a request and hand them off to the onSpend
                // callback.
                notifyDetails := func(spendRequest SpendRequest,
                        prevOut wire.OutPoint, inputIdx uint32) {

                        Log.Debugf("Found spend of %v: spend_tx=%v, "+
                                "block_height=%d", spendRequest, txHash,
                                blockHeight)

                        onSpend(spendRequest, &SpendDetail{
                                SpentOutPoint:     &prevOut,
                                SpenderTxHash:     txHash,
                                SpendingTx:        tx.MsgTx(),
                                SpenderInputIndex: inputIdx,
                                SpendingHeight:    int32(blockHeight),
                        })
                }

                for i, txIn := range tx.MsgTx().TxIn {
                        // We'll re-derive the script of the output being spent
                        // to determine if the inputs spends any registered
                        // requests.
                        prevOut := txIn.PreviousOutPoint
                        pkScript, err := txscript.ComputePkScript(
                                txIn.SignatureScript, txIn.Witness,
                        )
                        if err != nil {
                                continue
                        }
                        spendRequest := SpendRequest{
                                OutPoint: prevOut,
                                PkScript: pkScript,
                        }

                        // If we have any, we'll record their spend height so
                        // that notifications get dispatched to the respective
                        // clients.
                        if _, ok := n.spendNotifications[spendRequest]; ok {
                                notifyDetails(spendRequest, prevOut, uint32(i))
                        }

                        // Now try with an empty taproot key pkScript, since we
                        // cannot derive the spent pkScript directly from the
                        // witness. But we have the outpoint, which should be
                        // enough.
                        spendRequest.PkScript = ZeroTaprootPkScript
                        if _, ok := n.spendNotifications[spendRequest]; ok {
                                notifyDetails(spendRequest, prevOut, uint32(i))
                        }

                        // Restore the pkScript but try with a zero outpoint
                        // instead (won't be possible for Taproot).
                        spendRequest.PkScript = pkScript
                        spendRequest.OutPoint = ZeroOutPoint
                        if _, ok := n.spendNotifications[spendRequest]; ok {
                                notifyDetails(spendRequest, prevOut, uint32(i))
                        }
                }
        }

        // We'll also check its outputs to determine if there are any
        // outstanding confirmation requests.
        if onConf != nil {
                // notifyDetails is a helper closure that will construct the
                // confirmation details of a request and hand them off to the
                // onConf callback.
                notifyDetails := func(confRequest ConfRequest) {
                        Log.Debugf("Found initial confirmation of %v: "+
                                "height=%d, hash=%v", confRequest,
                                blockHeight, block.Hash())

                        details := &TxConfirmation{
                                Tx:          tx.MsgTx(),
                                BlockHash:   block.Hash(),
                                BlockHeight: blockHeight,
                                TxIndex:     uint32(tx.Index()),
                                Block:       block.MsgBlock(),
                        }

                        onConf(confRequest, details)
                }

                for _, txOut := range tx.MsgTx().TxOut {
                        // We'll parse the script of the output to determine if
                        // we have any registered requests for it or the
                        // transaction itself.
                        pkScript, err := txscript.ParsePkScript(txOut.PkScript)
                        if err != nil {
                                continue
                        }
                        confRequest := ConfRequest{
                                TxID:     *txHash,
                                PkScript: pkScript,
                        }

                        // If we have any, we'll record their confirmed height
                        // so that notifications get dispatched when they
                        // reaches the clients' desired number of confirmations.
                        if _, ok := n.confNotifications[confRequest]; ok {
                                notifyDetails(confRequest)
                        }
                        confRequest.TxID = ZeroHash
                        if _, ok := n.confNotifications[confRequest]; ok {
                                notifyDetails(confRequest)
                        }
                }
        }
}

// handleConfDetailsAtTip tracks the confirmation height of the txid/output
// script in order to properly dispatch a confirmation notification after
// meeting each request's desired number of confirmations for all current and
// future registered clients.
func (n *TxNotifier) handleConfDetailsAtTip(confRequest ConfRequest,
        details *TxConfirmation) {

        // TODO(wilmer): cancel pending historical rescans if any?
        confSet := n.confNotifications[confRequest]

        // If we already have details for this request, we don't want to add it
        // again since we have already dispatched notifications for it.
        if confSet.details != nil {
                Log.Warnf("Ignoring address reuse for %s at height %d.",
                        confRequest, details.BlockHeight)
                return
        }

        confSet.rescanStatus = rescanComplete
        confSet.details = details

        for _, ntfn := range confSet.ntfns {
                // In the event that this notification was aware that the
                // transaction/output script was reorged out of the chain, we'll
                // consume the reorg notification if it hasn't been done yet
                // already.
                select {
                case <-ntfn.Event.NegativeConf:
                default:
                }

                // We'll note this client's required number of confirmations so
                // that we can notify them when expected.
                confHeight := details.BlockHeight + ntfn.NumConfirmations - 1
                ntfnSet, exists := n.ntfnsByConfirmHeight[confHeight]
                if !exists {
                        ntfnSet = make(map[*ConfNtfn]struct{})
                        n.ntfnsByConfirmHeight[confHeight] = ntfnSet
                }
                ntfnSet[ntfn] = struct{}{}
        }

        // We'll also note the initial confirmation height in order to correctly
        // handle dispatching notifications when the transaction/output script
        // gets reorged out of the chain.
        txSet, exists := n.confsByInitialHeight[details.BlockHeight]
        if !exists {
                txSet = make(map[ConfRequest]struct{})
                n.confsByInitialHeight[details.BlockHeight] = txSet
        }
        txSet[confRequest] = struct{}{}
}

// handleSpendDetailsAtTip tracks the spend height of the outpoint/output script
// in order to properly dispatch a spend notification for all current and future
// registered clients.
func (n *TxNotifier) handleSpendDetailsAtTip(spendRequest SpendRequest,
        details *SpendDetail) {

        // TODO(wilmer): cancel pending historical rescans if any?
        spendSet := n.spendNotifications[spendRequest]
        spendSet.rescanStatus = rescanComplete
        spendSet.details = details

        for _, ntfn := range spendSet.ntfns {
                // In the event that this notification was aware that the
                // spending transaction of its outpoint/output script was
                // reorged out of the chain, we'll consume the reorg
                // notification if it hasn't been done yet already.
                select {
                case <-ntfn.Event.Reorg:
                default:
                }
        }

        // We'll note the spending height of the request in order to correctly
        // handle dispatching notifications when the spending transactions gets
        // reorged out of the chain.
        spendHeight := uint32(details.SpendingHeight)
        opSet, exists := n.spendsByHeight[spendHeight]
        if !exists {
                opSet = make(map[SpendRequest]struct{})
                n.spendsByHeight[spendHeight] = opSet
        }
        opSet[spendRequest] = struct{}{}

        Log.Debugf("Spend request %v spent at tip=%d", spendRequest,
                spendHeight)
}

// NotifyHeight dispatches confirmation and spend notifications to the clients
// who registered for a notification which has been fulfilled at the passed
// height.
func (n *TxNotifier) NotifyHeight(height uint32) error {
        n.Lock()
        defer n.Unlock()

        // First, we'll dispatch an update to all of the notification clients
        // for our watched requests with the number of confirmations left at
        // this new height.
        for _, confRequests := range n.confsByInitialHeight {
                for confRequest := range confRequests {
                        confSet := n.confNotifications[confRequest]
                        for _, ntfn := range confSet.ntfns {
                                // blockHeight is the height of the block which
                                // contains the transaction.
                                blockHeight := confSet.details.BlockHeight
                                txConfHeight := blockHeight +
                                        ntfn.NumConfirmations - 1
                                numConfsLeft := txConfHeight - height

                                // Since we don't clear notifications until
                                // transactions/output scripts are no longer
                                // under the risk of being reorganized out of
                                // the chain, we'll skip sending updates for
                                // those that have already been confirmed.
                                if int32(numConfsLeft) < 0 {
                                        continue
                                }

                                err := n.notifyNumConfsLeft(ntfn, TxUpdateInfo{
                                        NumConfsLeft: numConfsLeft,
                                        BlockHeight:  blockHeight,
                                })
                                if err != nil {
                                        return err
                                }
                        }
                }
        }

        // Then, we'll dispatch notifications for all the requests that have
        // become confirmed at this new block height.
        for ntfn := range n.ntfnsByConfirmHeight[height] {
                confSet := n.confNotifications[ntfn.ConfRequest]

                // The default notification we assigned above includes the
                // block along with the rest of the details. However not all
                // clients want the block, so we make a copy here w/o the block
                // if needed so we can give clients only what they ask for.
                confDetails := *confSet.details
                if !ntfn.includeBlock {
                        confDetails.Block = nil
                }

                // If the `confDetails` has already been sent before, we'll
                // skip it and continue processing the next one.
                if ntfn.dispatched {
                        Log.Debugf("Skipped dispatched conf details for "+
                                "request %v conf_id=%v", ntfn.ConfRequest,
                                ntfn.ConfID)

                        continue
                }

                Log.Debugf("Dispatching %v confirmation notification for "+
                        "conf_id=%v, %v", ntfn.NumConfirmations, ntfn.ConfID,
                        ntfn.ConfRequest)

                select {
                case ntfn.Event.Confirmed <- &confDetails:
                        ntfn.dispatched = true
                case <-n.quit:
                        return ErrTxNotifierExiting
                }
        }
        delete(n.ntfnsByConfirmHeight, height)

        // Finally, we'll dispatch spend notifications for all the requests that
        // were spent at this new block height.
        for spendRequest := range n.spendsByHeight[height] {
                spendSet := n.spendNotifications[spendRequest]
                for _, ntfn := range spendSet.ntfns {
                        err := n.dispatchSpendDetails(ntfn, spendSet.details)
                        if err != nil {
                                return err
                        }
                }
        }

        return nil
}

// DisconnectTip handles the tip of the current chain being disconnected during
// a chain reorganization. If any watched requests were included in this block,
// internal structures are updated to ensure confirmation/spend notifications
// are consumed (if not already), and reorg notifications are dispatched
// instead. Confirmation/spend notifications will be dispatched again upon block
// inclusion.
func (n *TxNotifier) DisconnectTip(blockHeight uint32) error {
        select {
        case <-n.quit:
                return ErrTxNotifierExiting
        default:
        }

        n.Lock()
        defer n.Unlock()

        if blockHeight != n.currentHeight {
                return fmt.Errorf("received blocks out of order: "+
                        "current height=%d, disconnected height=%d",
                        n.currentHeight, blockHeight)
        }
        n.currentHeight--
        n.reorgDepth++

        // With the block disconnected, we'll update the confirm and spend hints
        // for our notification requests to reflect the new height, except for
        // those that have confirmed/spent at previous heights.
        n.updateHints(blockHeight)

        // We'll go through all of our watched confirmation requests and attempt
        // to drain their notification channels to ensure sending notifications
        // to the clients is always non-blocking.
        for initialHeight, txHashes := range n.confsByInitialHeight {
                for txHash := range txHashes {
                        // If the transaction/output script has been reorged out
                        // of the chain, we'll make sure to remove the cached
                        // confirmation details to prevent notifying clients
                        // with old information.
                        confSet := n.confNotifications[txHash]
                        if initialHeight == blockHeight {
                                confSet.details = nil
                        }

                        for _, ntfn := range confSet.ntfns {
                                // First, we'll attempt to drain an update
                                // from each notification to ensure sends to the
                                // Updates channel are always non-blocking.
                                select {
                                case <-ntfn.Event.Updates:
                                case <-n.quit:
                                        return ErrTxNotifierExiting
                                default:
                                }

                                // We also reset the num of confs update.
                                ntfn.numConfsLeft = ntfn.NumConfirmations

                                // Then, we'll check if the current
                                // transaction/output script was included in the
                                // block currently being disconnected. If it
                                // was, we'll need to dispatch a reorg
                                // notification to the client.
                                if initialHeight == blockHeight {
                                        err := n.dispatchConfReorg(
                                                ntfn, blockHeight,
                                        )
                                        if err != nil {
                                                return err
                                        }
                                }
                        }
                }
        }

        // We'll also go through our watched spend requests and attempt to drain
        // their dispatched notifications to ensure dispatching notifications to
        // clients later on is always non-blocking. We're only interested in
        // requests whose spending transaction was included at the height being
        // disconnected.
        for op := range n.spendsByHeight[blockHeight] {
                // Since the spending transaction is being reorged out of the
                // chain, we'll need to clear out the spending details of the
                // request.
                spendSet := n.spendNotifications[op]
                spendSet.details = nil

                // For all requests which have had a spend notification
                // dispatched, we'll attempt to drain it and send a reorg
                // notification instead.
                for _, ntfn := range spendSet.ntfns {
                        if err := n.dispatchSpendReorg(ntfn); err != nil {
                                return err
                        }
                }
        }

        // Finally, we can remove the requests that were confirmed and/or spent
        // at the height being disconnected. We'll still continue to track them
        // until they have been confirmed/spent and are no longer under the risk
        // of being reorged out of the chain again.
        delete(n.confsByInitialHeight, blockHeight)
        delete(n.spendsByHeight, blockHeight)

        return nil
}

// updateHints attempts to update the confirm and spend hints for all relevant
// requests respectively. The height parameter is used to determine which
// requests we should update based on whether a new block is being
// connected/disconnected.
//
// NOTE: This must be called with the TxNotifier's lock held and after its
// height has already been reflected by a block being connected/disconnected.
func (n *TxNotifier) updateHints(height uint32) {
        // TODO(wilmer): update under one database transaction.
        //
        // To update the height hint for all the required confirmation requests
        // under one database transaction, we'll gather the set of unconfirmed
        // requests along with the ones that confirmed at the height being
        // connected/disconnected.
        confRequests := n.unconfirmedRequests()
        for confRequest := range n.confsByInitialHeight[height] {
                confRequests = append(confRequests, confRequest)
        }
        err := n.confirmHintCache.CommitConfirmHint(
                n.currentHeight, confRequests...,
        )
        if err != nil {
                // The error is not fatal as this is an optimistic optimization,
                // so we'll avoid returning an error.
                Log.Debugf("Unable to update confirm hints to %d for "+
                        "%v: %v", n.currentHeight, confRequests, err)
        }

        // Similarly, to update the height hint for all the required spend
        // requests under one database transaction, we'll gather the set of
        // unspent requests along with the ones that were spent at the height
        // being connected/disconnected.
        spendRequests := n.unspentRequests()
        for spendRequest := range n.spendsByHeight[height] {
                spendRequests = append(spendRequests, spendRequest)
        }
        err = n.spendHintCache.CommitSpendHint(n.currentHeight, spendRequests...)
        if err != nil {
                // The error is not fatal as this is an optimistic optimization,
                // so we'll avoid returning an error.
                Log.Debugf("Unable to update spend hints to %d for "+
                        "%v: %v", n.currentHeight, spendRequests, err)
        }
}

// unconfirmedRequests returns the set of confirmation requests that are
// still seen as unconfirmed by the TxNotifier.
//
// NOTE: This method must be called with the TxNotifier's lock held.
func (n *TxNotifier) unconfirmedRequests() []ConfRequest {
        var unconfirmed []ConfRequest
        for confRequest, confNtfnSet := range n.confNotifications {
                // If the notification is already aware of its confirmation
                // details, or it's in the process of learning them, we'll skip
                // it as we can't yet determine if it's confirmed or not.
                if confNtfnSet.rescanStatus != rescanComplete ||
                        confNtfnSet.details != nil {
                        continue
                }

                unconfirmed = append(unconfirmed, confRequest)
        }

        return unconfirmed
}

// unspentRequests returns the set of spend requests that are still seen as
// unspent by the TxNotifier.
//
// NOTE: This method must be called with the TxNotifier's lock held.
func (n *TxNotifier) unspentRequests() []SpendRequest {
        var unspent []SpendRequest
        for spendRequest, spendNtfnSet := range n.spendNotifications {
                // If the notification is already aware of its spend details, or
                // it's in the process of learning them, we'll skip it as we
                // can't yet determine if it's unspent or not.
                if spendNtfnSet.rescanStatus != rescanComplete ||
                        spendNtfnSet.details != nil {
                        continue
                }

                unspent = append(unspent, spendRequest)
        }

        return unspent
}

// dispatchConfReorg dispatches a reorg notification to the client if the
// confirmation notification was already delivered.
//
// NOTE: This must be called with the TxNotifier's lock held.
func (n *TxNotifier) dispatchConfReorg(ntfn *ConfNtfn,
        heightDisconnected uint32) error {

        // If the request's confirmation notification has yet to be dispatched,
        // we'll need to clear its entry within the ntfnsByConfirmHeight index
        // to prevent from notifying the client once the notifier reaches the
        // confirmation height.
        if !ntfn.dispatched {
                confHeight := heightDisconnected + ntfn.NumConfirmations - 1
                ntfnSet, exists := n.ntfnsByConfirmHeight[confHeight]

                // We also signal the reorg to the notifier in case the
                // subscriber is also interested in the reorgs before the
                // transaction received its required confirmation.
                //
                // Because as soon as a new block is connected which has the
                // transaction included again we preemptively read the buffered
                // channel.
                select {
                case ntfn.Event.NegativeConf <- int32(n.reorgDepth):
                case <-n.quit:
                        return ErrTxNotifierExiting
                }

                if exists {
                        delete(ntfnSet, ntfn)
                }
                return nil
        }

        // Otherwise, the entry within the ntfnsByConfirmHeight has already been
        // deleted, so we'll attempt to drain the confirmation notification to
        // ensure sends to the Confirmed channel are always non-blocking.
        select {
        case <-ntfn.Event.Confirmed:
        case <-n.quit:
                return ErrTxNotifierExiting
        default:
        }

        ntfn.dispatched = false

        // Send a negative confirmation notification to the client indicating
        // how many blocks have been disconnected successively.
        select {
        case ntfn.Event.NegativeConf <- int32(n.reorgDepth):
        case <-n.quit:
                return ErrTxNotifierExiting
        }

        return nil
}

// dispatchSpendReorg dispatches a reorg notification to the client if a spend
// notiification was already delivered.
//
// NOTE: This must be called with the TxNotifier's lock held.
func (n *TxNotifier) dispatchSpendReorg(ntfn *SpendNtfn) error {
        if !ntfn.dispatched {
                return nil
        }

        // Attempt to drain the spend notification to ensure sends to the Spend
        // channel are always non-blocking.
        select {
        case <-ntfn.Event.Spend:
        default:
        }

        // Send a reorg notification to the client in order for them to
        // correctly handle reorgs.
        select {
        case ntfn.Event.Reorg <- struct{}{}:
        case <-n.quit:
                return ErrTxNotifierExiting
        }

        ntfn.dispatched = false

        return nil
}

// TearDown is to be called when the owner of the TxNotifier is exiting. This
// 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()

        for _, confSet := range n.confNotifications {
                for confID, ntfn := range confSet.ntfns {
                        close(ntfn.Event.Confirmed)
                        close(ntfn.Event.Updates)
                        close(ntfn.Event.NegativeConf)
                        close(ntfn.Event.Done)
                        delete(confSet.ntfns, confID)
                }
        }

        for _, spendSet := range n.spendNotifications {
                for spendID, ntfn := range spendSet.ntfns {
                        close(ntfn.Event.Spend)
                        close(ntfn.Event.Reorg)
                        close(ntfn.Event.Done)
                        delete(spendSet.ntfns, spendID)
                }
        }
}

// notifyNumConfsLeft sends the number of confirmations left to the
// notification subscriber through the Event.Updates channel, along with the
// block height in which the transaction was included.
//
// NOTE: must be used with the TxNotifier's lock held.
func (n *TxNotifier) notifyNumConfsLeft(ntfn *ConfNtfn,
        info TxUpdateInfo) error {

        // If the number left is no less than the recorded value, we can skip
        // sending it as it means this same value has already been sent before.
        if info.NumConfsLeft >= ntfn.numConfsLeft {
                Log.Debugf("Skipped dispatched update (numConfsLeft=%v) for "+
                        "request %v conf_id=%v", info.NumConfsLeft,
                        ntfn.ConfRequest, ntfn.ConfID)

                return nil
        }

        // Update the number of confirmations left to the notification.
        ntfn.numConfsLeft = info.NumConfsLeft

        select {
        case ntfn.Event.Updates <- info:
        case <-n.quit:
                return ErrTxNotifierExiting
        }

        return nil
}

lightningnetwork / lnd / 17685638204

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous