769 lines
25 KiB
Go
769 lines
25 KiB
Go
package chainntnfs
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/btcsuite/btcd/chaincfg/chainhash"
|
|
"github.com/btcsuite/btcutil"
|
|
)
|
|
|
|
var (
|
|
// ErrTxConfNotifierExiting is an error returned when attempting to
|
|
// interact with the TxConfNotifier but it been shut down.
|
|
ErrTxConfNotifierExiting = errors.New("TxConfNotifier is exiting")
|
|
)
|
|
|
|
// 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.
|
|
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
|
|
|
|
// details describes the transaction's position is the blockchain. May be
|
|
// nil for unconfirmed transactions.
|
|
details *TxConfirmation
|
|
|
|
// 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.
|
|
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
|
|
}
|
|
|
|
// NewConfirmationEvent constructs a new ConfirmationEvent with newly opened
|
|
// channels.
|
|
func NewConfirmationEvent(numConfs uint32) *ConfirmationEvent {
|
|
return &ConfirmationEvent{
|
|
Confirmed: make(chan *TxConfirmation, 1),
|
|
Updates: make(chan uint32, numConfs),
|
|
NegativeConf: make(chan int32, 1),
|
|
}
|
|
}
|
|
|
|
// TxConfNotifier is used to register transaction confirmation notifications and
|
|
// dispatch them as the transactions confirm. A client can request to be
|
|
// notified when a particular transaction has sufficient on-chain confirmations
|
|
// (or be notified immediately if the tx already does), and the TxConfNotifier
|
|
// will watch changes to the blockchain in order to satisfy these requests.
|
|
type TxConfNotifier 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 confirmation 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 reorganized
|
|
// 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{}
|
|
|
|
// hintCache 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.
|
|
hintCache ConfirmHintCache
|
|
|
|
// quit is closed in order to signal that the notifier is gracefully
|
|
// exiting.
|
|
quit chan struct{}
|
|
|
|
sync.Mutex
|
|
}
|
|
|
|
// rescanState indicates the progression of a registration before the notifier
|
|
// can begin dispatching confirmations at tip.
|
|
type rescanState uint8
|
|
|
|
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 map[uint64]*ConfNtfn
|
|
rescanStatus rescanState
|
|
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,
|
|
}
|
|
}
|
|
|
|
// NewTxConfNotifier creates a TxConfNotifier. The current height of the
|
|
// blockchain is accepted as a parameter.
|
|
func NewTxConfNotifier(startHeight uint32, reorgSafetyLimit uint32,
|
|
hintCache ConfirmHintCache) *TxConfNotifier {
|
|
|
|
return &TxConfNotifier{
|
|
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{}),
|
|
hintCache: hintCache,
|
|
quit: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// Register 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 (tcn *TxConfNotifier) Register(
|
|
ntfn *ConfNtfn) (*HistoricalConfDispatch, error) {
|
|
|
|
select {
|
|
case <-tcn.quit:
|
|
return nil, ErrTxConfNotifierExiting
|
|
default:
|
|
}
|
|
|
|
// 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 := tcn.hintCache.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)
|
|
}
|
|
|
|
tcn.Lock()
|
|
defer tcn.Unlock()
|
|
|
|
confSet, ok := tcn.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()
|
|
tcn.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, tcn.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 > tcn.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: tcn.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 (tcn *TxConfNotifier) UpdateConfDetails(txid chainhash.Hash,
|
|
details *TxConfirmation) error {
|
|
|
|
select {
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
default:
|
|
}
|
|
|
|
// Ensure we hold the lock throughout handling the notification to
|
|
// prevent the notifier from advancing its height underneath us.
|
|
tcn.Lock()
|
|
defer tcn.Unlock()
|
|
|
|
// First, we'll determine whether we have an active notification for
|
|
// this transaction with the given ID.
|
|
confSet, ok := tcn.confNotifications[txid]
|
|
if !ok {
|
|
return fmt.Errorf("no notification found with TxID %v", txid)
|
|
}
|
|
|
|
// 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)
|
|
return nil
|
|
}
|
|
|
|
if details.BlockHeight > tcn.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 := tcn.hintCache.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 = tcn.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 (tcn *TxConfNotifier) 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
|
|
}
|
|
|
|
// Set the confirmation details for this notification, only if the
|
|
// notification doesn't already have details. This ensure we only fall
|
|
// through the following logic at most once, which could cause the
|
|
// buffered channels to block when exceeding their allocated capacity.
|
|
if ntfn.details != nil {
|
|
return nil
|
|
}
|
|
ntfn.details = details
|
|
|
|
// 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 <= tcn.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 <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
}
|
|
|
|
select {
|
|
case ntfn.Event.Confirmed <- details:
|
|
ntfn.dispatched = true
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
}
|
|
} 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 := tcn.ntfnsByConfirmHeight[confHeight]
|
|
if !exists {
|
|
ntfnSet = make(map[*ConfNtfn]struct{})
|
|
tcn.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 - tcn.currentHeight
|
|
select {
|
|
case ntfn.Event.Updates <- numConfsLeft:
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
}
|
|
}
|
|
|
|
// 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 + tcn.reorgSafetyLimit
|
|
if reorgSafeHeight > tcn.currentHeight {
|
|
txSet, exists := tcn.txsByInitialHeight[blockHeight]
|
|
if !exists {
|
|
txSet = make(map[chainhash.Hash]struct{})
|
|
tcn.txsByInitialHeight[blockHeight] = txSet
|
|
}
|
|
txSet[*ntfn.TxID] = struct{}{}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ConnectTip handles a new block extending the current chain. This checks each
|
|
// transaction in the block to see if any watched transactions are included.
|
|
// Also, if any watched transactions now have the required number of
|
|
// confirmations as a result of this block being connected, this dispatches
|
|
// notifications.
|
|
func (tcn *TxConfNotifier) ConnectTip(blockHash *chainhash.Hash,
|
|
blockHeight uint32, txns []*btcutil.Tx) error {
|
|
|
|
select {
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
default:
|
|
}
|
|
|
|
tcn.Lock()
|
|
defer tcn.Unlock()
|
|
|
|
if blockHeight != tcn.currentHeight+1 {
|
|
return fmt.Errorf("Received blocks out of order: "+
|
|
"current height=%d, new height=%d",
|
|
tcn.currentHeight, blockHeight)
|
|
}
|
|
tcn.currentHeight++
|
|
tcn.reorgDepth = 0
|
|
|
|
// Record any newly confirmed transactions by their confirmed height so
|
|
// that notifications get dispatched when the transactions reach their
|
|
// required number of confirmations. We'll also watch these transactions
|
|
// at the height they were included in the chain so reorgs can be
|
|
// handled correctly.
|
|
for _, tx := range txns {
|
|
txHash := tx.Hash()
|
|
|
|
// Check if we have any pending notifications for this txid. If
|
|
// none are found, we can proceed to the next transaction.
|
|
confSet, ok := tcn.confNotifications[*txHash]
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
Log.Debugf("Block contains txid=%v, constructing details",
|
|
txHash)
|
|
|
|
details := &TxConfirmation{
|
|
BlockHash: blockHash,
|
|
BlockHeight: blockHeight,
|
|
TxIndex: uint32(tx.Index()),
|
|
}
|
|
|
|
confSet.details = details
|
|
for _, ntfn := range confSet.ntfns {
|
|
ntfn.details = details
|
|
|
|
confHeight := blockHeight + ntfn.NumConfirmations - 1
|
|
ntfnSet, exists := tcn.ntfnsByConfirmHeight[confHeight]
|
|
if !exists {
|
|
ntfnSet = make(map[*ConfNtfn]struct{})
|
|
tcn.ntfnsByConfirmHeight[confHeight] = ntfnSet
|
|
}
|
|
ntfnSet[ntfn] = struct{}{}
|
|
|
|
txSet, exists := tcn.txsByInitialHeight[blockHeight]
|
|
if !exists {
|
|
txSet = make(map[chainhash.Hash]struct{})
|
|
tcn.txsByInitialHeight[blockHeight] = txSet
|
|
}
|
|
txSet[*txHash] = struct{}{}
|
|
}
|
|
}
|
|
|
|
// In order 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 current
|
|
// height. To do so, we'll iterate over the confNotifications map, which
|
|
// contains the transactions we currently have notifications for. Since
|
|
// this map doesn't tell us whether the transaction has confirmed or
|
|
// not, we'll need to look at txsByInitialHeight to determine so.
|
|
var txsToUpdateHints []chainhash.Hash
|
|
for confirmedTx := range tcn.txsByInitialHeight[tcn.currentHeight] {
|
|
txsToUpdateHints = append(txsToUpdateHints, confirmedTx)
|
|
}
|
|
out:
|
|
for maybeUnconfirmedTx, confSet := range tcn.confNotifications {
|
|
// We shouldn't update the confirm hints if we still have a
|
|
// pending rescan in progress. We'll skip writing any for
|
|
// notification sets that haven't reached rescanComplete.
|
|
if confSet.rescanStatus != rescanComplete {
|
|
continue
|
|
}
|
|
|
|
for height, confirmedTxs := range tcn.txsByInitialHeight {
|
|
// Skip the transactions that confirmed at the new block
|
|
// height as those have already been added.
|
|
if height == blockHeight {
|
|
continue
|
|
}
|
|
|
|
// If the transaction was found within the set of
|
|
// confirmed transactions at this height, we'll skip it.
|
|
if _, ok := confirmedTxs[maybeUnconfirmedTx]; ok {
|
|
continue out
|
|
}
|
|
}
|
|
txsToUpdateHints = append(txsToUpdateHints, maybeUnconfirmedTx)
|
|
}
|
|
|
|
if len(txsToUpdateHints) > 0 {
|
|
err := tcn.hintCache.CommitConfirmHint(
|
|
tcn.currentHeight, txsToUpdateHints...,
|
|
)
|
|
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", tcn.currentHeight, txsToUpdateHints,
|
|
err)
|
|
}
|
|
}
|
|
|
|
// 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 tcn.txsByInitialHeight {
|
|
for txHash := range txHashes {
|
|
confSet := tcn.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 <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Then, we'll dispatch notifications for all the transactions that have
|
|
// become confirmed at this new block height.
|
|
for ntfn := range tcn.ntfnsByConfirmHeight[tcn.currentHeight] {
|
|
Log.Infof("Dispatching %v conf notification for %v",
|
|
ntfn.NumConfirmations, ntfn.TxID)
|
|
|
|
select {
|
|
case ntfn.Event.Confirmed <- ntfn.details:
|
|
ntfn.dispatched = true
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
}
|
|
}
|
|
delete(tcn.ntfnsByConfirmHeight, tcn.currentHeight)
|
|
|
|
// Clear entries from confNotifications and confTxsByInitialHeight. We
|
|
// assume that reorgs deeper than the reorg safety limit do not happen,
|
|
// so we can clear out entries for the block that is now mature.
|
|
if tcn.currentHeight >= tcn.reorgSafetyLimit {
|
|
matureBlockHeight := tcn.currentHeight - tcn.reorgSafetyLimit
|
|
for txHash := range tcn.txsByInitialHeight[matureBlockHeight] {
|
|
delete(tcn.confNotifications, txHash)
|
|
}
|
|
delete(tcn.txsByInitialHeight, matureBlockHeight)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// DisconnectTip handles the tip of the current chain being disconnected during
|
|
// a chain reorganization. If any watched transactions were included in this
|
|
// block, internal structures are updated to ensure a confirmation notification
|
|
// is not sent unless the transaction is included in the new chain.
|
|
func (tcn *TxConfNotifier) DisconnectTip(blockHeight uint32) error {
|
|
select {
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
default:
|
|
}
|
|
|
|
tcn.Lock()
|
|
defer tcn.Unlock()
|
|
|
|
if blockHeight != tcn.currentHeight {
|
|
return fmt.Errorf("Received blocks out of order: "+
|
|
"current height=%d, disconnected height=%d",
|
|
tcn.currentHeight, blockHeight)
|
|
}
|
|
tcn.currentHeight--
|
|
tcn.reorgDepth++
|
|
|
|
// Rewind the height hint for all watched transactions.
|
|
var txs []chainhash.Hash
|
|
for tx := range tcn.confNotifications {
|
|
txs = append(txs, tx)
|
|
}
|
|
|
|
err := tcn.hintCache.CommitConfirmHint(tcn.currentHeight, txs...)
|
|
if err != nil {
|
|
Log.Errorf("Unable to update confirm hint to %d for %v: %v",
|
|
tcn.currentHeight, txs, err)
|
|
return err
|
|
}
|
|
|
|
// 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 tcn.txsByInitialHeight {
|
|
for txHash := range txHashes {
|
|
confSet := tcn.confNotifications[txHash]
|
|
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 <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
default:
|
|
}
|
|
|
|
// Then, we'll check if the current transaction
|
|
// was included in the block currently being
|
|
// disconnected. If it was, we'll need to take
|
|
// some necessary precautions.
|
|
if initialHeight == blockHeight {
|
|
// If the transaction's confirmation notification
|
|
// has already been dispatched, we'll attempt to
|
|
// notify the client it was reorged out of the chain.
|
|
if ntfn.dispatched {
|
|
// Attempt to drain the confirmation notification
|
|
// to ensure sends to the Confirmed channel are
|
|
// always non-blocking.
|
|
select {
|
|
case <-ntfn.Event.Confirmed:
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
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(tcn.reorgDepth):
|
|
case <-tcn.quit:
|
|
return ErrTxConfNotifierExiting
|
|
}
|
|
|
|
continue
|
|
}
|
|
|
|
// Otherwise, since the transactions was reorged out
|
|
// of the chain, we can safely remove its accompanying
|
|
// confirmation notification.
|
|
confHeight := blockHeight + ntfn.NumConfirmations - 1
|
|
ntfnSet, exists := tcn.ntfnsByConfirmHeight[confHeight]
|
|
if !exists {
|
|
continue
|
|
}
|
|
delete(ntfnSet, ntfn)
|
|
|
|
// Intuitively, we should also remove
|
|
// the txHash from confNotifications if
|
|
// the ntfnSet is now empty. However, we
|
|
// will not do so since we may want to
|
|
// continue rewinding the height hints
|
|
// for this txid.
|
|
//
|
|
// NOTE(conner): safe to delete if
|
|
// blockHeight is below client-provided
|
|
// height hint?
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Finally, we can remove the transactions we're currently watching that
|
|
// were included in this block height.
|
|
delete(tcn.txsByInitialHeight, blockHeight)
|
|
|
|
return nil
|
|
}
|
|
|
|
// TearDown is to be called when the owner of the TxConfNotifier is exiting.
|
|
// This closes the event channels of all registered notifications that have
|
|
// not been dispatched yet.
|
|
func (tcn *TxConfNotifier) TearDown() {
|
|
tcn.Lock()
|
|
defer tcn.Unlock()
|
|
|
|
close(tcn.quit)
|
|
|
|
for _, confSet := range tcn.confNotifications {
|
|
for _, ntfn := range confSet.ntfns {
|
|
if ntfn.dispatched {
|
|
continue
|
|
}
|
|
|
|
close(ntfn.Event.Confirmed)
|
|
close(ntfn.Event.Updates)
|
|
close(ntfn.Event.NegativeConf)
|
|
}
|
|
}
|
|
}
|