chainbackends: consume real reorg depth and Done from lndclient

.golangci.yml

@@ -83,6 +83,11 @@ linters:
         # Allow btcwallet to follow the 0.21 lnd/lndclient stack while
         # taproot-assets still requires the released v0.16.17 tag.
         - github.com/btcsuite/btcwallet
+        # Temporary forks that surface the chain-notifier reorg depth and
+        # Done events over gRPC. Drop once these land upstream and tag
+        # (tracked in darepo-client#827).
+        - github.com/lightningnetwork/lnd
+        - github.com/lightninglabs/lndclient
   disable:
     # We instead use our own custom line length linter called `ll` since
     # then we can ignore log lines.

btcwbackend/chain_backend.go

@@ -38,8 +38,13 @@ type ChainBackend struct {
 	neutrinoCS *neutrino.ChainService
 
 	// notifier provides chain notification services backed by
-	// neutrino's compact block filter scanning.
-	notifier *neutrinonotify.NeutrinoNotifier
+	// neutrino's compact block filter scanning. Stored under the
+	// chainntnfs interface (concrete type is
+	// *neutrinonotify.NeutrinoNotifier in production) so reorg-aware
+	// forwarder tests can substitute a stub that drives the full
+	// Confirmed / NegativeConf / Done lifecycle without spinning up a
+	// neutrino chain service.
+	notifier chainntnfs.ChainNotifier
 
 	// feeEstimator provides fee estimation from a web API since
 	// neutrino has no mempool visibility.
@@ -571,40 +576,89 @@ func (b *ChainBackend) RegisterConf(ctx context.Context, txid *chainhash.Hash,
 		cancelOnce.Do(event.Cancel)
 	}
 
+	// Create channels to convert neutrino's confirmation lifecycle
+	// (which already drives chainntnfs's NegativeConf/Done channels
+	// directly — neutrino emits these natively from its compact-
+	// block-filter scanner) into our backend-agnostic types.
+	// NegativeConf carries a reorg depth that the cross-backend
+	// chainsource layer intentionally drops, so we forward a bare
+	// struct{} signal on reorgChan instead.
 	confChan := make(chan *chainsource.TxConfirmation, 1)
+	reorgChan := make(chan uint64, 1)
+	doneChan := make(chan struct{}, 1)
 
 	go func() {
+		// Defers run in LIFO order. event.Cancel() must run first
+		// so the upstream notifier stops writing to its internal
+		// channels before we cancel notifyCtx (which any in-flight
+		// downstream sends are still using) and finally close the
+		// outgoing chans. Reversing this order would race the
+		// upstream notifier against closed channels.
 		defer close(confChan)
+		defer close(reorgChan)
+		defer close(doneChan)
 		defer cancel()
 		defer safeCancel()
 
-		select {
-		case lndConf, ok := <-event.Confirmed:
-			if !ok {
-				return
-			}
+		for {
+			select {
+			case lndConf, ok := <-event.Confirmed:
+				if !ok {
+					return
+				}
 
-			conf := &chainsource.TxConfirmation{
-				BlockHash:   lndConf.BlockHash,
-				BlockHeight: lndConf.BlockHeight,
-				TxIndex:     lndConf.TxIndex,
-				Tx:          lndConf.Tx,
-				Block:       lndConf.Block,
-			}
+				conf := &chainsource.TxConfirmation{
+					BlockHash:   lndConf.BlockHash,
+					BlockHeight: lndConf.BlockHeight,
+					TxIndex:     lndConf.TxIndex,
+					Tx:          lndConf.Tx,
+					Block:       lndConf.Block,
+				}
+
+				select {
+				case confChan <- conf:
+				case <-notifyCtx.Done():
+					return
+				}
+
+			case _, ok := <-event.NegativeConf:
+				if !ok {
+					event.NegativeConf = nil
+
+					continue
+				}
+
+				select {
+				case reorgChan <- uint64(0):
+				case <-notifyCtx.Done():
+					return
+				}
+
+			case _, ok := <-event.Done:
+				if !ok {
+					event.Done = nil
+
+					continue
+				}
+
+				select {
+				case doneChan <- struct{}{}:
+				case <-notifyCtx.Done():
+					return
+				}
+
+				return
 
-			select {
-			case confChan <- conf:
 			case <-notifyCtx.Done():
 				return
 			}
-
-		case <-notifyCtx.Done():
-			return
 		}
 	}()
 
 	return &chainsource.ConfRegistration{
 		Confirmed: confChan,
+		Reorged:   reorgChan,
+		Done:      doneChan,
 		Cancel: func() {
 			cancel()
 			safeCancel()
@@ -641,40 +695,83 @@ func (b *ChainBackend) RegisterSpend(ctx context.Context,
 		cancelOnce.Do(event.Cancel)
 	}
 
+	// Create channels to convert neutrino's spend lifecycle into our
+	// backend-agnostic types. neutrino's Reorg carries no payload, so
+	// we forward a bare struct{} signal.
 	spendChan := make(chan *chainsource.SpendDetail, 1)
+	reorgChan := make(chan uint64, 1)
+	doneChan := make(chan struct{}, 1)
 
 	go func() {
+		// LIFO defer: event.Cancel() first so the upstream notifier
+		// stops writing, then cancel notifyCtx so in-flight downstream
+		// sends unblock, and finally close the outgoing chans.
 		defer close(spendChan)
+		defer close(reorgChan)
+		defer close(doneChan)
 		defer cancel()
 		defer safeCancel()
 
-		select {
-		case lndSpend, ok := <-event.Spend:
-			if !ok {
-				return
-			}
+		for {
+			select {
+			case lndSpend, ok := <-event.Spend:
+				if !ok {
+					return
+				}
 
-			spend := &chainsource.SpendDetail{
-				SpentOutPoint:     lndSpend.SpentOutPoint,
-				SpenderTxHash:     lndSpend.SpenderTxHash,
-				SpendingTx:        lndSpend.SpendingTx,
-				SpenderInputIndex: lndSpend.SpenderInputIndex,
-				SpendingHeight:    lndSpend.SpendingHeight,
-			}
+				spend := &chainsource.SpendDetail{
+					SpentOutPoint: lndSpend.SpentOutPoint,
+					SpenderTxHash: lndSpend.SpenderTxHash,
+					SpendingTx:    lndSpend.SpendingTx,
+					SpenderInputIndex: lndSpend.
+						SpenderInputIndex,
+					SpendingHeight: lndSpend.SpendingHeight,
+				}
+
+				select {
+				case spendChan <- spend:
+				case <-notifyCtx.Done():
+					return
+				}
+
+			case _, ok := <-event.Reorg:
+				if !ok {
+					event.Reorg = nil
+
+					continue
+				}
+
+				select {
+				case reorgChan <- uint64(0):
+				case <-notifyCtx.Done():
+					return
+				}
+
+			case _, ok := <-event.Done:
+				if !ok {
+					event.Done = nil
+
+					continue
+				}
+
+				select {
+				case doneChan <- struct{}{}:
+				case <-notifyCtx.Done():
+					return
+				}
+
+				return
 
-			select {
-			case spendChan <- spend:
 			case <-notifyCtx.Done():
 				return
 			}
-
-		case <-notifyCtx.Done():
-			return
 		}
 	}()
 
 	return &chainsource.SpendRegistration{
-		Spend: spendChan,
+		Spend:   spendChan,
+		Reorged: reorgChan,
+		Done:    doneChan,
 		Cancel: func() {
 			cancel()
 			safeCancel()