Release v0.1.0

vendor/github.com/lightninglabs/neutrino/pushtx/broadcaster.go

@@ -0,0 +1,290 @@
+package pushtx
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/btcsuite/btcwallet/wtxmgr"
+	"github.com/lightninglabs/neutrino/blockntfns"
+)
+
+var (
+	// ErrBroadcastStopped is an error returned when we attempt to process a
+	// request to broadcast a transaction but the Broadcaster has already
+	// been stopped.
+	ErrBroadcasterStopped = errors.New("broadcaster has been stopped")
+)
+
+const (
+	// DefaultRebroadcastInterval is the default period that we'll wait
+	// between blocks to attempt another rebroadcast.
+	DefaultRebroadcastInterval = time.Minute
+)
+
+// broadcastReq is an internal message the Broadcaster will use to process
+// transaction broadcast requests.
+type broadcastReq struct {
+	tx      *wire.MsgTx
+	errChan chan error
+}
+
+// Config contains all of the external dependencies required for the Broadcaster
+// to properly carry out its duties.
+type Config struct {
+	// Broadcast broadcasts a transaction to the network. We expect certain
+	// BroadcastError's to be returned to handle special cases, namely
+	// errors with the codes Mempool and Confirmed.
+	Broadcast func(*wire.MsgTx) error
+
+	// SubscribeBlocks returns a block subscription that delivers block
+	// notifications in order. This will be used to rebroadcast all
+	// transactions once a new block arrives.
+	SubscribeBlocks func() (*blockntfns.Subscription, error)
+
+	// RebroadcastInterval is the interval that we'll continually try to
+	// re-broadcast transactions in-between new block arrival.
+	RebroadcastInterval time.Duration
+}
+
+// Broadcaster is a subsystem responsible for reliably broadcasting transactions
+// to the network. Each transaction will be rebroadcast upon every new block
+// being connected/disconnected to/from the chain.
+type Broadcaster struct {
+	start sync.Once
+	stop  sync.Once
+
+	cfg Config
+
+	// broadcastReqs is a channel through which new transaction broadcast
+	// requests from external callers will be streamed through.
+	broadcastReqs chan *broadcastReq
+
+	quit chan struct{}
+	wg   sync.WaitGroup
+}
+
+// NewBroadcaster creates a new Broadcaster backed by the given config.
+func NewBroadcaster(cfg *Config) *Broadcaster {
+	b := &Broadcaster{
+		cfg:           *cfg,
+		broadcastReqs: make(chan *broadcastReq),
+		quit:          make(chan struct{}),
+	}
+
+	return b
+}
+
+// Start starts all of the necessary steps for the Broadcaster to begin properly
+// carrying out its duties.
+func (b *Broadcaster) Start() error {
+	var err error
+	b.start.Do(func() {
+		sub, err := b.cfg.SubscribeBlocks()
+		if err != nil {
+			err = fmt.Errorf("unable to subscribe for block "+
+				"notifications: %v", err)
+			return
+		}
+
+		b.wg.Add(1)
+		go b.broadcastHandler(sub)
+	})
+	return err
+}
+
+// Stop halts the Broadcaster from rebroadcasting pending transactions.
+func (b *Broadcaster) Stop() {
+	b.stop.Do(func() {
+		close(b.quit)
+		b.wg.Wait()
+	})
+}
+
+// broadcastHandler is the main event handler of the Broadcaster responsible for
+// handling new broadcast requests, rebroadcasting transactions upon every new
+// block, etc.
+//
+// NOTE: This must be run as a goroutine.
+func (b *Broadcaster) broadcastHandler(sub *blockntfns.Subscription) {
+	defer b.wg.Done()
+	defer sub.Cancel()
+
+	log.Infof("Broadcaster now active")
+
+	// transactions is the set of transactions we have broadcast so far,
+	// and are still not confirmed.
+	transactions := make(map[chainhash.Hash]*wire.MsgTx)
+
+	// confChan is a channel used to notify the broadcast handler about
+	// confirmed transactions.
+	confChan := make(chan chainhash.Hash)
+
+	// The rebroadcast semaphore is used to ensure we have only one
+	// rebroadcast running at a time.
+	rebroadcastSem := make(chan struct{}, 1)
+	rebroadcastSem <- struct{}{}
+
+	// triggerRebroadcast is a helper method that checks whether the
+	// rebroadcast semaphore is available, and if it is spawns a goroutine
+	// to rebroadcast all pending transactions.
+	triggerRebroadcast := func() {
+		select {
+		// If the rebroadcast semaphore is available, start a
+		// new goroutine to exectue a rebroadcast.
+		case <-rebroadcastSem:
+		default:
+			log.Tracef("Existing rebroadcast still in " +
+				"progress")
+			return
+		}
+
+		// Make a copy of the current set of transactions to hand to
+		// the goroutine.
+		txs := make(map[chainhash.Hash]*wire.MsgTx)
+		for k, v := range transactions {
+			txs[k] = v.Copy()
+		}
+
+		b.wg.Add(1)
+		go func() {
+			defer b.wg.Done()
+
+			b.rebroadcast(txs, confChan)
+			rebroadcastSem <- struct{}{}
+		}()
+
+	}
+
+	reBroadcastTicker := time.NewTicker(b.cfg.RebroadcastInterval)
+	defer reBroadcastTicker.Stop()
+
+	for {
+		select {
+		// A new broadcast request was submitted by an external caller.
+		case req := <-b.broadcastReqs:
+			err := b.cfg.Broadcast(req.tx)
+			if err != nil && !IsBroadcastError(err, Mempool) {
+				log.Errorf("Broadcast attempt failed: %v", err)
+				req.errChan <- err
+				continue
+			}
+
+			transactions[req.tx.TxHash()] = req.tx
+			req.errChan <- nil
+
+		// A tx was confirmed, and we can remove it from our set of
+		// transactions.
+		case txHash := <-confChan:
+			delete(transactions, txHash)
+
+		// A new block notification has arrived, so we'll rebroadcast
+		// all of our pending transactions.
+		case _, ok := <-sub.Notifications:
+			if !ok {
+				log.Warn("Unable to rebroadcast transactions: " +
+					"block subscription was canceled")
+				continue
+			}
+			triggerRebroadcast()
+
+		// Between blocks, we'll also try to attempt additional
+		// re-broadcasts to ensure a timely confirmation.
+		case <-reBroadcastTicker.C:
+			triggerRebroadcast()
+
+		case <-b.quit:
+			return
+		}
+	}
+}
+
+// rebroadcast rebroadcasts all of the currently pending transactions. Care has
+// been taken to ensure that the transactions are sorted in their dependency
+// order to prevent peers from deeming our transactions as invalid due to
+// broadcasting them before their pending dependencies.
+func (b *Broadcaster) rebroadcast(txs map[chainhash.Hash]*wire.MsgTx,
+	confChan chan<- chainhash.Hash) {
+
+	// Return immediately if there are no transactions to re-broadcast.
+	if len(txs) == 0 {
+		return
+	}
+
+	log.Debugf("Re-broadcasting %d transactions", len(txs))
+
+	sortedTxs := wtxmgr.DependencySort(txs)
+	for _, tx := range sortedTxs {
+		// Before attempting to broadcast this transaction, we check
+		// whether we are shutting down.
+		select {
+		case <-b.quit:
+			return
+		default:
+		}
+
+		err := b.cfg.Broadcast(tx)
+		switch {
+		// If the transaction has already confirmed on-chain, we can
+		// stop broadcasting it further.
+		//
+		// TODO(wilmer); This should ideally be implemented by checking
+		// the chain ourselves rather than trusting our peers.
+		case IsBroadcastError(err, Confirmed):
+			log.Debugf("Re-broadcast of txid=%v, now confirmed!",
+				tx.TxHash())
+
+			select {
+			case confChan <- tx.TxHash():
+			case <-b.quit:
+				return
+			}
+			continue
+
+		// If the transaction already exists within our peers' mempool,
+		// we'll continue to rebroadcast it to ensure it actually
+		// propagates throughout the network.
+		//
+		// TODO(wilmer): Rate limit peers that have already accepted our
+		// transaction into their mempool to prevent resending to them
+		// every time.
+		case IsBroadcastError(err, Mempool):
+			log.Debugf("Re-broadcast of txid=%v, still "+
+				"pending...", tx.TxHash())
+
+			continue
+
+		case err != nil:
+			log.Errorf("Unable to rebroadcast transaction %v: %v",
+				tx.TxHash(), err)
+			continue
+		}
+	}
+}
+
+// Broadcast submits a request to the Broadcaster to reliably broadcast the
+// given transaction. An error won't be returned if the transaction already
+// exists within the mempool. Any transaction broadcast through this method will
+// be rebroadcast upon every change of the tip of the chain.
+func (b *Broadcaster) Broadcast(tx *wire.MsgTx) error {
+	errChan := make(chan error, 1)
+
+	select {
+	case b.broadcastReqs <- &broadcastReq{
+		tx:      tx,
+		errChan: errChan,
+	}:
+	case <-b.quit:
+		return ErrBroadcasterStopped
+	}
+
+	select {
+	case err := <-errChan:
+		return err
+	case <-b.quit:
+		return ErrBroadcasterStopped
+	}
+}

vendor/github.com/lightninglabs/neutrino/pushtx/error.go

@@ -0,0 +1,152 @@
+package pushtx
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/btcsuite/btcd/wire"
+)
+
+// BroadcastErrorCode uniquely identifies the broadcast error.
+type BroadcastErrorCode uint8
+
+const (
+	// Unknown is the code used when a transaction has been rejected by some
+	// unknown reason by a peer.
+	Unknown BroadcastErrorCode = iota
+
+	// Invalid is the code used when a transaction has been deemed invalid
+	// by a peer.
+	Invalid
+
+	// InsufficientFee is the code used when a transaction has been deemed
+	// as having an insufficient fee by a peer.
+	InsufficientFee
+
+	// Mempool is the code used when a transaction already exists in a
+	// peer's mempool.
+	Mempool
+
+	// Confirmed is the code used when a transaction has been deemed as
+	// confirmed in the chain by a peer.
+	Confirmed
+)
+
+func (c BroadcastErrorCode) String() string {
+	switch c {
+	case Invalid:
+		return "Invalid"
+	case InsufficientFee:
+		return "InsufficientFee"
+	case Mempool:
+		return "Mempool"
+	case Confirmed:
+		return "Confirmed"
+	default:
+		return "Unknown"
+	}
+}
+
+// BroadcastError is an error type that encompasses the different possible
+// broadcast errors returned by the network.
+type BroadcastError struct {
+	// Code is the uniquely identifying code of the broadcast error.
+	Code BroadcastErrorCode
+
+	// Reason is the string detailing the reason as to why the transaction
+	// was rejected.
+	Reason string
+}
+
+// A compile-time constraint to ensure BroadcastError satisfies the error
+// interface.
+var _ error = (*BroadcastError)(nil)
+
+// Error returns the reason of the broadcast error.
+func (e *BroadcastError) Error() string {
+	return e.Reason
+}
+
+// IsBroadcastError is a helper function that can be used to determine whether
+// an error is a BroadcastError that matches any of the specified codes.
+func IsBroadcastError(err error, codes ...BroadcastErrorCode) bool {
+	broadcastErr, ok := err.(*BroadcastError)
+	if !ok {
+		return false
+	}
+
+	for _, code := range codes {
+		if broadcastErr.Code == code {
+			return true
+		}
+	}
+
+	return false
+}
+
+// ParseBroadcastError maps a peer's reject message for a transaction to a
+// BroadcastError.
+func ParseBroadcastError(msg *wire.MsgReject, peerAddr string) *BroadcastError {
+	// We'll determine the appropriate broadcast error code by looking at
+	// the reject's message code and reason. The only reject codes returned
+	// from peers (bitcoind and btcd) when attempting to accept a
+	// transaction into their mempool are:
+	//   RejectInvalid, RejectNonstandard, RejectInsufficientFee,
+	//   RejectDuplicate
+	var code BroadcastErrorCode
+	switch {
+	// The cases below apply for reject messages sent from any kind of peer.
+	case msg.Code == wire.RejectInvalid || msg.Code == wire.RejectNonstandard:
+		code = Invalid
+
+	case msg.Code == wire.RejectInsufficientFee:
+		code = InsufficientFee
+
+	// The cases below apply for reject messages sent from bitcoind peers.
+	//
+	// If the transaction double spends an unconfirmed transaction in the
+	// peer's mempool, then we'll deem it as invalid.
+	case msg.Code == wire.RejectDuplicate &&
+		strings.Contains(msg.Reason, "txn-mempool-conflict"):
+		code = Invalid
+
+	// If the transaction was rejected due to it already existing in the
+	// peer's mempool, then return an error signaling so.
+	case msg.Code == wire.RejectDuplicate &&
+		strings.Contains(msg.Reason, "txn-already-in-mempool"):
+		code = Mempool
+
+	// If the transaction was rejected due to it already existing in the
+	// chain according to our peer, then we'll return an error signaling so.
+	case msg.Code == wire.RejectDuplicate &&
+		strings.Contains(msg.Reason, "txn-already-known"):
+		code = Confirmed
+
+	// The cases below apply for reject messages sent from btcd peers.
+	//
+	// If the transaction double spends an unconfirmed transaction in the
+	// peer's mempool, then we'll deem it as invalid.
+	case msg.Code == wire.RejectDuplicate &&
+		strings.Contains(msg.Reason, "already spent"):
+		code = Invalid
+
+	// If the transaction was rejected due to it already existing in the
+	// peer's mempool, then return an error signaling so.
+	case msg.Code == wire.RejectDuplicate &&
+		strings.Contains(msg.Reason, "already have transaction"):
+		code = Mempool
+
+	// If the transaction was rejected due to it already existing in the
+	// chain according to our peer, then we'll return an error signaling so.
+	case msg.Code == wire.RejectDuplicate &&
+		strings.Contains(msg.Reason, "transaction already exists"):
+		code = Confirmed
+
+	// Any other reject messages will use the unknown code.
+	default:
+		code = Unknown
+	}
+
+	reason := fmt.Sprintf("rejected by %v: %v", peerAddr, msg.Reason)
+	return &BroadcastError{Code: code, Reason: reason}
+}

vendor/github.com/lightninglabs/neutrino/pushtx/log.go

@@ -0,0 +1,43 @@
+package pushtx
+
+import "github.com/btcsuite/btclog"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	DisableLog()
+}
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until either UseLogger or SetLogWriter are called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}
+
+// LogClosure is a closure that can be printed with %v to be used to
+// generate expensive-to-create data for a detailed log level and avoid doing
+// the work if the data isn't printed.
+type logClosure func() string
+
+// String invokes the log closure and returns the results string.
+func (c logClosure) String() string {
+	return c()
+}
+
+// newLogClosure returns a new closure over the passed function which allows
+// it to be used as a parameter in a logging function that is only invoked when
+// the logging level is such that the message will actually be logged.
+func newLogClosure(c func() string) logClosure {
+	return logClosure(c)
+}