Beat [3/4]: Implement Consumer on chainWatcher and resolvers

chainio/blockbeat.go

@@ -1,9 +1,12 @@
 package chainio
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
+	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btclog"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/chainntnfs"
@@ -162,3 +165,122 @@ func (b Beat) notifyAndWait(c Consumer) error {
 
 	return nil
 }
+
+// HasOutpointSpent queries the block to find a spending tx that spends the
+// given outpoint. Returns the spend details if found, otherwise nil.
+//
+// NOTE: Part of the Blockbeat interface.
+func (b Beat) HasOutpointSpent(outpoint wire.OutPoint) *chainntnfs.SpendDetail {
+	b.log.Tracef("Querying spending tx for outpoint=%v", outpoint)
+
+	// Iterate all the txns in this block.
+	for _, tx := range b.epoch.Block.Transactions {
+		txHash := tx.TxHash()
+
+		// Iterate all the inputs in this tx.
+		for i, txIn := range tx.TxIn {
+			// Skip if the input doesn't spend the outpoint.
+			if txIn.PreviousOutPoint != outpoint {
+				continue
+			}
+
+			// Found a match, return the spend details.
+			details := &chainntnfs.SpendDetail{
+				SpentOutPoint:     &outpoint,
+				SpenderTxHash:     &txHash,
+				SpendingTx:        tx,
+				SpenderInputIndex: uint32(i),
+				SpendingHeight:    b.epoch.Height,
+			}
+
+			return details
+		}
+	}
+
+	return nil
+}
+
+// ErrPkScriptMismatch is returned when the expected pkScript doesn't match the
+// actual pkScript.
+var ErrPkScriptMismatch = errors.New("pkscript mismatch")
+
+// HasOutpointSpentByScript queries the block to find a spending tx that spends
+// the given outpoint using the pkScript.
+//
+// NOTE: Part of the Blockbeat interface.
+func (b Beat) HasOutpointSpentByScript(outpoint wire.OutPoint,
+	pkScript txscript.PkScript) (*chainntnfs.SpendDetail, error) {
+
+	b.log.Tracef("Querying spending tx for outpoint=%v, pkScript=%v",
+		outpoint, pkScript)
+
+	// For taproot outputs, we will skip matching the pkScript as we cannot
+	// derive the spent pkScript directly from the witness.
+	isTaproot := pkScript.Class() == txscript.WitnessV1TaprootTy
+
+	// matchTxIn is a helper closure that checks if the txIn spends the
+	// given outpoint using the specified pkScript. Returns an error if the
+	// outpoint is found but the pkScript doesn't match.
+	matchTxIn := func(txIn *wire.TxIn) (bool, error) {
+		prevOut := txIn.PreviousOutPoint
+
+		// Exit early if the input doesn't spend the outpoint.
+		if prevOut != outpoint {
+			return false, nil
+		}
+
+		// If this is a taproot output, we skip matching the pkScript.
+		if isTaproot {
+			return true, nil
+		}
+
+		// Compute the script and matches it with the pkScript.
+		script, err := txscript.ComputePkScript(
+			txIn.SignatureScript, txIn.Witness,
+		)
+		if err != nil {
+			b.log.Errorf("Failed to compute pkscript: %v", err)
+			return false, err
+		}
+
+		// Check if the scripts match.
+		if script != pkScript {
+			return false, fmt.Errorf("%w: want %v, got %v",
+				ErrPkScriptMismatch, pkScript, script)
+		}
+
+		return true, nil
+	}
+
+	// Iterate all the txns in this block.
+	for _, tx := range b.epoch.Block.Transactions {
+		txHash := tx.TxHash()
+
+		// Iterate all the inputs in this tx.
+		for i, txIn := range tx.TxIn {
+			// Check if the input spends the outpoint.
+			found, err := matchTxIn(txIn)
+			if err != nil {
+				return nil, err
+			}
+
+			// Skip if the input cannot be matched.
+			if !found {
+				continue
+			}
+
+			// Found a match, return the spend details.
+			details := &chainntnfs.SpendDetail{
+				SpentOutPoint:     &outpoint,
+				SpenderTxHash:     &txHash,
+				SpendingTx:        tx,
+				SpenderInputIndex: uint32(i),
+				SpendingHeight:    b.epoch.Height,
+			}
+
+			return details, nil
+		}
+	}
+
+	return nil, nil
+}

chainio/interface.go

@@ -1,5 +1,11 @@
 package chainio
 
+import (
+	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/chainntnfs"
+)
+
 // Blockbeat defines an interface that can be used by subsystems to retrieve
 // block data. It is sent by the BlockbeatDispatcher whenever a new block is
 // received. Once the subsystem finishes processing the block, it must signal
@@ -32,6 +38,17 @@ type Blockbeat interface {
 	// DispatchSequential sends the blockbeat to the specified consumers
 	// sequentially.
 	DispatchSequential(consumers []Consumer) error
+
+	// HasOutpointSpentByScript queries the block to find a spending tx
+	// that spends the given outpoint using the pkScript. Return an error
+	// is the outpoint is spent but using a different pkScript.
+	HasOutpointSpentByScript(outpoint wire.OutPoint,
+		pkScript txscript.PkScript) (*chainntnfs.SpendDetail, error)
+
+	// HasOutpointSpent queries the block to find a spending tx that spends
+	// the given outpoint. Returns the spend details if found, otherwise
+	// nil.
+	HasOutpointSpent(outpoint wire.OutPoint) *chainntnfs.SpendDetail
 }
 
 // Consumer defines a blockbeat consumer interface. Subsystems that need block

chainio/mocks.go

@@ -0,0 +1,70 @@
+package chainio
+
+import (
+	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/stretchr/testify/mock"
+)
+
+// MockBeat is a mock implementation of the Beat interface.
+type MockBeat struct {
+	mock.Mock
+}
+
+// Compile-time check to ensure MockBeat satisfies the Blockbeat interface.
+var _ Blockbeat = (*MockBeat)(nil)
+
+// Height returns the height of the block epoch.
+func (m *MockBeat) Height() int32 {
+	args := m.Called()
+
+	return args.Get(0).(int32)
+}
+
+func (m *MockBeat) NotifyBlockProcessed(err error, quitChan chan struct{}) {
+	m.Called(err, quitChan)
+}
+
+// DispatchSequential takes a list of consumers and notify them about the new
+// epoch sequentially.
+func (m *MockBeat) DispatchSequential(consumers []Consumer) error {
+	args := m.Called(consumers)
+
+	return args.Error(0)
+}
+
+// DispatchConcurrent notifies each consumer concurrently about the blockbeat.
+func (m *MockBeat) DispatchConcurrent(consumers []Consumer) error {
+	args := m.Called(consumers)
+
+	return args.Error(0)
+}
+
+// HasOutpointSpentByScript queries the block to find a spending tx that spends
+// the given outpoint using the pkScript.
+func (m *MockBeat) HasOutpointSpentByScript(outpoint wire.OutPoint,
+	pkScript txscript.PkScript) (*chainntnfs.SpendDetail, error) {
+
+	args := m.Called(outpoint, pkScript)
+
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(*chainntnfs.SpendDetail), args.Error(1)
+}
+
+// HasOutpointSpent queries the block to find a spending tx that spends the
+// given outpoint. Returns the spend details if found, otherwise nil.
+func (m *MockBeat) HasOutpointSpent(
+	outpoint wire.OutPoint) *chainntnfs.SpendDetail {
+
+	args := m.Called(outpoint)
+
+	if args.Get(0) == nil {
+		return nil
+	}
+
+	return args.Get(0).(*chainntnfs.SpendDetail)
+}

chanrestore.go

@@ -8,6 +8,7 @@ import (
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/chanbackup"
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/contractcourt"
@@ -286,6 +287,9 @@ func (c *chanDBRestorer) RestoreChansFromSingles(backups ...chanbackup.Single) e
 
 	ltndLog.Infof("Informing chain watchers of new restored channels")
 
+	// Create a slice of channel points.
+	chanPoints := make([]wire.OutPoint, 0, len(channelShells))
+
 	// Finally, we'll need to inform the chain arbitrator of these new
 	// channels so we'll properly watch for their ultimate closure on chain
 	// and sweep them via the DLP.
@@ -294,8 +298,15 @@ func (c *chanDBRestorer) RestoreChansFromSingles(backups ...chanbackup.Single) e
 		if err != nil {
 			return err
 		}
+
+		chanPoints = append(
+			chanPoints, restoredChannel.Chan.FundingOutpoint,
+		)
 	}
 
+	// With all the channels restored, we'll now re-send the blockbeat.
+	c.chainArb.RedispatchBlockbeat(chanPoints)
+
 	return nil
 }
 
@@ -314,7 +325,7 @@ func (s *server) ConnectPeer(nodePub *btcec.PublicKey, addrs []net.Addr) error {
 	// to ensure the new connection is created after this new link/channel
 	// is known.
 	if err := s.DisconnectPeer(nodePub); err != nil {
-		ltndLog.Infof("Peer(%v) is already connected, proceeding "+
+		ltndLog.Infof("Peer(%x) is already connected, proceeding "+
 			"with chan restore", nodePub.SerializeCompressed())
 	}