lnd.xprv/chainntnfs/txnotifier.go

1389 lines
46 KiB
Go
Raw Normal View History

package chainntnfs
import (
"errors"
"fmt"
"sync"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil"
)
var (
// ErrTxNotifierExiting is an error returned when attempting to interact
// with the TxNotifier but it been shut down.
ErrTxNotifierExiting = errors.New("TxNotifier is exiting")
// ErrTxMaxConfs signals that the user requested a number of
// confirmations beyond the reorg safety limit.
ErrTxMaxConfs = errors.New("too many confirmations requested")
)
// 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
// single txid. 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.
ntfns map[uint64]*ConfNtfn
// rescanStatus represents the current rescan state for the transaction.
rescanStatus rescanState
// details serves as a cache of the confirmation details of a
// transaction that we'll use to determine if a transaction has already
// confirmed at the time of registration.
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 an outpoint.
// 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.
ntfns map[uint64]*SpendNtfn
// rescanStatus represents the current rescan state for the outpoint.
rescanStatus rescanState
// details serves as a cache of the spend details for an outpoint that
// we'll use to determine if an outpoint 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,
}
}
// ConfNtfn represents a notifier client's request to receive a notification
// once the target transaction 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.
ConfID uint64
// TxID is the hash of the transaction for which confirmation notifications
// are requested.
TxID *chainhash.Hash
// PkScript is the public key script of an outpoint created in this
// transaction.
//
// NOTE: This value MUST be set when the dispatch is to be performed
// using compact filters.
PkScript []byte
// 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 identifier. The parameters include the start and end block
// heights specifying the range of blocks to scan.
type HistoricalConfDispatch struct {
// TxID is the transaction ID to search for in the historical dispatch.
TxID *chainhash.Hash
// PkScript is a public key script from an output created by this
// transaction.
//
// NOTE: This value MUST be set when the dispatch is to be performed
// using compact filters.
PkScript []byte
// 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
}
// SpendNtfn represents a client's request to receive a notification once an
// outpoint 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.
SpendID uint64
// OutPoint is the outpoint for which a client has requested a spend
// notification for.
OutPoint wire.OutPoint
// PkScript is the script of the outpoint. This is needed in order to
// match compact filters when attempting a historical rescan to
// determine if the outpoint has already been spent.
PkScript []byte
// 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. 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. The parameters include the start
// and end block heights specifying the range of blocks to scan.
type HistoricalSpendDispatch struct {
// OutPoint is the outpoint which we should attempt to find the spending
OutPoint wire.OutPoint
// PkScript is the script of the outpoint. This is needed in order to
// match compact filters when attempting a historical rescan.
PkScript []byte
// 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
}
// TxNotifier is a struct responsible for delivering transaction notifications
// to subscribers. These notifications can be of two different types:
// transaction confirmations and/or outpoint spends. The TxNotifier will watch
// the blockchain as new blocks come in, in order to satisfy its client
// requests.
type TxNotifier struct {
// 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 notification requests by transaction
// hash.
confNotifications map[chainhash.Hash]*confNtfnSet
// txsByInitialHeight is an index of watched transactions by the height
// that they are included at in the blockchain. This is tracked so that
// incorrect notifications are not sent if a transaction is reorged out
// of the chain and so that negative confirmations can be recognized.
txsByInitialHeight map[uint32]map[chainhash.Hash]struct{}
// ntfnsByConfirmHeight is an index of notification requests by the
// height at which the transaction will have sufficient confirmations.
ntfnsByConfirmHeight map[uint32]map[*ConfNtfn]struct{}
// spendNotifications is an index of all active notification requests
// per outpoint.
spendNotifications map[wire.OutPoint]*spendNtfnSet
// opsBySpendHeight is an index that keeps tracks of the spending height
// of an outpoint we are currently tracking notifications for. This is
// used in order to recover from the spending transaction of an outpoint
// being reorged out of the chain.
opsBySpendHeight map[uint32]map[wire.OutPoint]struct{}
// confirmHintCache is a cache used to maintain the latest height hints
// for transactions. Each height hint represents the earliest height at
// which the transactions could have been confirmed within the chain.
confirmHintCache ConfirmHintCache
// spendHintCache is a cache used to maintain the latest height hints
// for outpoints. Each height hint represents the earliest height at
// which the outpoints 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[chainhash.Hash]*confNtfnSet),
txsByInitialHeight: make(map[uint32]map[chainhash.Hash]struct{}),
ntfnsByConfirmHeight: make(map[uint32]map[*ConfNtfn]struct{}),
spendNotifications: make(map[wire.OutPoint]*spendNtfnSet),
opsBySpendHeight: make(map[uint32]map[wire.OutPoint]struct{}),
confirmHintCache: confirmHintCache,
spendHintCache: spendHintCache,
quit: make(chan struct{}),
}
}
// RegisterConf handles a new notification request. The client will be notified
// when the transaction gets a sufficient number of confirmations on the
// blockchain. The registration succeeds if no error is returned. If the
// returned HistoricalConfDispatch is non-nil, the caller is responsible for
// attempting to manually rescan blocks for the txid between the start and end
// heights.
//
// NOTE: If the transaction 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 to confirm even though it already
// has.
func (n *TxNotifier) RegisterConf(ntfn *ConfNtfn) (*HistoricalConfDispatch, error) {
select {
case <-n.quit:
return nil, ErrTxNotifierExiting
default:
}
// Enforce that we will not dispatch confirmations beyond the reorg
// safety limit.
if ntfn.NumConfirmations > n.reorgSafetyLimit {
return nil, ErrTxMaxConfs
}
// 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.TxID)
if err == nil {
if hint > startHeight {
Log.Debugf("Using height hint %d retrieved "+
"from cache for %v", hint, *ntfn.TxID)
startHeight = hint
}
} else if err != ErrConfirmHintNotFound {
Log.Errorf("Unable to query confirm hint for %v: %v",
*ntfn.TxID, err)
}
n.Lock()
defer n.Unlock()
confSet, ok := n.confNotifications[*ntfn.TxID]
if !ok {
// If this is the first registration for this txid, construct a
// confSet to coalesce all notifications for the same txid.
confSet = newConfNtfnSet()
n.confNotifications[*ntfn.TxID] = confSet
}
confSet.ntfns[ntfn.ConfID] = ntfn
switch confSet.rescanStatus {
// A prior rescan has already completed and we are actively watching at
// tip for this txid.
case rescanComplete:
// If conf 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 conf for txid=%v "+
"on registration since rescan has finished", ntfn.TxID)
return nil, n.dispatchConfDetails(ntfn, confSet.details)
// A rescan is already in progress, return here to prevent dispatching
// another. When the scan returns, this notifications details will be
// updated as well.
case rescanPending:
Log.Debugf("Waiting for pending rescan to finish before "+
"notifying txid=%v at tip", ntfn.TxID)
return nil, 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
// is to be confirmed at a height greater than the conf 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 rescan for txid=%v ", ntfn.TxID)
// Set the rescan status to complete, which will allow the conf
// notifier to start delivering messages for this set
// immediately.
confSet.rescanStatus = rescanComplete
return nil, nil
}
Log.Debugf("Dispatching historical rescan for txid=%v ", ntfn.TxID)
// 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{
TxID: ntfn.TxID,
PkScript: ntfn.PkScript,
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 dispatch, nil
}
// 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 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(txid chainhash.Hash,
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 notification for
// this transaction with the given ID.
confSet, ok := n.confNotifications[txid]
if !ok {
return fmt.Errorf("no notification found with TxID %v", txid)
}
// If the conf 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 was
// included in a block, so we should defer until handling it then within
// ConnectTip.
if details == nil {
Log.Debugf("Conf details for txid=%v not found during "+
"historical dispatch, waiting to dispatch at tip", txid)
// 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, txid,
)
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, txid, err)
}
return nil
}
if details.BlockHeight > n.currentHeight {
Log.Debugf("Conf details for txid=%v found above current "+
"height, waiting to dispatch at tip", txid)
return nil
}
Log.Debugf("Updating conf details for txid=%v details", txid)
err := n.confirmHintCache.CommitConfirmHint(details.BlockHeight, txid)
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, txid, 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 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.TxID)
return nil
}
// Now, we'll examine whether the transaction of this
// notification 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 conf notification for %v",
ntfn.NumConfirmations, ntfn.TxID)
// We'll send a 0 value to the Updates channel,
// indicating that the transaction 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 conf notification for %v at tip ",
ntfn.NumConfirmations, ntfn.TxID)
// 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 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 if it's
// still possible for it to get reorged out of the chain.
blockHeight := details.BlockHeight
reorgSafeHeight := blockHeight + n.reorgSafetyLimit
if reorgSafeHeight > n.currentHeight {
txSet, exists := n.txsByInitialHeight[blockHeight]
if !exists {
txSet = make(map[chainhash.Hash]struct{})
n.txsByInitialHeight[blockHeight] = txSet
}
txSet[*ntfn.TxID] = struct{}{}
}
return nil
}
// RegisterSpend handles a new spend notification request. The client will be
// notified once the outpoint is detected as spent within the chain.
//
// The registration succeeds if no error is returned. If the returned
// HistoricalSpendDisaptch is non-nil, the caller is responsible for attempting
// to determine whether the outpoint has been spent between the start and end
// heights.
//
// NOTE: If the outpoint 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 to
// be spent at tip, even though it already has.
func (n *TxNotifier) RegisterSpend(ntfn *SpendNtfn) (*HistoricalSpendDispatch, error) {
select {
case <-n.quit:
return nil, ErrTxNotifierExiting
default:
}
// 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.OutPoint)
if err == nil {
if hint > startHeight {
Log.Debugf("Using height hint %d retrieved from cache "+
"for %v", startHeight, ntfn.OutPoint)
startHeight = hint
}
} else if err != ErrSpendHintNotFound {
Log.Errorf("Unable to query spend hint for %v: %v",
ntfn.OutPoint, err)
}
n.Lock()
defer n.Unlock()
Log.Infof("New spend subscription: spend_id=%d, outpoint=%v, "+
"height_hint=%d", ntfn.SpendID, ntfn.OutPoint, ntfn.HeightHint)
// Keep track of the notification request so that we can properly
// dispatch a spend notification later on.
spendSet, ok := n.spendNotifications[ntfn.OutPoint]
if !ok {
// If this is the first registration for the outpoint, we'll
// construct a spendNtfnSet to coalesce all notifications.
spendSet = newSpendNtfnSet()
n.spendNotifications[ntfn.OutPoint] = 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 outpoint have already been
// determined and cached, then we can use them to immediately dispatch
// the spend notification to the client.
case rescanComplete:
return nil, n.dispatchSpendDetails(ntfn, spendSet.details)
// If there is an active rescan to determine whether the outpoint has
// been spent, then we won't trigger another one.
case rescanPending:
return nil, nil
// Otherwise, we'll fall through and let the caller know that a rescan
// should be dispatched to determine whether the outpoint 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.OutPoint, n.currentHeight)
// We'll also set the rescan status as complete to ensure that
// spend hints for this outpoint get updated upon
// connected/disconnected blocks.
spendSet.rescanStatus = rescanComplete
return nil, nil
}
// We'll set the rescan status to pending to ensure subsequent
// notifications don't also attempt a historical dispatch.
spendSet.rescanStatus = rescanPending
return &HistoricalSpendDispatch{
OutPoint: ntfn.OutPoint,
PkScript: ntfn.PkScript,
StartHeight: startHeight,
EndHeight: n.currentHeight,
}, nil
}
// CancelSpend cancels an existing request for a spend notification of an
// outpoint. The request is identified by its spend ID.
func (n *TxNotifier) CancelSpend(op wire.OutPoint, spendID uint64) {
select {
case <-n.quit:
return
default:
}
n.Lock()
defer n.Unlock()
Log.Infof("Canceling spend notification: spend_id=%d, outpoint=%v",
spendID, op)
spendSet, ok := n.spendNotifications[op]
if !ok {
return
}
ntfn, ok := spendSet.ntfns[spendID]
if !ok {
return
}
// 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)
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
// outputs 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 *wire.MsgTx, txHeight int32) 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()
// Grab the set of active registered outpoints to determine if the
// transaction spends any of them.
spendNtfns := n.spendNotifications
// We'll check if this transaction spends an output that has an existing
// spend notification for it.
for i, txIn := range tx.TxIn {
// If this input doesn't spend an existing registered outpoint,
// we'll go on to the next.
prevOut := txIn.PreviousOutPoint
if _, ok := spendNtfns[prevOut]; !ok {
continue
}
// Otherwise, we'll create a spend summary and send off the
// details to the notification subscribers.
txHash := tx.TxHash()
details := &SpendDetail{
SpentOutPoint: &prevOut,
SpenderTxHash: &txHash,
SpendingTx: tx,
SpenderInputIndex: uint32(i),
SpendingHeight: txHeight,
}
if err := n.updateSpendDetails(prevOut, details); err != nil {
return err
}
}
return nil
}
// UpdateSpendDetails attempts to update the spend details for all active spend
// notification requests for an outpoint. 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 the outpoint, the spend details may be nil.
//
// NOTE: A notification request for the outpoint must be registered first to
// ensure notifications are delivered.
func (n *TxNotifier) UpdateSpendDetails(op wire.OutPoint,
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(op, details)
}
// updateSpendDetails attempts to update the spend details for all active spend
// notification requests for an outpoint. 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 the outpoint, the spend details may be nil.
//
// NOTE: This method must be called with the TxNotifier's lock held.
func (n *TxNotifier) updateSpendDetails(op wire.OutPoint,
details *SpendDetail) error {
// Mark the ongoing historical rescan for this outpoint as finished.
// This will allow us to update the spend hints for this outpoint at
// tip.
spendSet, ok := n.spendNotifications[op]
if !ok {
return fmt.Errorf("no notifications found for outpoint %v", op)
}
// 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 outpoint, 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 outpoint, 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, op)
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, op, err)
}
return nil
}
// If the historical rescan found the spending transaction for this
// outpoint, 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 outpoint 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), op,
)
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, op, 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 spend notification for outpoint=%v at height=%d",
ntfn.OutPoint, 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 for which we
// currently have an active spend registration for.
//
// 2. The transaction is a transaction for which we currently have an active
// confirmation registration for.
//
// In the event that the transaction is relevant, a confirmation/spend
// notification will be dispatched to the relevant clients. Confirmation
// notifications will only be dispatched for transactions that have met the
// required number of confirmations required by the client.
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 {
txHash := tx.Hash()
// In order to determine if this transaction is relevant to the
// notifier, we'll check its inputs for any outstanding spend
// notifications.
for i, txIn := range tx.MsgTx().TxIn {
prevOut := txIn.PreviousOutPoint
spendSet, ok := n.spendNotifications[prevOut]
if !ok {
continue
}
// If we have any, we'll record its spend height so that
// notifications get dispatched to the respective
// clients.
spendDetails := &SpendDetail{
SpentOutPoint: &prevOut,
SpenderTxHash: txHash,
SpendingTx: tx.MsgTx(),
SpenderInputIndex: uint32(i),
SpendingHeight: int32(blockHeight),
}
// TODO(wilmer): cancel pending historical rescans if any?
spendSet.rescanStatus = rescanComplete
spendSet.details = spendDetails
for _, ntfn := range spendSet.ntfns {
// In the event that this notification was aware
// that the spending transaction of its outpoint
// 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 outpoints spending height in order to
// correctly handle dispatching notifications when the
// spending transactions gets reorged out of the chain.
opSet, exists := n.opsBySpendHeight[blockHeight]
if !exists {
opSet = make(map[wire.OutPoint]struct{})
n.opsBySpendHeight[blockHeight] = opSet
}
opSet[prevOut] = struct{}{}
}
// Check if we have any pending notifications for this txid. If
// none are found, we can proceed to the next transaction.
confSet, ok := n.confNotifications[*txHash]
if !ok {
continue
}
Log.Debugf("Block contains txid=%v, constructing details",
txHash)
// If we have any, we'll record its confirmed height so that
// notifications get dispatched when the transaction reaches the
// clients' desired number of confirmations.
details := &TxConfirmation{
BlockHash: blockHash,
BlockHeight: blockHeight,
TxIndex: uint32(tx.Index()),
}
// TODO(wilmer): cancel pending historical rescans if any?
confSet.rescanStatus = rescanComplete
confSet.details = details
for _, ntfn := range confSet.ntfns {
// In the event that this notification was aware that
// the transaction 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 := 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 gets reorged out of the chain.
txSet, exists := n.txsByInitialHeight[blockHeight]
if !exists {
txSet = make(map[chainhash.Hash]struct{})
n.txsByInitialHeight[blockHeight] = txSet
}
txSet[*txHash] = struct{}{}
}
}
// Now that we've determined which transactions were confirmed and which
// outpoints were spent within the new block, we can update their
// entries in their respective caches, along with all of our unconfirmed
// transactions and unspent outpoints.
n.updateHints(blockHeight)
// Next, we'll dispatch an update to all of the notification clients for
// our watched transactions with the number of confirmations left at
// this new height.
for _, txHashes := range n.txsByInitialHeight {
for txHash := range txHashes {
confSet := n.confNotifications[txHash]
for _, ntfn := range confSet.ntfns {
txConfHeight := confSet.details.BlockHeight +
ntfn.NumConfirmations - 1
numConfsLeft := txConfHeight - blockHeight
// Since we don't clear notifications until
// transactions are no longer under the risk of
// being reorganized out of the chain, we'll
// skip sending updates for transactions 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 transactions that have
// become confirmed at this new block height.
for ntfn := range n.ntfnsByConfirmHeight[blockHeight] {
confSet := n.confNotifications[*ntfn.TxID]
Log.Infof("Dispatching %v conf notification for %v",
ntfn.NumConfirmations, ntfn.TxID)
select {
case ntfn.Event.Confirmed <- confSet.details:
ntfn.dispatched = true
case <-n.quit:
return ErrTxNotifierExiting
}
}
delete(n.ntfnsByConfirmHeight, blockHeight)
// We'll also dispatch spend notifications for all the outpoints that
// were spent at this new block height.
for op := range n.opsBySpendHeight[blockHeight] {
spendSet := n.spendNotifications[op]
for _, ntfn := range spendSet.ntfns {
err := n.dispatchSpendDetails(ntfn, spendSet.details)
if err != nil {
return err
}
}
}
// Finally, we'll clear the entries from our set of notifications for
// transactions and outpoints that are no longer under the risk of being
// reorged out of the chain.
if blockHeight >= n.reorgSafetyLimit {
matureBlockHeight := blockHeight - n.reorgSafetyLimit
for tx := range n.txsByInitialHeight[matureBlockHeight] {
delete(n.confNotifications, tx)
}
delete(n.txsByInitialHeight, matureBlockHeight)
for op := range n.opsBySpendHeight[matureBlockHeight] {
delete(n.spendNotifications, op)
}
delete(n.opsBySpendHeight, matureBlockHeight)
}
return nil
}
// DisconnectTip handles the tip of the current chain being disconnected during
// a chain reorganization. If any watched transactions or spending transactions
// for registered outpoints 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 transactions and outpoints 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 transactions and attempt to drain
// their notification channels to ensure sending notifications to the
// clients is always non-blocking.
for initialHeight, txHashes := range n.txsByInitialHeight {
for txHash := range txHashes {
// If the transaction 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
// 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 outpoints and attempt to drain
// their dispatched notifications to ensure dispatching notifications to
// clients later on is always non-blocking. We're only interested in
// outpoints whose spending transaction was included at the height being
// disconnected.
for op := range n.opsBySpendHeight[blockHeight] {
// Since the spending transaction is being reorged out of the
// chain, we'll need to clear out the spending details of the
// outpoint.
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 transactions that were confirmed and the
// outpoints that were 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.txsByInitialHeight, blockHeight)
delete(n.opsBySpendHeight, blockHeight)
return nil
}
// updateHints attempts to update the confirm and spend hints for all relevant
// transactions and outpoints respectively. The height parameter is used to
// determine which transactions and outpoints 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 transactions under one
// database transaction, we'll gather the set of unconfirmed
// transactions along with the ones that confirmed at the height being
// connected/disconnected.
txsToUpdateHints := n.unconfirmedTxs()
for confirmedTx := range n.txsByInitialHeight[height] {
txsToUpdateHints = append(txsToUpdateHints, confirmedTx)
}
err := n.confirmHintCache.CommitConfirmHint(
n.currentHeight, txsToUpdateHints...,
)
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, txsToUpdateHints, err)
}
// Similarly, to update the height hint for all the required outpoints
// under one database transaction, we'll gather the set of unspent
// outpoints along with the ones that were spent at the height being
// connected/disconnected.
opsToUpdateHints := n.unspentOutPoints()
for spentOp := range n.opsBySpendHeight[height] {
opsToUpdateHints = append(opsToUpdateHints, spentOp)
}
err = n.spendHintCache.CommitSpendHint(
n.currentHeight, opsToUpdateHints...,
)
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, opsToUpdateHints, err)
}
}
// unconfirmedTxs returns the set of transactions that are still seen as
// unconfirmed by the TxNotifier.
//
// NOTE: This method must be called with the TxNotifier's lock held.
func (n *TxNotifier) unconfirmedTxs() []chainhash.Hash {
var unconfirmedTxs []chainhash.Hash
for tx, 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
}
unconfirmedTxs = append(unconfirmedTxs, tx)
}
return unconfirmedTxs
}
// unspentOutPoints returns the set of outpoints that are still seen as unspent
// by the TxNotifier.
//
// NOTE: This method must be called with the TxNotifier's lock held.
func (n *TxNotifier) unspentOutPoints() []wire.OutPoint {
var unspentOps []wire.OutPoint
for op, 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
}
unspentOps = append(unspentOps, op)
}
return unspentOps
}
// 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 transaction'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() {
n.Lock()
defer n.Unlock()
close(n.quit)
for _, confSet := range n.confNotifications {
for _, ntfn := range confSet.ntfns {
close(ntfn.Event.Confirmed)
close(ntfn.Event.Updates)
close(ntfn.Event.NegativeConf)
}
}
for _, spendSet := range n.spendNotifications {
for _, ntfn := range spendSet.ntfns {
close(ntfn.Event.Spend)
close(ntfn.Event.Reorg)
}
}
}