Merge pull request #2682 from valentinewallace/fix-subscribe-chans-flake

lnd_test: fix channel event subscription test flake.

lntest/itest/lnd_test.go

@@ -5972,10 +5972,13 @@ func testBasicChannelCreationAndUpdates(net *lntest.NetworkHarness, t *harnessTe
 		amount      = lnd.MaxBtcFundingAmount
 	)
 
-	// Let Bob subscribe to channel notifications.
+	// Subscribe Bob and Alice to channel event notifications.
 	bobChanSub := subscribeChannelNotifications(ctxb, t, net.Bob)
 	defer close(bobChanSub.quit)
 
+	aliceChanSub := subscribeChannelNotifications(ctxb, t, net.Alice)
+	defer close(aliceChanSub.quit)
+
 	// Open the channel between Alice and Bob, asserting that the
 	// channel has been properly open on-chain.
 	chanPoints := make([]*lnrpc.ChannelPoint, numChannels)
@@ -5989,31 +5992,52 @@ func testBasicChannelCreationAndUpdates(net *lntest.NetworkHarness, t *harnessTe
 		)
 	}
 
-	// Since each of the channels just became open, Bob should we receive an
-	// open and an active notification for each channel.
+	// Since each of the channels just became open, Bob and Alice should
+	// each receive an open and an active notification for each channel.
 	var numChannelUpds int
-	for numChannelUpds < 2*numChannels {
-		select {
-		case update := <-bobChanSub.updateChan:
-			switch update.Type {
-			case lnrpc.ChannelEventUpdate_ACTIVE_CHANNEL:
-			case lnrpc.ChannelEventUpdate_OPEN_CHANNEL:
-			default:
-				t.Fatalf("update type mismatch: expected open or active "+
-					"channel notification, got: %v", update.Type)
+	const totalNtfns = 2 * numChannels
+	verifyOpenUpdatesReceived := func(sub channelSubscription) error {
+		numChannelUpds = 0
+		for numChannelUpds < totalNtfns {
+			select {
+			case update := <-sub.updateChan:
+				switch update.Type {
+				case lnrpc.ChannelEventUpdate_ACTIVE_CHANNEL:
+					if numChannelUpds%2 != 1 {
+						return fmt.Errorf("expected open" +
+							"channel ntfn, got active " +
+							"channel ntfn instead")
+					}
+				case lnrpc.ChannelEventUpdate_OPEN_CHANNEL:
+					if numChannelUpds%2 != 0 {
+						return fmt.Errorf("expected active" +
+							"channel ntfn, got open" +
+							"channel ntfn instead")
+					}
+				default:
+					return fmt.Errorf("update type mismatch: "+
+						"expected open or active channel "+
+						"notification, got: %v",
+						update.Type)
+				}
+				numChannelUpds++
+			case <-time.After(time.Second * 10):
+				return fmt.Errorf("timeout waiting for channel "+
+					"notifications, only received %d/%d "+
+					"chanupds", numChannelUpds,
+					totalNtfns)
 			}
-			numChannelUpds++
-		case <-time.After(time.Second * 10):
-			t.Fatalf("timeout waiting for channel notifications, "+
-				"only received %d/%d chanupds", numChannelUpds,
-				numChannels)
 		}
+
+		return nil
 	}
 
-	// Subscribe Alice to channel updates so we can test that both remote
-	// and local force close notifications are received correctly.
-	aliceChanSub := subscribeChannelNotifications(ctxb, t, net.Alice)
-	defer close(aliceChanSub.quit)
+	if err := verifyOpenUpdatesReceived(bobChanSub); err != nil {
+		t.Fatalf("error verifying open updates: %v", err)
+	}
+	if err := verifyOpenUpdatesReceived(aliceChanSub); err != nil {
+		t.Fatalf("error verifying open updates: %v", err)
+	}
 
 	// Close the channel between Alice and Bob, asserting that the channel
 	// has been properly closed on-chain.