1948 lines
63 KiB
Go
1948 lines
63 KiB
Go
package chainntnfs
|
|
|
|
import (
|
|
"bytes"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"sync/atomic"
|
|
|
|
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
|
"github.com/btcsuite/btcd/txscript"
|
|
"github.com/btcsuite/btcd/wire"
|
|
"github.com/btcsuite/btcutil"
|
|
"github.com/lightningnetwork/lnd/channeldb"
|
|
)
|
|
|
|
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
|
|
)
|
|
|
|
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)
|
|
)
|
|
|
|
// 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)
|
|
}
|
|
|
|
// ConfHintKey returns the key that will be used to index the confirmation
|
|
// request's hint within the height hint cache.
|
|
func (r ConfRequest) ConfHintKey() ([]byte, error) {
|
|
if r.TxID == ZeroHash {
|
|
return r.PkScript.Script(), nil
|
|
}
|
|
|
|
var txid bytes.Buffer
|
|
if err := channeldb.WriteElement(&txid, r.TxID); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return txid.Bytes(), nil
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// HistoricalConfDispatch parameterizes 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 being 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
|
|
|
|
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)
|
|
}
|
|
|
|
// SpendHintKey returns the key that will be used to index the spend request's
|
|
// hint within the height hint cache.
|
|
func (r SpendRequest) SpendHintKey() ([]byte, error) {
|
|
if r.OutPoint == ZeroOutPoint {
|
|
return r.PkScript.Script(), nil
|
|
}
|
|
|
|
var outpoint bytes.Buffer
|
|
err := channeldb.WriteElement(&outpoint, r.OutPoint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return outpoint.Bytes(), nil
|
|
}
|
|
|
|
// 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 disptached
|
|
// to the client.
|
|
dispatched bool
|
|
}
|
|
|
|
// HistoricalSpendDispatch parameterizes 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 recan 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) (*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,
|
|
}, 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) (*ConfRegistration, error) {
|
|
|
|
select {
|
|
case <-n.quit:
|
|
return nil, ErrTxNotifierExiting
|
|
default:
|
|
}
|
|
|
|
// We'll start by performing a series of validation checks.
|
|
ntfn, err := n.newConfNtfn(txid, pkScript, numConfs, heightHint)
|
|
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", 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",
|
|
ntfn.ConfRequest)
|
|
|
|
err := n.dispatchConfDetails(ntfn, confSet.details)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &ConfRegistration{
|
|
Event: ntfn.Event,
|
|
HistoricalDispatch: nil,
|
|
Height: n.currentHeight,
|
|
}, nil
|
|
|
|
// A rescan is already in progress, return here to prevent dispatching
|
|
// another. When the rescan returns, this notification's details will be
|
|
// 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.Infof("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)
|
|
delete(confSet.ntfns, confID)
|
|
}
|
|
|
|
// 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 {
|
|
err = n.dispatchConfDetails(ntfn, details)
|
|
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 no details are provided, return early as we can't dispatch.
|
|
if details == nil {
|
|
Log.Debugf("Unable to dispatch %v, no details provided",
|
|
ntfn.ConfRequest)
|
|
|
|
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.Infof("Dispatching %v confirmation notification for %v",
|
|
ntfn.NumConfirmations, ntfn.ConfRequest)
|
|
|
|
// We'll send a 0 value to the Updates channel,
|
|
// indicating that the transaction/output script has already
|
|
// been confirmed.
|
|
select {
|
|
case ntfn.Event.Updates <- 0:
|
|
case <-n.quit:
|
|
return ErrTxNotifierExiting
|
|
}
|
|
|
|
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
|
|
select {
|
|
case ntfn.Event.Updates <- numConfsLeft:
|
|
case <-n.quit:
|
|
return ErrTxNotifierExiting
|
|
}
|
|
}
|
|
|
|
// 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", 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.Infof("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",
|
|
ntfn.SpendRequest)
|
|
|
|
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.Infof("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(tx, nil, 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)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
|
|
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 {
|
|
return nil
|
|
}
|
|
|
|
Log.Infof("Dispatching confirmed spend notification for %v at height=%d",
|
|
ntfn.SpendRequest, n.currentHeight)
|
|
|
|
select {
|
|
case ntfn.Event.Spend <- details:
|
|
ntfn.dispatched = true
|
|
case <-n.quit:
|
|
return ErrTxNotifierExiting
|
|
}
|
|
|
|
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(blockHash *chainhash.Hash, blockHeight uint32,
|
|
txns []*btcutil.Tx) 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.
|
|
for _, tx := range txns {
|
|
n.filterTx(
|
|
tx, blockHash, 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
|
|
}
|
|
}
|
|
|
|
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(tx *btcutil.Tx, blockHash *chainhash.Hash,
|
|
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))
|
|
}
|
|
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, blockHash)
|
|
|
|
details := &TxConfirmation{
|
|
Tx: tx.MsgTx(),
|
|
BlockHash: blockHash,
|
|
BlockHeight: blockHeight,
|
|
TxIndex: uint32(tx.Index()),
|
|
}
|
|
|
|
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{}{}
|
|
}
|
|
|
|
// 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 {
|
|
txConfHeight := confSet.details.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
|
|
}
|
|
|
|
select {
|
|
case ntfn.Event.Updates <- numConfsLeft:
|
|
case <-n.quit:
|
|
return ErrTxNotifierExiting
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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]
|
|
|
|
Log.Infof("Dispatching %v confirmation notification for %v",
|
|
ntfn.NumConfirmations, ntfn.ConfRequest)
|
|
|
|
select {
|
|
case ntfn.Event.Confirmed <- confSet.details:
|
|
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:
|
|
}
|
|
|
|
// 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]
|
|
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 _, ntfn := range confSet.ntfns {
|
|
close(ntfn.Event.Confirmed)
|
|
close(ntfn.Event.Updates)
|
|
close(ntfn.Event.NegativeConf)
|
|
close(ntfn.Event.Done)
|
|
}
|
|
}
|
|
|
|
for _, spendSet := range n.spendNotifications {
|
|
for _, ntfn := range spendSet.ntfns {
|
|
close(ntfn.Event.Spend)
|
|
close(ntfn.Event.Reorg)
|
|
close(ntfn.Event.Done)
|
|
}
|
|
}
|
|
}
|