itest: fix test flakes from open channel not found and tx not found in mempool

breacharbiter.go

@@ -176,6 +176,7 @@ func (b *breachArbiter) start() error {
 	}, func() {
 		breachRetInfos = make(map[wire.OutPoint]retributionInfo)
 	}); err != nil {
+		brarLog.Errorf("Unable to create retribution info: %v", err)
 		return err
 	}
 
@@ -190,6 +191,9 @@ func (b *breachArbiter) start() error {
 		return err
 	}
 
+	brarLog.Debugf("Found %v closing channels, %v retribution records",
+		len(closedChans), len(breachRetInfos))
+
 	// Using the set of non-pending, closed channels, reconcile any
 	// discrepancies between the channeldb and the retribution store by
 	// removing any retribution information for which we have already
@@ -199,6 +203,9 @@ func (b *breachArbiter) start() error {
 	// TODO(halseth): no need continue on IsPending once closed channels
 	// actually means close transaction is confirmed.
 	for _, chanSummary := range closedChans {
+		brarLog.Debugf("Working on close channel: %v, is_pending: %v",
+			chanSummary.ChanPoint, chanSummary.IsPending)
+
 		if chanSummary.IsPending {
 			continue
 		}
@@ -212,6 +219,9 @@ func (b *breachArbiter) start() error {
 				return err
 			}
 			delete(breachRetInfos, *chanPoint)
+
+			brarLog.Debugf("Skipped closed channel: %v",
+				chanSummary.ChanPoint)
 		}
 	}
 
@@ -220,6 +230,9 @@ func (b *breachArbiter) start() error {
 	for chanPoint := range breachRetInfos {
 		retInfo := breachRetInfos[chanPoint]
 
+		brarLog.Debugf("Handling breach handoff on startup "+
+			"for ChannelPoint(%v)", chanPoint)
+
 		// Register for a notification when the breach transaction is
 		// confirmed on chain.
 		breachTXID := retInfo.commitHash

contractcourt/channel_arbitrator.go

@@ -435,10 +435,10 @@ func (c *ChannelArbitrator) Start(state *chanArbStartState) error {
 		}
 	}
 
-	log.Debugf("Starting ChannelArbitrator(%v), htlc_set=%v",
+	log.Debugf("Starting ChannelArbitrator(%v), htlc_set=%v, state=%v",
 		c.cfg.ChanPoint, newLogClosure(func() string {
 			return spew.Sdump(c.activeHTLCs)
-		}),
+		}), state.currentState,
 	)
 
 	// Set our state from our starting state.
@@ -795,7 +795,7 @@ func (c *ChannelArbitrator) stateStep(
 		// default state. If this isn't a self initiated event (we're
 		// checking due to a chain update), then we'll exit now.
 		if len(chainActions) == 0 && trigger == chainTrigger {
-			log.Tracef("ChannelArbitrator(%v): no actions for "+
+			log.Debugf("ChannelArbitrator(%v): no actions for "+
 				"chain trigger, terminating", c.cfg.ChanPoint)
 
 			return StateDefault, closeTx, nil
@@ -1303,7 +1303,7 @@ func (c *ChannelArbitrator) advanceState(
 	// transition to is that same state that we started at.
 	for {
 		priorState = c.state
-		log.Tracef("ChannelArbitrator(%v): attempting state step with "+
+		log.Debugf("ChannelArbitrator(%v): attempting state step with "+
 			"trigger=%v from state=%v", c.cfg.ChanPoint, trigger,
 			priorState)
 
@@ -1324,7 +1324,7 @@ func (c *ChannelArbitrator) advanceState(
 		// our prior state back as the next state, then we'll
 		// terminate.
 		if nextState == priorState {
-			log.Tracef("ChannelArbitrator(%v): terminating at "+
+			log.Debugf("ChannelArbitrator(%v): terminating at "+
 				"state=%v", c.cfg.ChanPoint, nextState)
 			return nextState, forceCloseTx, nil
 		}

docs/release-notes/release-notes-0.14.0.md

@@ -239,6 +239,10 @@ you.
 
 * [Upgraded miekg/dns to improve the security posture](https://github.com/lightningnetwork/lnd/pull/5738)
 
+* [Fixed flakes caused by graph topology subcription](https://github.com/lightningnetwork/lnd/pull/5611).
+
+* [Order of the start/stop on subsystems are changed to promote better safety](https://github.com/lightningnetwork/lnd/pull/1783).
+
 ## Database
 
 * [Ensure single writer for legacy

lntest/harness.go

@@ -1263,56 +1263,52 @@ func (n *NetworkHarness) CloseChannel(lnNode *HarnessNode,
 		}
 	}
 
-	closeReq := &lnrpc.CloseChannelRequest{
-		ChannelPoint: cp,
-		Force:        force,
-	}
-	closeRespStream, err := lnNode.CloseChannel(ctx, closeReq)
-	if err != nil {
-		return nil, nil, fmt.Errorf("unable to close channel: %v", err)
-	}
+	var (
+		closeRespStream lnrpc.Lightning_CloseChannelClient
+		closeTxid       *chainhash.Hash
+	)
 
-	errChan := make(chan error)
-	fin := make(chan *chainhash.Hash)
-	go func() {
-		// Consume the "channel close" update in order to wait for the closing
-		// transaction to be broadcast, then wait for the closing tx to be seen
-		// within the network.
+	err = wait.NoError(func() error {
+		closeReq := &lnrpc.CloseChannelRequest{
+			ChannelPoint: cp, Force: force,
+		}
+		closeRespStream, err = lnNode.CloseChannel(ctx, closeReq)
+		if err != nil {
+			return fmt.Errorf("unable to close channel: %v", err)
+		}
+
+		// Consume the "channel close" update in order to wait for the
+		// closing transaction to be broadcast, then wait for the
+		// closing tx to be seen within the network.
 		closeResp, err := closeRespStream.Recv()
 		if err != nil {
-			errChan <- fmt.Errorf("unable to recv() from close "+
+			return fmt.Errorf("unable to recv() from close "+
 				"stream: %v", err)
-			return
 		}
 		pendingClose, ok := closeResp.Update.(*lnrpc.CloseStatusUpdate_ClosePending)
 		if !ok {
-			errChan <- fmt.Errorf("expected channel close update, "+
+			return fmt.Errorf("expected channel close update, "+
 				"instead got %v", pendingClose)
-			return
 		}
 
-		closeTxid, err := chainhash.NewHash(pendingClose.ClosePending.Txid)
+		closeTxid, err = chainhash.NewHash(
+			pendingClose.ClosePending.Txid,
+		)
 		if err != nil {
-			errChan <- fmt.Errorf("unable to decode closeTxid: "+
+			return fmt.Errorf("unable to decode closeTxid: "+
 				"%v", err)
-			return
 		}
 		if err := n.waitForTxInMempool(ctx, *closeTxid); err != nil {
-			errChan <- fmt.Errorf("error while waiting for "+
+			return fmt.Errorf("error while waiting for "+
 				"broadcast tx: %v", err)
-			return
 		}
-		fin <- closeTxid
-	}()
-
-	// Wait until either the deadline for the context expires, an error
-	// occurs, or the channel close update is received.
-	select {
-	case err := <-errChan:
+		return nil
+	}, ChannelCloseTimeout)
+	if err != nil {
 		return nil, nil, err
-	case closeTxid := <-fin:
-		return closeRespStream, closeTxid, nil
 	}
+
+	return closeRespStream, closeTxid, nil
 }
 
 // WaitForChannelClose waits for a notification from the passed channel close