Merge pull request #1635 from halseth/funding-broadcast-fail

Move funding tx broadcasting to Fundingmanager

fundingmanager.go

@@ -961,23 +961,42 @@ func (f *fundingManager) handleFundingOpen(fmsg *fundingOpenMsg) {
 	// Check number of pending channels to be smaller than maximum allowed
 	// number and send ErrorGeneric to remote peer if condition is
 	// violated.
-	peerIDKey := newSerializedKey(fmsg.peer.IdentityKey())
+	peerPubKey := fmsg.peer.IdentityKey()
+	peerIDKey := newSerializedKey(peerPubKey)
 
 	msg := fmsg.msg
 	amt := msg.FundingAmount
 
+	// We count the number of pending channels for this peer. This is the
+	// sum of the active reservations and the channels pending open in the
+	// database.
+	f.resMtx.RLock()
+	numPending := len(f.activeReservations[peerIDKey])
+	f.resMtx.RUnlock()
+
+	channels, err := f.cfg.Wallet.Cfg.Database.FetchOpenChannels(peerPubKey)
+	if err != nil {
+		f.failFundingFlow(
+			fmsg.peer, fmsg.msg.PendingChannelID, err,
+		)
+		return
+	}
+
+	for _, c := range channels {
+		if c.IsPending {
+			numPending++
+		}
+	}
+
 	// TODO(roasbeef): modify to only accept a _single_ pending channel per
 	// block unless white listed
-	f.resMtx.RLock()
+	if numPending >= cfg.MaxPendingChannels {
-	if len(f.activeReservations[peerIDKey]) >= cfg.MaxPendingChannels {
-		f.resMtx.RUnlock()
 		f.failFundingFlow(
 			fmsg.peer, fmsg.msg.PendingChannelID,
 			lnwire.ErrMaxPendingChannels,
 		)
 		return
 	}
-	f.resMtx.RUnlock()
 
 	// We'll also reject any requests to create channels until we're fully
 	// synced to the network as we won't be able to properly validate the
@@ -1351,6 +1370,10 @@ func (f *fundingManager) handleFundingCreated(fmsg *fundingCreatedMsg) {
 		return
 	}
 
+	// The channel is marked IsPending in the database, and can be removed
+	// from the set of active reservations.
+	f.deleteReservationCtx(peerKey, fmsg.msg.PendingChannelID)
+
 	// If something goes wrong before the funding transaction is confirmed,
 	// we use this convenience method to delete the pending OpenChannel
 	// from the database.
@@ -1420,11 +1443,6 @@ func (f *fundingManager) handleFundingCreated(fmsg *fundingCreatedMsg) {
 
 	// At this point we have sent our last funding message to the
 	// initiating peer before the funding transaction will be broadcast.
-	// The only thing left to do before we can delete this reservation
-	// is wait for the funding transaction. Lock the reservation so it
-	// is not pruned by the zombie sweeper.
-	resCtx.lock()
-
 	// With this last message, our job as the responder is now complete.
 	// We'll wait for the funding transaction to reach the specified number
 	// of confirmations, then start normal operations.
@@ -1473,8 +1491,6 @@ func (f *fundingManager) handleFundingCreated(fmsg *fundingCreatedMsg) {
 		}
 
 		// Success, funding transaction was confirmed.
-		f.deleteReservationCtx(peerKey, fmsg.msg.PendingChannelID)
-
 		err := f.handleFundingConfirmation(
 			fmsg.peer, completeChan, shortChanID,
 		)
@@ -1542,13 +1558,36 @@ func (f *fundingManager) handleFundingSigned(fmsg *fundingSignedMsg) {
 	// transaction. We'll verify the signature for validity, then commit
 	// the state to disk as we can now open the channel.
 	commitSig := fmsg.msg.CommitSig.ToSignatureBytes()
-	completeChan, err := resCtx.reservation.CompleteReservation(nil, commitSig)
+	completeChan, err := resCtx.reservation.CompleteReservation(
+		nil, commitSig,
+	)
 	if err != nil {
-		fndgLog.Errorf("Unable to complete reservation sign complete: %v", err)
+		fndgLog.Errorf("Unable to complete reservation sign "+
+			"complete: %v", err)
 		f.failFundingFlow(fmsg.peer, pendingChanID, err)
 		return
 	}
 
+	// The channel is now marked IsPending in the database, and we can
+	// delete it from our set of active reservations.
+	f.deleteReservationCtx(peerKey, pendingChanID)
+
+	// Broadcast the finalized funding transaction to the network.
+	fundingTx := completeChan.FundingTxn
+	fndgLog.Infof("Broadcasting funding tx for ChannelPoint(%v): %v",
+		completeChan.FundingOutpoint, spew.Sdump(fundingTx))
+
+	err = f.cfg.PublishTransaction(fundingTx)
+	if err != nil {
+		fndgLog.Errorf("unable to broadcast funding "+
+			"txn: %v", err)
+		// We failed to broadcast the funding transaction, but watch
+		// the channel regardless, in case the transaction made it to
+		// the network. We will retry broadcast at startup.
+		// TODO(halseth): retry more often? Handle with CPFP? Just
+		// delete from the DB?
+	}
+
 	// Now that we have a finalized reservation for this funding flow,
 	// we'll send the to be active channel to the ChainArbitrator so it can
 	// watch for any on-chin actions before the channel has fully
@@ -1576,11 +1615,7 @@ func (f *fundingManager) handleFundingSigned(fmsg *fundingSignedMsg) {
 	}
 
 	// At this point we have broadcast the funding transaction and done all
-	// necessary processing. The only thing left to do before we can delete
+	// necessary processing.
-	// this reservation is wait for the funding transaction. Lock the
-	// reservation so it is not pruned by the zombie sweeper.
-	resCtx.lock()
-
 	f.wg.Add(1)
 	go func() {
 		defer f.wg.Done()
@@ -1659,8 +1694,6 @@ func (f *fundingManager) handleFundingSigned(fmsg *fundingSignedMsg) {
 			},
 		}
 
-		f.deleteReservationCtx(peerKey, pendingChanID)
-
 		err = f.annAfterSixConfs(completeChan, shortChanID)
 		if err != nil {
 			fndgLog.Errorf("failed sending channel announcement: %v",

fundingmanager_test.go

@@ -10,6 +10,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"runtime"
 	"testing"
 	"time"
 
@@ -249,8 +250,7 @@ func createTestFundingManager(t *testing.T, privKey *btcec.PrivateKey,
 	shutdownChan := make(chan struct{})
 
 	wc := &mockWalletController{
-		rootKey:               alicePrivKey,
+		rootKey: alicePrivKey,
-		publishedTransactions: publTxChan,
 	}
 	signer := &mockSigner{
 		key: alicePrivKey,
@@ -350,6 +350,10 @@ func createTestFundingManager(t *testing.T, privKey *btcec.PrivateKey,
 		ReportShortChanID: func(wire.OutPoint) error {
 			return nil
 		},
+		PublishTransaction: func(txn *wire.MsgTx) error {
+			publTxChan <- txn
+			return nil
+		},
 		ZombieSweeperInterval: 1 * time.Hour,
 		ReservationTimeout:    1 * time.Nanosecond,
 	})
@@ -445,11 +449,11 @@ func recreateAliceFundingManager(t *testing.T, alice *testNode) {
 	}
 }
 
-func setupFundingManagers(t *testing.T) (*testNode, *testNode) {
+func setupFundingManagers(t *testing.T, maxPendingChannels int) (*testNode, *testNode) {
 	// We need to set the global config, as fundingManager uses
 	// MaxPendingChannels, and it is usually set in lndMain().
 	cfg = &config{
-		MaxPendingChannels: defaultMaxPendingChannels,
+		MaxPendingChannels: maxPendingChannels,
 	}
 
 	aliceTestDir, err := ioutil.TempDir("", "alicelnwallet")
@@ -566,6 +570,11 @@ func openChannel(t *testing.T, alice, bob *testNode, localFundingAmt,
 		t, bob.msgChan, "AcceptChannel",
 	).(*lnwire.AcceptChannel)
 
+	// They now should both have pending reservations for this channel
+	// active.
+	assertNumPendingReservations(t, alice, bobPubKey, 1)
+	assertNumPendingReservations(t, bob, alicePubKey, 1)
+
 	// Forward the response to Alice.
 	alice.fundingMgr.processFundingAccept(acceptChannelResponse, bob)
 
@@ -612,6 +621,12 @@ func openChannel(t *testing.T, alice, bob *testNode, localFundingAmt,
 		Hash:  publ.TxHash(),
 		Index: 0,
 	}
+
+	// Finally, make sure neither have active reservation for the channel
+	// now pending open in the database.
+	assertNumPendingReservations(t, alice, bobPubKey, 0)
+	assertNumPendingReservations(t, bob, alicePubKey, 0)
+
 	return fundingOutPoint
 }
 
@@ -660,6 +675,8 @@ func assertFundingMsgSent(t *testing.T, msgChan chan lnwire.Message,
 		sentMsg, ok = msg.(*lnwire.FundingSigned)
 	case "FundingLocked":
 		sentMsg, ok = msg.(*lnwire.FundingLocked)
+	case "Error":
+		sentMsg, ok = msg.(*lnwire.Error)
 	default:
 		t.Fatalf("unknown message type: %s", msgType)
 	}
@@ -670,8 +687,10 @@ func assertFundingMsgSent(t *testing.T, msgChan chan lnwire.Message,
 			t.Fatalf("expected %s to be sent, instead got error: %v",
 				msgType, lnwire.ErrorCode(errorMsg.Data[0]))
 		}
-		t.Fatalf("expected %s to be sent, instead got %T",
+
-			msgType, msg)
+		_, _, line, _ := runtime.Caller(1)
+		t.Fatalf("expected %s to be sent, instead got %T at %v",
+			msgType, msg, line)
 	}
 
 	return sentMsg
@@ -937,7 +956,7 @@ func assertHandleFundingLocked(t *testing.T, alice, bob *testNode) {
 }
 
 func TestFundingManagerNormalWorkflow(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -948,19 +967,10 @@ func TestFundingManagerNormalWorkflow(t *testing.T) {
 	fundingOutPoint := openChannel(t, alice, bob, 500000, 0, 1, updateChan,
 		true)
 
-	// Make sure both reservations time out and then run both zombie sweepers.
-	time.Sleep(1 * time.Millisecond)
-	go alice.fundingMgr.pruneZombieReservations()
-	go bob.fundingMgr.pruneZombieReservations()
-
 	// Check that neither Alice nor Bob sent an error message.
 	assertErrorNotSent(t, alice.msgChan)
 	assertErrorNotSent(t, bob.msgChan)
 
-	// Check that neither reservation has been pruned.
-	assertNumPendingReservations(t, alice, bobPubKey, 1)
-	assertNumPendingReservations(t, bob, alicePubKey, 1)
-
 	// Notify that transaction was mined.
 	alice.mockNotifier.oneConfChannel <- &chainntnfs.TxConfirmation{}
 	bob.mockNotifier.oneConfChannel <- &chainntnfs.TxConfirmation{}
@@ -1016,7 +1026,7 @@ func TestFundingManagerNormalWorkflow(t *testing.T) {
 }
 
 func TestFundingManagerRestartBehavior(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// Run through the process of opening the channel, up until the funding
@@ -1147,7 +1157,7 @@ func TestFundingManagerRestartBehavior(t *testing.T) {
 // server to notify when the peer comes online, in case sending the
 // fundingLocked message fails the first time.
 func TestFundingManagerOfflinePeer(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// Run through the process of opening the channel, up until the funding
@@ -1278,7 +1288,7 @@ func TestFundingManagerOfflinePeer(t *testing.T) {
 // will properly clean up a zombie reservation that times out after the
 // initFundingMsg has been handled.
 func TestFundingManagerPeerTimeoutAfterInitFunding(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1338,7 +1348,7 @@ func TestFundingManagerPeerTimeoutAfterInitFunding(t *testing.T) {
 // will properly clean up a zombie reservation that times out after the
 // fundingOpenMsg has been handled.
 func TestFundingManagerPeerTimeoutAfterFundingOpen(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1407,7 +1417,7 @@ func TestFundingManagerPeerTimeoutAfterFundingOpen(t *testing.T) {
 // will properly clean up a zombie reservation that times out after the
 // fundingAcceptMsg has been handled.
 func TestFundingManagerPeerTimeoutAfterFundingAccept(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1481,7 +1491,7 @@ func TestFundingManagerPeerTimeoutAfterFundingAccept(t *testing.T) {
 }
 
 func TestFundingManagerFundingTimeout(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1526,7 +1536,7 @@ func TestFundingManagerFundingTimeout(t *testing.T) {
 // the channel initiator, that it does not timeout when the lnd restarts.
 func TestFundingManagerFundingNotTimeoutInitiator(t *testing.T) {
 
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1593,7 +1603,7 @@ func TestFundingManagerFundingNotTimeoutInitiator(t *testing.T) {
 // continues to operate as expected in case we receive a duplicate fundingLocked
 // message.
 func TestFundingManagerReceiveFundingLockedTwice(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1682,7 +1692,7 @@ func TestFundingManagerReceiveFundingLockedTwice(t *testing.T) {
 // handles receiving a fundingLocked after the its own fundingLocked and channel
 // announcement is sent and gets restarted.
 func TestFundingManagerRestartAfterChanAnn(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1756,7 +1766,7 @@ func TestFundingManagerRestartAfterChanAnn(t *testing.T) {
 // fundingManager continues to operate as expected after it has received
 // fundingLocked and then gets restarted.
 func TestFundingManagerRestartAfterReceivingFundingLocked(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1826,7 +1836,7 @@ func TestFundingManagerRestartAfterReceivingFundingLocked(t *testing.T) {
 // (a channel not supposed to be announced to the rest of the network),
 // the announcementSignatures nor the nodeAnnouncement messages are sent.
 func TestFundingManagerPrivateChannel(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1906,7 +1916,7 @@ func TestFundingManagerPrivateChannel(t *testing.T) {
 // announcement signatures nor the node announcement messages are sent upon
 // restart.
 func TestFundingManagerPrivateRestart(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// We will consume the channel updates as we go, so no buffering is needed.
@@ -1996,7 +2006,7 @@ func TestFundingManagerPrivateRestart(t *testing.T) {
 // TestFundingManagerCustomChannelParameters checks that custom requirements we
 // specify during the channel funding flow is preserved correcly on both sides.
 func TestFundingManagerCustomChannelParameters(t *testing.T) {
-	alice, bob := setupFundingManagers(t)
+	alice, bob := setupFundingManagers(t, defaultMaxPendingChannels)
 	defer tearDownFundingManagers(t, alice, bob)
 
 	// This is the custom parameters we'll use.
@@ -2088,39 +2098,6 @@ func TestFundingManagerCustomChannelParameters(t *testing.T) {
 		t, alice.msgChan, "FundingCreated",
 	).(*lnwire.FundingCreated)
 
-	// Give the message to Bob.
-	bob.fundingMgr.processFundingCreated(fundingCreated, alice)
-
-	// Finally, Bob should send the FundingSigned message.
-	fundingSigned := assertFundingMsgSent(
-		t, bob.msgChan, "FundingSigned",
-	).(*lnwire.FundingSigned)
-
-	// Forward the signature to Alice.
-	alice.fundingMgr.processFundingSigned(fundingSigned, bob)
-
-	// After Alice processes the singleFundingSignComplete message, she will
-	// broadcast the funding transaction to the network. We expect to get a
-	// channel update saying the channel is pending.
-	var pendingUpdate *lnrpc.OpenStatusUpdate
-	select {
-	case pendingUpdate = <-updateChan:
-	case <-time.After(time.Second * 5):
-		t.Fatalf("alice did not send OpenStatusUpdate_ChanPending")
-	}
-
-	_, ok = pendingUpdate.Update.(*lnrpc.OpenStatusUpdate_ChanPending)
-	if !ok {
-		t.Fatal("OpenStatusUpdate was not OpenStatusUpdate_ChanPending")
-	}
-
-	// Wait for Alice to published the funding tx to the network.
-	select {
-	case <-alice.publTxChan:
-	case <-time.After(time.Second * 5):
-		t.Fatalf("alice did not publish funding tx")
-	}
-
 	// Helper method for checking the CSV delay stored for a reservation.
 	assertDelay := func(resCtx *reservationWithCtx,
 		ourDelay, theirDelay uint16) error {
@@ -2190,4 +2167,198 @@ func TestFundingManagerCustomChannelParameters(t *testing.T) {
 	if err := assertMinHtlc(resCtx, minHtlc, 5); err != nil {
 		t.Fatal(err)
 	}
+
+	// Give the message to Bob.
+	bob.fundingMgr.processFundingCreated(fundingCreated, alice)
+
+	// Finally, Bob should send the FundingSigned message.
+	fundingSigned := assertFundingMsgSent(
+		t, bob.msgChan, "FundingSigned",
+	).(*lnwire.FundingSigned)
+
+	// Forward the signature to Alice.
+	alice.fundingMgr.processFundingSigned(fundingSigned, bob)
+
+	// After Alice processes the singleFundingSignComplete message, she will
+	// broadcast the funding transaction to the network. We expect to get a
+	// channel update saying the channel is pending.
+	var pendingUpdate *lnrpc.OpenStatusUpdate
+	select {
+	case pendingUpdate = <-updateChan:
+	case <-time.After(time.Second * 5):
+		t.Fatalf("alice did not send OpenStatusUpdate_ChanPending")
+	}
+
+	_, ok = pendingUpdate.Update.(*lnrpc.OpenStatusUpdate_ChanPending)
+	if !ok {
+		t.Fatal("OpenStatusUpdate was not OpenStatusUpdate_ChanPending")
+	}
+
+	// Wait for Alice to published the funding tx to the network.
+	select {
+	case <-alice.publTxChan:
+	case <-time.After(time.Second * 5):
+		t.Fatalf("alice did not publish funding tx")
+	}
+}
+
+// TestFundingManagerMaxPendingChannels checks that trying to open another
+// channel with the same peer when MaxPending channels are pending fails.
+func TestFundingManagerMaxPendingChannels(t *testing.T) {
+	const maxPending = 4
+
+	alice, bob := setupFundingManagers(t, maxPending)
+	defer tearDownFundingManagers(t, alice, bob)
+
+	// Create openChanReqs for maxPending+1 channels.
+	var initReqs []*openChanReq
+	for i := 0; i < maxPending+1; i++ {
+		updateChan := make(chan *lnrpc.OpenStatusUpdate)
+		errChan := make(chan error, 1)
+		initReq := &openChanReq{
+			targetPubkey:    bob.privKey.PubKey(),
+			chainHash:       *activeNetParams.GenesisHash,
+			localFundingAmt: 5000000,
+			pushAmt:         lnwire.NewMSatFromSatoshis(0),
+			private:         false,
+			updates:         updateChan,
+			err:             errChan,
+		}
+		initReqs = append(initReqs, initReq)
+	}
+
+	// Kick of maxPending+1 funding workflows.
+	var accepts []*lnwire.AcceptChannel
+	var lastOpen *lnwire.OpenChannel
+	for i, initReq := range initReqs {
+		alice.fundingMgr.initFundingWorkflow(bob, initReq)
+
+		// Alice should have sent the OpenChannel message to Bob.
+		var aliceMsg lnwire.Message
+		select {
+		case aliceMsg = <-alice.msgChan:
+		case err := <-initReq.err:
+			t.Fatalf("error init funding workflow: %v", err)
+		case <-time.After(time.Second * 5):
+			t.Fatalf("alice did not send OpenChannel message")
+		}
+
+		openChannelReq, ok := aliceMsg.(*lnwire.OpenChannel)
+		if !ok {
+			errorMsg, gotError := aliceMsg.(*lnwire.Error)
+			if gotError {
+				t.Fatalf("expected OpenChannel to be sent "+
+					"from bob, instead got error: %v",
+					lnwire.ErrorCode(errorMsg.Data[0]))
+			}
+			t.Fatalf("expected OpenChannel to be sent from "+
+				"alice, instead got %T", aliceMsg)
+		}
+
+		// Let Bob handle the init message.
+		bob.fundingMgr.processFundingOpen(openChannelReq, alice)
+
+		// Bob should answer with an AcceptChannel message for the
+		// first maxPending channels.
+		if i < maxPending {
+			acceptChannelResponse := assertFundingMsgSent(
+				t, bob.msgChan, "AcceptChannel",
+			).(*lnwire.AcceptChannel)
+			accepts = append(accepts, acceptChannelResponse)
+			continue
+		}
+
+		// For the last channel, Bob should answer with an error.
+		lastOpen = openChannelReq
+		_ = assertFundingMsgSent(
+			t, bob.msgChan, "Error",
+		).(*lnwire.Error)
+
+	}
+
+	// Forward the responses to Alice.
+	var signs []*lnwire.FundingSigned
+	for _, accept := range accepts {
+		alice.fundingMgr.processFundingAccept(accept, bob)
+
+		// Alice responds with a FundingCreated message.
+		fundingCreated := assertFundingMsgSent(
+			t, alice.msgChan, "FundingCreated",
+		).(*lnwire.FundingCreated)
+
+		// Give the message to Bob.
+		bob.fundingMgr.processFundingCreated(fundingCreated, alice)
+
+		// Finally, Bob should send the FundingSigned message.
+		fundingSigned := assertFundingMsgSent(
+			t, bob.msgChan, "FundingSigned",
+		).(*lnwire.FundingSigned)
+
+		signs = append(signs, fundingSigned)
+	}
+
+	// Sending another init request from Alice should still make Bob
+	// respond with an error.
+	bob.fundingMgr.processFundingOpen(lastOpen, alice)
+	_ = assertFundingMsgSent(
+		t, bob.msgChan, "Error",
+	).(*lnwire.Error)
+
+	// Give the FundingSigned messages to Alice.
+	for i, sign := range signs {
+		alice.fundingMgr.processFundingSigned(sign, bob)
+
+		// Alice should send a status update for each channel, and
+		// publish a funding tx to the network.
+		var pendingUpdate *lnrpc.OpenStatusUpdate
+		select {
+		case pendingUpdate = <-initReqs[i].updates:
+		case <-time.After(time.Second * 5):
+			t.Fatalf("alice did not send OpenStatusUpdate_ChanPending")
+		}
+
+		_, ok := pendingUpdate.Update.(*lnrpc.OpenStatusUpdate_ChanPending)
+		if !ok {
+			t.Fatal("OpenStatusUpdate was not OpenStatusUpdate_ChanPending")
+		}
+
+		select {
+		case <-alice.publTxChan:
+		case <-time.After(time.Second * 5):
+			t.Fatalf("alice did not publish funding tx")
+		}
+
+	}
+
+	// Sending another init request from Alice should still make Bob
+	// respond with an error, since the funding transactions are not
+	// confirmed yet,
+	bob.fundingMgr.processFundingOpen(lastOpen, alice)
+	_ = assertFundingMsgSent(
+		t, bob.msgChan, "Error",
+	).(*lnwire.Error)
+
+	// Notify that the transactions were mined.
+	for i := 0; i < maxPending; i++ {
+		alice.mockNotifier.oneConfChannel <- &chainntnfs.TxConfirmation{}
+		bob.mockNotifier.oneConfChannel <- &chainntnfs.TxConfirmation{}
+
+		// Expect both to be sending FundingLocked.
+		_ = assertFundingMsgSent(
+			t, alice.msgChan, "FundingLocked",
+		).(*lnwire.FundingLocked)
+
+		_ = assertFundingMsgSent(
+			t, bob.msgChan, "FundingLocked",
+		).(*lnwire.FundingLocked)
+
+	}
+
+	// Now opening another channel should work.
+	bob.fundingMgr.processFundingOpen(lastOpen, alice)
+
+	// Bob should answer with an AcceptChannel message.
+	_ = assertFundingMsgSent(
+		t, bob.msgChan, "AcceptChannel",
+	).(*lnwire.AcceptChannel)
 }

lnwallet/interface_test.go

@@ -427,6 +427,11 @@ func testDualFundingReservationWorkflow(miner *rpctest.Harness,
 		t.Fatalf("channel not detected as dual funder")
 	}
 
+	// Let Alice publish the funding transaction.
+	if err := alice.PublishTransaction(fundingTx); err != nil {
+		t.Fatalf("unable to publish funding tx: %v", err)
+	}
+
 	// Mine a single block, the funding transaction should be included
 	// within this block.
 	err = waitForMempoolTx(miner, &fundingSha)
@@ -843,6 +848,11 @@ func testSingleFunderReservationWorkflow(miner *rpctest.Harness,
 			channeldb.SingleFunder, bobChannels[0].ChanType)
 	}
 
+	// Let Alice publish the funding transaction.
+	if err := alice.PublishTransaction(fundingTx); err != nil {
+		t.Fatalf("unable to publish funding tx: %v", err)
+	}
+
 	// Mine a single block, the funding transaction should be included
 	// within this block.
 	err = waitForMempoolTx(miner, &fundingSha)

lnwallet/reservation.go

@@ -434,11 +434,11 @@ func (r *ChannelReservation) OurSignatures() ([]*InputScript, []byte) {
 // https://github.com/bitcoin/bips/blob/master/bip-0069.mediawiki.
 // Additionally, verification is performed in order to ensure that the
 // counterparty supplied a valid signature to our version of the commitment
-// transaction.  Once this method returns, caller's should then call
+// transaction.  Once this method returns, caller's should broadcast the
-// .WaitForChannelOpen() which will block until the funding transaction obtains
+// created funding transaction, then call .WaitForChannelOpen() which will
-// the configured number of confirmations. Once the method unblocks, a
+// block until the funding transaction obtains the configured number of
-// LightningChannel instance is returned, marking the channel available for
+// confirmations. Once the method unblocks, a LightningChannel instance is
-// updates.
+// returned, marking the channel available for updates.
 func (r *ChannelReservation) CompleteReservation(fundingInputScripts []*InputScript,
 	commitmentSig []byte) (*channeldb.OpenChannel, error) {
 

lnwallet/wallet.go

@@ -1096,16 +1096,6 @@ func (l *LightningWallet) handleFundingCounterPartySigs(msg *addCounterPartySigs
 		return
 	}
 
-	walletLog.Infof("Broadcasting funding tx for ChannelPoint(%v): %v",
-		res.partialState.FundingOutpoint, spew.Sdump(fundingTx))
-
-	// Broadcast the finalized funding transaction to the network.
-	if err := l.PublishTransaction(fundingTx); err != nil {
-		msg.err <- err
-		msg.completeChan <- nil
-		return
-	}
-
 	msg.completeChan <- res.partialState
 	msg.err <- nil
 }

mock.go

@@ -4,6 +4,7 @@ import (
 	"crypto/sha256"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"github.com/btcsuite/btcd/btcec"
 	"github.com/btcsuite/btcd/chaincfg"
@@ -191,6 +192,7 @@ type mockWalletController struct {
 	rootKey               *btcec.PrivateKey
 	prevAddres            btcutil.Address
 	publishedTransactions chan *wire.MsgTx
+	index                 uint32
 }
 
 // BackEnd returns "mock" to signify a mock wallet controller.
@@ -231,16 +233,17 @@ func (*mockWalletController) SendOutputs(outputs []*wire.TxOut,
 
 // ListUnspentWitness is called by the wallet when doing coin selection. We just
 // need one unspent for the funding transaction.
-func (*mockWalletController) ListUnspentWitness(confirms int32) ([]*lnwallet.Utxo, error) {
+func (m *mockWalletController) ListUnspentWitness(confirms int32) ([]*lnwallet.Utxo, error) {
 	utxo := &lnwallet.Utxo{
 		AddressType: lnwallet.WitnessPubKey,
 		Value:       btcutil.Amount(10 * btcutil.SatoshiPerBitcoin),
 		PkScript:    make([]byte, 22),
 		OutPoint: wire.OutPoint{
 			Hash:  chainhash.Hash{},
-			Index: 0,
+			Index: m.index,
 		},
 	}
+	atomic.AddUint32(&m.index, 1)
 	var ret []*lnwallet.Utxo
 	ret = append(ret, utxo)
 	return ret, nil