contractcourt: unify+simplify new re-org aware logic

contractcourt/chain_watcher.go

@@ -695,6 +695,93 @@ func newChainSet(chanState *channeldb.OpenChannel) (*chainSet, error) {
 	}, nil
 }
 
+// spendProcessResult holds the results of processing a detected spend.
+type spendProcessResult struct {
+	// pendingSpend is the spend to track (nil if fast-path was used).
+	pendingSpend *chainntnfs.SpendDetail
+
+	// confNtfn is the confirmation notification (nil if fast-path or
+	// error).
+	confNtfn *chainntnfs.ConfirmationEvent
+}
+
+// processDetectedSpend handles a newly detected spend from either blockbeat or
+// spend notification. It determines whether to use the fast-path (single conf)
+// or async-path (multiple confs), and returns the updated state.
+//
+// For single-confirmation mode (numConfs == 1), it immediately dispatches the
+// close event and returns empty result. For multi-confirmation mode, it
+// registers for confirmations and returns the new pending state.
+func (c *chainWatcher) processDetectedSpend(
+	spend *chainntnfs.SpendDetail, source string,
+	currentPendingSpend *chainntnfs.SpendDetail,
+	currentConfNtfn *chainntnfs.ConfirmationEvent) spendProcessResult {
+
+	// FAST PATH: Single confirmation mode dispatches immediately.
+	if c.handleSpendDispatch(spend, source) {
+		if currentConfNtfn != nil {
+			currentConfNtfn.Cancel()
+		}
+
+		return spendProcessResult{}
+	}
+
+	// ASYNC PATH: Multiple confirmations (production).
+	// STATE TRANSITION: None -> Pending.
+	log.Infof("ChannelPoint(%v): detected spend from %s, "+
+		"transitioning to %v", c.cfg.chanState.FundingOutpoint,
+		source, spendStatePending)
+
+	// Check for duplicate spend detection.
+	if currentPendingSpend != nil {
+		if *currentPendingSpend.SpenderTxHash == *spend.SpenderTxHash {
+			log.Debugf("ChannelPoint(%v): ignoring duplicate "+
+				"spend detection for tx %v",
+				c.cfg.chanState.FundingOutpoint,
+				spend.SpenderTxHash)
+
+			return spendProcessResult{
+				pendingSpend: currentPendingSpend,
+				confNtfn:     currentConfNtfn,
+			}
+		}
+
+		// Different spend detected. Cancel existing confNtfn.
+		log.Warnf("ChannelPoint(%v): detected different spend tx %v, "+
+			"replacing pending tx %v",
+			c.cfg.chanState.FundingOutpoint,
+			spend.SpenderTxHash, currentPendingSpend.SpenderTxHash)
+
+		if currentConfNtfn != nil {
+			currentConfNtfn.Cancel()
+		}
+	}
+
+	numConfs := c.requiredConfsForSpend()
+	txid := spend.SpenderTxHash
+
+	newConfNtfn, err := c.cfg.notifier.RegisterConfirmationsNtfn(
+		txid, spend.SpendingTx.TxOut[0].PkScript, numConfs,
+		uint32(spend.SpendingHeight),
+	)
+	if err != nil {
+		log.Errorf("Unable to register confirmations: %v", err)
+
+		return spendProcessResult{
+			pendingSpend: currentPendingSpend,
+			confNtfn:     currentConfNtfn,
+		}
+	}
+
+	log.Infof("ChannelPoint(%v): waiting for %d confirmations of "+
+		"spend tx %v", c.cfg.chanState.FundingOutpoint, numConfs, txid)
+
+	return spendProcessResult{
+		pendingSpend: spend,
+		confNtfn:     newConfNtfn,
+	}
+}
+
 // closeObserver is a dedicated goroutine that will watch for any closes of the
 // channel that it's watching on chain. It implements a state machine to handle
 // spend detection and confirmation with reorg protection. The states are:
@@ -747,57 +834,6 @@ func (c *chainWatcher) closeObserver() {
 	log.Infof("Close observer for ChannelPoint(%v) active",
 		c.cfg.chanState.FundingOutpoint)
 
-	// handleSpendDetection processes a newly detected spend by registering
-	// for confirmations. Returns the new confNtfn or error.
-	handleSpendDetection := func(
-		spend *chainntnfs.SpendDetail,
-	) (*chainntnfs.ConfirmationEvent, error) {
-
-		// If we already have a pending spend, check if it's the same
-		// transaction. This can happen if both the spend notification
-		// and blockbeat detect the same spend.
-		if pendingSpend != nil {
-			if *pendingSpend.SpenderTxHash == *spend.SpenderTxHash {
-				log.Debugf("ChannelPoint(%v): ignoring "+
-					"duplicate spend detection for tx %v",
-					c.cfg.chanState.FundingOutpoint,
-					spend.SpenderTxHash)
-
-				return confNtfn, nil
-			}
-
-			// Different spend detected. Cancel existing confNtfn
-			// and replace with new one.
-			log.Warnf("ChannelPoint(%v): detected different "+
-				"spend tx %v, replacing pending tx %v",
-				c.cfg.chanState.FundingOutpoint,
-				spend.SpenderTxHash,
-				pendingSpend.SpenderTxHash)
-
-			if confNtfn != nil {
-				confNtfn.Cancel()
-			}
-		}
-
-		numConfs := c.requiredConfsForSpend()
-		txid := spend.SpenderTxHash
-
-		newConfNtfn, err := c.cfg.notifier.RegisterConfirmationsNtfn(
-			txid, spend.SpendingTx.TxOut[0].PkScript,
-			numConfs, uint32(spend.SpendingHeight),
-		)
-		if err != nil {
-			return nil, fmt.Errorf("register confirmations: %w",
-				err)
-		}
-
-		log.Infof("ChannelPoint(%v): waiting for %d confirmations "+
-			"of spend tx %v", c.cfg.chanState.FundingOutpoint,
-			numConfs, txid)
-
-		return newConfNtfn, nil
-	}
-
 	for {
 		// We only listen to confirmation channels when we have a
 		// pending spend. By setting these to nil when not needed, Go's
@@ -813,6 +849,9 @@ func (c *chainWatcher) closeObserver() {
 		}
 
 		select {
+		// A new block beat has just arrived, we'll handle the block
+		// beat, and see if it contains the spend of our funding
+		// transaction or not.
 		case beat := <-c.BlockbeatChan:
 			log.Debugf("ChainWatcher(%v) received blockbeat %v",
 				c.cfg.chanState.FundingOutpoint, beat.Height())
@@ -822,73 +861,33 @@ func (c *chainWatcher) closeObserver() {
 				continue
 			}
 
-			// FAST PATH: Check if we should dispatch immediately
+			result := c.processDetectedSpend(
-			// for single-confirmation scenarios.
+				spend, "blockbeat", pendingSpend, confNtfn,
-			if c.handleSpendDispatch(spend, "blockbeat") {
+			)
-				if confNtfn != nil {
-					confNtfn.Cancel()
-					confNtfn = nil
-				}
-				pendingSpend = nil
-				continue
-			}
 
-			// ASYNC PATH: Multiple confirmations (production).
+			pendingSpend = result.pendingSpend
-			// STATE TRANSITION: None -> Pending (from blockbeat).
+			confNtfn = result.confNtfn
-			// We've detected a spend, but don't process it yet.
-			// Instead, register for confirmations to protect
-			// against shallow reorgs.
-			log.Infof("ChannelPoint(%v): detected spend from "+
-				"blockbeat, transitioning to %v",
-				c.cfg.chanState.FundingOutpoint,
-				spendStatePending)
 
-			newConfNtfn, err := handleSpendDetection(spend)
+		// A direct spend was just detected, we'll process the new spend
-			if err != nil {
+		// then see if we need to dispatch instantly, or wait around for
-				log.Errorf("Unable to handle spend "+
+		// additional confirmations.
-					"detection: %v", err)
-				continue
-			}
-			pendingSpend = spend
-			confNtfn = newConfNtfn
-
-		// STATE TRANSITION: None -> Pending.
-		// We've detected a spend, but don't process it yet. Instead,
-		// register for confirmations to protect against shallow reorgs.
 		case spend, ok := <-spendNtfn.Spend:
 			if !ok {
 				return
 			}
 
-			// FAST PATH: Check if we should dispatch immediately
+			result := c.processDetectedSpend(
-			// for single-confirmation scenarios.
+				spend, "spend notification", pendingSpend,
-			if c.handleSpendDispatch(spend, "spend notification") {
+				confNtfn,
-				if confNtfn != nil {
+			)
-					confNtfn.Cancel()
-					confNtfn = nil
-				}
-				pendingSpend = nil
-				continue
-			}
 
-			// ASYNC PATH: Multiple confirmations (production).
+			pendingSpend = result.pendingSpend
-			log.Infof("ChannelPoint(%v): detected spend from "+
+			confNtfn = result.confNtfn
-				"notification, transitioning to %v",
-				c.cfg.chanState.FundingOutpoint,
-				spendStatePending)
 
-			newConfNtfn, err := handleSpendDetection(spend)
-			if err != nil {
-				log.Errorf("Unable to handle spend "+
-					"detection: %v", err)
-				continue
-			}
-			pendingSpend = spend
-			confNtfn = newConfNtfn
-
-		// STATE TRANSITION: Pending -> Confirmed
 		// The spend has reached required confirmations. It's now safe
 		// to process since we've protected against shallow reorgs.
+		//
+		// * STATE TRANSITION: Pending -> Confirmed
 		case conf, ok := <-confChan:
 			if !ok {
 				log.Errorf("Confirmation channel closed " +
@@ -911,10 +910,11 @@ func (c *chainWatcher) closeObserver() {
 			confNtfn = nil
 			pendingSpend = nil
 
-		// STATE TRANSITION: Pending -> None
 		// A reorg removed the spend tx. We reset to initial state and
 		// wait for ANY new spend (could be the same tx re-mined, or a
 		// different tx like an RBF replacement).
+		//
+		// * STATE TRANSITION: Pending -> None
 		case reorgDepth, ok := <-negativeConfChan:
 			if !ok {
 				log.Errorf("Negative conf channel closed " +