Merge pull request #814 from cfromknecht/rebcast-funding-txn

Rebroadcast Funding Txn

breacharbiter_test.go

@@ -1412,6 +1412,7 @@ func createInitChannels(revocationWindow int) (*lnwallet.LightningChannel, *lnwa
 		RemoteCommitment:        aliceCommit,
 		Db:                      dbAlice,
 		Packager:                channeldb.NewChannelPackager(shortChanID),
+		FundingTxn:              testTx,
 	}
 	bobChannelState := &channeldb.OpenChannel{
 		LocalChanCfg:            bobCfg,

channeldb/channel.go

@@ -403,6 +403,14 @@ type OpenChannel struct {
 	// failures and reforward HTLCs that were not fully processed.
 	Packager FwdPackager
 
+	// FundingTxn is the transaction containing this channel's funding
+	// outpoint. Upon restarts, this txn will be rebroadcast if the channel
+	// is found to be pending.
+	//
+	// NOTE: This value will only be populated for single-funder channels
+	// for which we are the initiator.
+	FundingTxn *wire.MsgTx
+
 	// TODO(roasbeef): eww
 	Db *DB
 
@@ -1797,6 +1805,13 @@ func putChanInfo(chanBucket *bolt.Bucket, channel *OpenChannel) error {
 		return err
 	}
 
+	// For single funder channels that we initiated, write the funding txn.
+	if channel.ChanType == SingleFunder && channel.IsInitiator {
+		if err := writeElement(&w, channel.FundingTxn); err != nil {
+			return err
+		}
+	}
+
 	writeChanConfig := func(b io.Writer, c *ChannelConfig) error {
 		return writeElements(b,
 			c.DustLimit, c.MaxPendingAmount, c.ChanReserve, c.MinHTLC,
@@ -1898,6 +1913,13 @@ func fetchChanInfo(chanBucket *bolt.Bucket, channel *OpenChannel) error {
 		return err
 	}
 
+	// For single funder channels that we initiated, read the funding txn.
+	if channel.ChanType == SingleFunder && channel.IsInitiator {
+		if err := readElement(r, &channel.FundingTxn); err != nil {
+			return err
+		}
+	}
+
 	readChanConfig := func(b io.Reader, c *ChannelConfig) error {
 		return readElements(b,
 			&c.DustLimit, &c.MaxPendingAmount, &c.ChanReserve,

channeldb/channel_test.go

@@ -220,6 +220,7 @@ func createTestChannelState(cdb *DB) (*OpenChannel, error) {
 		RevocationStore:         store,
 		Db:                      cdb,
 		Packager:                NewChannelPackager(chanID),
+		FundingTxn:              testTx,
 	}, nil
 }
 

fundingmanager.go

@@ -161,6 +161,10 @@ type fundingConfig struct {
 	// funds from on-chain transaction outputs into Lightning channels.
 	Wallet *lnwallet.LightningWallet
 
+	// PublishTransaction facilitates the process of broadcasting a
+	// transaction to the network.
+	PublishTransaction func(*wire.MsgTx) error
+
 	// FeeEstimator calculates appropriate fee rates based on historical
 	// transaction information.
 	FeeEstimator lnwallet.FeeEstimator
@@ -422,6 +426,22 @@ func (f *fundingManager) Start() error {
 
 		f.localDiscoverySignals[chanID] = make(chan struct{})
 
+		// Rebroadcast the funding transaction for any pending channel
+		// that we initiated. If this operation fails due to a reported
+		// double spend, we treat this as an indicator that we have
+		// already broadcast this transaction. Otherwise, we simply log
+		// the error as there isn't anything we can currently do to
+		// recover.
+		if channel.ChanType == channeldb.SingleFunder &&
+			channel.IsInitiator {
+
+			err := f.cfg.PublishTransaction(channel.FundingTxn)
+			if err != nil && err != lnwallet.ErrDoubleSpend {
+				fndgLog.Warnf("unable to rebroadcast funding "+
+					"txn: %v", err)
+			}
+		}
+
 		confChan := make(chan *lnwire.ShortChannelID)
 		timeoutChan := make(chan struct{})
 

fundingmanager_test.go

@@ -337,6 +337,7 @@ func recreateAliceFundingManager(t *testing.T, alice *testNode) {
 	aliceMsgChan := make(chan lnwire.Message)
 	aliceAnnounceChan := make(chan lnwire.Message)
 	shutdownChan := make(chan struct{})
+	publishChan := make(chan *wire.MsgTx, 10)
 
 	oldCfg := alice.fundingMgr.cfg
 
@@ -376,6 +377,10 @@ func recreateAliceFundingManager(t *testing.T, alice *testNode) {
 		TempChanIDSeed: oldCfg.TempChanIDSeed,
 		ArbiterChan:    alice.arbiterChan,
 		FindChannel:    oldCfg.FindChannel,
+		PublishTransaction: func(txn *wire.MsgTx) error {
+			publishChan <- txn
+			return nil
+		},
 	})
 	if err != nil {
 		t.Fatalf("failed recreating aliceFundingManager: %v", err)
@@ -384,6 +389,7 @@ func recreateAliceFundingManager(t *testing.T, alice *testNode) {
 	alice.fundingMgr = f
 	alice.msgChan = aliceMsgChan
 	alice.announceChan = aliceAnnounceChan
+	alice.publTxChan = publishChan
 	alice.shutdownChannel = shutdownChan
 
 	if err = f.Start(); err != nil {
@@ -1228,6 +1234,13 @@ func TestFundingManagerFundingNotTimeoutInitiator(t *testing.T) {
 
 	recreateAliceFundingManager(t, alice)
 
+	// We should receive the rebroadcasted funding txn.
+	select {
+	case <-alice.publTxChan:
+	case <-time.After(time.Second * 5):
+		t.Fatalf("alice did not publish funding tx")
+	}
+
 	// Increase the height to 1 minus the maxWaitNumBlocksFundingConf height
 	alice.mockNotifier.epochChan <- &chainntnfs.BlockEpoch{
 		Height: fundingBroadcastHeight + maxWaitNumBlocksFundingConf - 1,

htlcswitch/test_utils.go

@@ -54,6 +54,40 @@ var (
 		"5445068131219452686511677818569431", 10)
 	_, _ = testSig.S.SetString("1880105606924982582529128710493133386286603"+
 		"3135609736119018462340006816851118", 10)
+
+	// testTx is used as the default funding txn for single-funder channels.
+	testTx = &wire.MsgTx{
+		Version: 1,
+		TxIn: []*wire.TxIn{
+			{
+				PreviousOutPoint: wire.OutPoint{
+					Hash:  chainhash.Hash{},
+					Index: 0xffffffff,
+				},
+				SignatureScript: []byte{0x04, 0x31, 0xdc, 0x00, 0x1b, 0x01, 0x62},
+				Sequence:        0xffffffff,
+			},
+		},
+		TxOut: []*wire.TxOut{
+			{
+				Value: 5000000000,
+				PkScript: []byte{
+					0x41, // OP_DATA_65
+					0x04, 0xd6, 0x4b, 0xdf, 0xd0, 0x9e, 0xb1, 0xc5,
+					0xfe, 0x29, 0x5a, 0xbd, 0xeb, 0x1d, 0xca, 0x42,
+					0x81, 0xbe, 0x98, 0x8e, 0x2d, 0xa0, 0xb6, 0xc1,
+					0xc6, 0xa5, 0x9d, 0xc2, 0x26, 0xc2, 0x86, 0x24,
+					0xe1, 0x81, 0x75, 0xe8, 0x51, 0xc9, 0x6b, 0x97,
+					0x3d, 0x81, 0xb0, 0x1c, 0xc3, 0x1f, 0x04, 0x78,
+					0x34, 0xbc, 0x06, 0xd6, 0xd6, 0xed, 0xf6, 0x20,
+					0xd1, 0x84, 0x24, 0x1a, 0x6a, 0xed, 0x8b, 0x63,
+					0xa6, // 65-byte signature
+					0xac, // OP_CHECKSIG
+				},
+			},
+		},
+		LockTime: 5,
+	}
 )
 
 var idSeqNum uint64
@@ -294,6 +328,7 @@ func createTestChannel(alicePrivKey, bobPrivKey []byte,
 		ShortChanID:             chanID,
 		Db:                      dbAlice,
 		Packager:                channeldb.NewChannelPackager(chanID),
+		FundingTxn:              testTx,
 	}
 
 	bobChannelState := &channeldb.OpenChannel{

lnd.go

@@ -275,10 +275,11 @@ func lndMain() error {
 		return err
 	}
 	fundingMgr, err := newFundingManager(fundingConfig{
-		IDKey:        idPrivKey.PubKey(),
-		Wallet:       activeChainControl.wallet,
-		Notifier:     activeChainControl.chainNotifier,
-		FeeEstimator: activeChainControl.feeEstimator,
+		IDKey:              idPrivKey.PubKey(),
+		Wallet:             activeChainControl.wallet,
+		PublishTransaction: activeChainControl.wallet.PublishTransaction,
+		Notifier:           activeChainControl.chainNotifier,
+		FeeEstimator:       activeChainControl.feeEstimator,
 		SignMessage: func(pubKey *btcec.PublicKey,
 			msg []byte) (*btcec.Signature, error) {
 

lnwallet/channel_test.go

@@ -57,6 +57,40 @@ var (
 	// The number of confirmations required to consider any created channel
 	// open.
 	numReqConfs = uint16(1)
+
+	// A serializable txn for testing funding txn.
+	testTx = &wire.MsgTx{
+		Version: 1,
+		TxIn: []*wire.TxIn{
+			{
+				PreviousOutPoint: wire.OutPoint{
+					Hash:  chainhash.Hash{},
+					Index: 0xffffffff,
+				},
+				SignatureScript: []byte{0x04, 0x31, 0xdc, 0x00, 0x1b, 0x01, 0x62},
+				Sequence:        0xffffffff,
+			},
+		},
+		TxOut: []*wire.TxOut{
+			{
+				Value: 5000000000,
+				PkScript: []byte{
+					0x41, // OP_DATA_65
+					0x04, 0xd6, 0x4b, 0xdf, 0xd0, 0x9e, 0xb1, 0xc5,
+					0xfe, 0x29, 0x5a, 0xbd, 0xeb, 0x1d, 0xca, 0x42,
+					0x81, 0xbe, 0x98, 0x8e, 0x2d, 0xa0, 0xb6, 0xc1,
+					0xc6, 0xa5, 0x9d, 0xc2, 0x26, 0xc2, 0x86, 0x24,
+					0xe1, 0x81, 0x75, 0xe8, 0x51, 0xc9, 0x6b, 0x97,
+					0x3d, 0x81, 0xb0, 0x1c, 0xc3, 0x1f, 0x04, 0x78,
+					0x34, 0xbc, 0x06, 0xd6, 0xd6, 0xed, 0xf6, 0x20,
+					0xd1, 0x84, 0x24, 0x1a, 0x6a, 0xed, 0x8b, 0x63,
+					0xa6, // 65-byte signature
+					0xac, // OP_CHECKSIG
+				},
+			},
+		},
+		LockTime: 5,
+	}
 )
 
 // initRevocationWindows simulates a new channel being opened within the p2p
@@ -308,6 +342,7 @@ func createTestChannels(revocationWindow int) (*LightningChannel,
 		RemoteCommitment:        aliceCommit,
 		Db:                      dbAlice,
 		Packager:                channeldb.NewChannelPackager(shortChanID),
+		FundingTxn:              testTx,
 	}
 	bobChannelState := &channeldb.OpenChannel{
 		LocalChanCfg:            bobCfg,

lnwallet/wallet.go

@@ -1073,10 +1073,12 @@ func (l *LightningWallet) handleFundingCounterPartySigs(msg *addCounterPartySigs
 	res.partialState.LocalChanCfg = res.ourContribution.toChanConfig()
 	res.partialState.RemoteChanCfg = res.theirContribution.toChanConfig()
 
+	// We'll also record the finalized funding txn, which will allow us to
+	// rebroadcast on startup in case we fail.
+	res.partialState.FundingTxn = fundingTx
+
 	// Add the complete funding transaction to the DB, in its open bucket
 	// which will be used for the lifetime of this channel.
-	// TODO(roasbeef):
-	//  * attempt to retransmit funding transactions on re-start
 	nodeAddr := res.nodeAddr
 	err = res.partialState.SyncPending(nodeAddr, uint32(bestHeight))
 	if err != nil {

test_utils.go

@@ -49,6 +49,40 @@ var (
 
 	// Just use some arbitrary bytes as delivery script.
 	dummyDeliveryScript = alicesPrivKey[:]
+
+	// testTx is used as the default funding txn for single-funder channels.
+	testTx = &wire.MsgTx{
+		Version: 1,
+		TxIn: []*wire.TxIn{
+			{
+				PreviousOutPoint: wire.OutPoint{
+					Hash:  chainhash.Hash{},
+					Index: 0xffffffff,
+				},
+				SignatureScript: []byte{0x04, 0x31, 0xdc, 0x00, 0x1b, 0x01, 0x62},
+				Sequence:        0xffffffff,
+			},
+		},
+		TxOut: []*wire.TxOut{
+			{
+				Value: 5000000000,
+				PkScript: []byte{
+					0x41, // OP_DATA_65
+					0x04, 0xd6, 0x4b, 0xdf, 0xd0, 0x9e, 0xb1, 0xc5,
+					0xfe, 0x29, 0x5a, 0xbd, 0xeb, 0x1d, 0xca, 0x42,
+					0x81, 0xbe, 0x98, 0x8e, 0x2d, 0xa0, 0xb6, 0xc1,
+					0xc6, 0xa5, 0x9d, 0xc2, 0x26, 0xc2, 0x86, 0x24,
+					0xe1, 0x81, 0x75, 0xe8, 0x51, 0xc9, 0x6b, 0x97,
+					0x3d, 0x81, 0xb0, 0x1c, 0xc3, 0x1f, 0x04, 0x78,
+					0x34, 0xbc, 0x06, 0xd6, 0xd6, 0xed, 0xf6, 0x20,
+					0xd1, 0x84, 0x24, 0x1a, 0x6a, 0xed, 0x8b, 0x63,
+					0xa6, // 65-byte signature
+					0xac, // OP_CHECKSIG
+				},
+			},
+		},
+		LockTime: 5,
+	}
 )
 
 // createTestPeer creates a channel between two nodes, and returns a peer for
@@ -219,6 +253,7 @@ func createTestPeer(notifier chainntnfs.ChainNotifier,
 		RemoteCommitment:        aliceCommit,
 		Db:                      dbAlice,
 		Packager:                channeldb.NewChannelPackager(shortChanID),
+		FundingTxn:              testTx,
 	}
 	bobChannelState := &channeldb.OpenChannel{
 		LocalChanCfg:            bobCfg,