lnwallet: add new rebroadcaster interface, use for background tx publish

In this commit, we add a new Rebroadcaster interface to be used for
publishing transactions passively in the background until they've been
confirmed on chain. This is useful if a tx drops out of the mempool, but
then the pool clears down and has more space available to accept the tx
at the current fee level.

lnwallet/config.go

@@ -56,4 +56,9 @@ type Config struct {
 	// NetParams is the set of parameters that tells the wallet which chain
 	// it will be operating on.
 	NetParams chaincfg.Params
+
+	// Rebroadcaster is an optional config param that can be used to
+	// passively rebroadcast transactions in the background until they're
+	// detected as being confirmed.
+	Rebroadcaster Rebroadcaster
 }

lnwallet/rebroadcaster.go

@@ -0,0 +1,27 @@
+package lnwallet
+
+import (
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/wire"
+)
+
+// Rebroadcaster is an abstract rebroadcaster instance that'll continually
+// rebroadcast transactions in the background until they're confirmed.
+type Rebroadcaster interface {
+	// Start launches all goroutines the rebroadcaster needs to operate.
+	Start() error
+
+	// Started returns true if the broadcaster is already active.
+	Started() bool
+
+	// Stop terminates the rebroadcaster and all goroutines it spawned.
+	Stop()
+
+	// Broadcast enqueues a transaction to be rebroadcast until it's been
+	// confirmed.
+	Broadcast(tx *wire.MsgTx) error
+
+	// MarkAsConfirmed marks a transaction as confirmed, so it won't be
+	// rebroadcast.
+	MarkAsConfirmed(txid chainhash.Hash)
+}

lnwallet/wallet.go

@@ -407,6 +407,15 @@ func (l *LightningWallet) Startup() error {
 		return err
 	}
 
+	if l.Cfg.Rebroadcaster != nil {
+		go func() {
+			if err := l.Cfg.Rebroadcaster.Start(); err != nil {
+				walletLog.Errorf("unable to start "+
+					"rebroadcaster: %v", err)
+			}
+		}()
+	}
+
 	l.wg.Add(1)
 	// TODO(roasbeef): multiple request handlers?
 	go l.requestHandler()
@@ -426,11 +435,74 @@ func (l *LightningWallet) Shutdown() error {
 		return err
 	}
 
+	if l.Cfg.Rebroadcaster != nil && l.Cfg.Rebroadcaster.Started() {
+		l.Cfg.Rebroadcaster.Stop()
+	}
+
 	close(l.quit)
 	l.wg.Wait()
 	return nil
 }
 
+// PublishTransaction wraps the wallet controller tx publish method with an
+// extra rebroadcaster layer if the sub-system is configured.
+func (l *LightningWallet) PublishTransaction(tx *wire.MsgTx,
+	label string) error {
+
+	sendTxToWallet := func() error {
+		return l.WalletController.PublishTransaction(tx, label)
+	}
+
+	// If we don't have rebroadcaster then we can exit early (and send only
+	// to the wallet).
+	if l.Cfg.Rebroadcaster == nil || !l.Cfg.Rebroadcaster.Started() {
+		return sendTxToWallet()
+	}
+
+	// We pass this into the rebroadcaster first, so the initial attempt
+	// will succeed if the transaction isn't yet in the mempool. However we
+	// ignore the error here as this might be resent on start up and the
+	// transaction already exists.
+	_ = l.Cfg.Rebroadcaster.Broadcast(tx)
+
+	// Then we pass things into the wallet as normal, which'll add the
+	// transaction label on disk.
+	if err := sendTxToWallet(); err != nil {
+		return err
+	}
+
+	// TODO(roasbeef): want diff height actually? no context though
+	_, bestHeight, err := l.Cfg.ChainIO.GetBestBlock()
+	if err != nil {
+		return err
+	}
+
+	txHash := tx.TxHash()
+	go func() {
+		const numConfs = 6
+
+		txConf, err := l.Cfg.Notifier.RegisterConfirmationsNtfn(
+			&txHash, tx.TxOut[0].PkScript, numConfs, uint32(bestHeight),
+		)
+		if err != nil {
+			return
+		}
+
+		select {
+		case <-txConf.Confirmed:
+			// TODO(roasbeef): also want to remove from
+			// rebroadcaster if conflict happens...deeper wallet
+			// integration?
+			l.Cfg.Rebroadcaster.MarkAsConfirmed(tx.TxHash())
+
+		case <-l.quit:
+			return
+		}
+	}()
+
+	return nil
+}
+
 // ConfirmedBalance returns the current confirmed balance of a wallet account.
 // This methods wraps the internal WalletController method so we're able to
 // properly hold the coin select mutex while we compute the balance.