Add stuck-transaction monitor for wallet transactions

pkg/bitcoin/chain.go

@@ -1,5 +1,7 @@
 package bitcoin
 
+import "context"
+
 // Chain defines an interface meant to be used for interaction with the
 // Bitcoin chain.
 type Chain interface {
@@ -11,7 +13,12 @@ type Chain interface {
 	// GetTransactionConfirmations gets the number of confirmations for the
 	// transaction with the given transaction hash. If the transaction with the
 	// given hash was not found on the chain, this function returns an error.
-	GetTransactionConfirmations(transactionHash Hash) (uint, error)
+	// The provided context bounds the lookup; implementations should abort the
+	// underlying call when it is cancelled.
+	GetTransactionConfirmations(
+		ctx context.Context,
+		transactionHash Hash,
+	) (uint, error)
 
 	// BroadcastTransaction broadcasts the given transaction over the
 	// network of the Bitcoin chain nodes. If the broadcast action could not be

pkg/bitcoin/chain_test.go

@@ -1,6 +1,7 @@
 package bitcoin
 
 import (
+	"context"
 	"fmt"
 	"sync"
 )
@@ -101,6 +102,7 @@ func (lc *localChain) GetTransactionMerkleProof(
 }
 
 func (lc *localChain) GetTransactionConfirmations(
+	_ context.Context,
 	transactionHash Hash,
 ) (uint, error) {
 	lc.transactionConfirmationsMutex.Lock()

pkg/bitcoin/electrum/electrum.go

@@ -83,6 +83,7 @@ func (c *Connection) GetTransaction(
 	txID := transactionHash.Hex(bitcoin.ReversedByteOrder)
 
 	rawTransaction, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(ctx context.Context, client *electrum.Client) (string, error) {
 			// We cannot use `GetTransaction` to get the the transaction details
@@ -130,6 +131,7 @@ func (c *Connection) GetTransaction(
 // transaction with the given transaction hash. If the transaction with the
 // given hash was not found on the chain, this function returns an error.
 func (c *Connection) GetTransactionConfirmations(
+	ctx context.Context,
 	transactionHash bitcoin.Hash,
 ) (uint, error) {
 	txID := transactionHash.Hex(bitcoin.ReversedByteOrder)
@@ -138,7 +140,16 @@ func (c *Connection) GetTransactionConfirmations(
 		zap.String("txID", txID),
 	)
 
+	// Bound the network calls below by both the caller's context and the
+	// connection lifetime context, so a cancelled caller aborts the lookup while
+	// connection shutdown still cancels it as before.
+	reqCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	stopOnParentDone := context.AfterFunc(c.parentCtx, cancel)
+	defer stopOnParentDone()
+
 	rawTransaction, err := requestWithRetry(
+		reqCtx,
 		c,
 		func(ctx context.Context, client *electrum.Client) (string, error) {
 			// We cannot use `GetTransaction` to get the transaction details
@@ -208,6 +219,7 @@ txOutLoop:
 		reversedScriptHashString := hex.EncodeToString(reversedScriptHash)
 
 		scriptHashHistory, err := requestWithRetry(
+			reqCtx,
 			c,
 			func(
 				ctx context.Context,
@@ -296,6 +308,7 @@ func (c *Connection) BroadcastTransaction(
 	var response string
 
 	response, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(ctx context.Context, client *electrum.Client) (string, error) {
 			return client.BroadcastTransaction(ctx, rawTx)
@@ -315,6 +328,7 @@ func (c *Connection) BroadcastTransaction(
 // latest block was not determined, this function returns an error.
 func (c *Connection) GetLatestBlockHeight() (uint, error) {
 	blockHeight, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(ctx context.Context, client *electrum.Client) (int32, error) {
 			tip, err := client.SubscribeHeadersSingle(ctx)
@@ -344,6 +358,7 @@ func (c *Connection) GetBlockHeader(
 	blockHeight uint,
 ) (*bitcoin.BlockHeader, error) {
 	getBlockHeaderResult, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(
 			ctx context.Context,
@@ -375,6 +390,7 @@ func (c *Connection) GetTransactionMerkleProof(
 	txID := transactionHash.Hex(bitcoin.ReversedByteOrder)
 
 	getMerkleProofResult, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(
 			ctx context.Context,
@@ -514,6 +530,7 @@ func (c *Connection) getConfirmedScriptHistory(
 	reversedScriptHashString := hex.EncodeToString(reversedScriptHash)
 
 	items, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(
 			ctx context.Context,
@@ -577,6 +594,7 @@ func (c *Connection) getConfirmedScriptHistory(
 // block height.
 func (c *Connection) GetCoinbaseTxHash(blockHeight uint) (bitcoin.Hash, error) {
 	txHashString, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(
 			ctx context.Context,
@@ -683,6 +701,7 @@ func (c *Connection) getScriptMempool(
 	reversedScriptHashString := hex.EncodeToString(reversedScriptHash)
 
 	items, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(
 			ctx context.Context,
@@ -885,6 +904,7 @@ func (c *Connection) getScriptUtxos(
 	reversedScriptHashString := hex.EncodeToString(reversedScriptHash)
 
 	items, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(
 			ctx context.Context,
@@ -1173,6 +1193,7 @@ func (c *Connection) verifyServer() error {
 	}
 
 	server, err := requestWithRetry(
+		c.parentCtx,
 		c,
 		func(ctx context.Context, client *electrum.Client) (*Server, error) {
 			serverVersion, protocolVersion, err := client.ServerVersion(ctx)
@@ -1213,6 +1234,7 @@ func (c *Connection) keepAlive() {
 		select {
 		case <-ticker.C:
 			_, err := requestWithRetry(
+				c.parentCtx,
 				c,
 				func(ctx context.Context, client *electrum.Client) (interface{}, error) {
 					return nil, client.Ping(ctx)
@@ -1265,6 +1287,7 @@ func connectWithRetry(
 }
 
 func requestWithRetry[K interface{}](
+	parentCtx context.Context,
 	c *Connection,
 	requestFn func(ctx context.Context, client *electrum.Client) (K, error),
 	requestName string,
@@ -1275,7 +1298,7 @@ func requestWithRetry[K interface{}](
 	var result K
 
 	err := wrappers.DoWithDefaultRetry(
-		c.parentCtx,
+		parentCtx,
 		c.config.RequestRetryTimeout,
 		func(ctx context.Context) error {
 			if err := c.reconnectIfShutdown(); err != nil {

pkg/bitcoin/electrum/electrum_integration_test.go

@@ -230,7 +230,7 @@ func TestGetTransactionConfirmations_Integration(t *testing.T) {
 				}
 				expectedConfirmations := latestBlockHeight - tx.BlockHeight
 
-				result, err := electrum.GetTransactionConfirmations(tx.TxHash)
+				result, err := electrum.GetTransactionConfirmations(context.Background(), tx.TxHash)
 				if err != nil {
 					t.Fatal(err)
 				}
@@ -252,7 +252,7 @@ func TestGetTransactionConfirmations_Negative_Integration(t *testing.T) {
 		electrum, cancelCtx := newTestConnection(t, testConfig.clientConfig)
 		defer cancelCtx()
 
-		_, err := electrum.GetTransactionConfirmations(invalidTxID)
+		_, err := electrum.GetTransactionConfirmations(context.Background(), invalidTxID)
 		if shouldSkipElectrumIntegrationError(err) {
 			t.Skipf("skipping due to transient electrum error: %v", err)
 		}

pkg/bitcoin/spv_proof.go

@@ -2,6 +2,7 @@ package bitcoin
 
 import (
 	"bytes"
+	"context"
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
@@ -39,6 +40,7 @@ func AssembleSpvProof(
 	btcChain Chain,
 ) (*Transaction, *SpvProof, error) {
 	confirmations, err := btcChain.GetTransactionConfirmations(
+		context.Background(),
 		transactionHash,
 	)
 	if err != nil {

pkg/clientinfo/performance.go

@@ -123,6 +123,8 @@ func (pm *PerformanceMetrics) registerAllMetrics() {
 		MetricWalletActionSuccessTotal,
 		MetricWalletActionFailedTotal,
 		MetricWalletHeartbeatFailuresTotal,
+		MetricStuckWalletTransactionsTotal,
+		MetricUnmonitoredWalletTransactionsTotal,
 		MetricCoordinationWindowsDetectedTotal,
 		MetricCoordinationProceduresExecutedTotal,
 		MetricCoordinationFailedTotal,
@@ -635,11 +637,13 @@ const (
 	MetricRedemptionProofSubmissionsFailedTotal  = "redemption_proof_submissions_failed_total"
 
 	// Wallet Action Metrics (aggregate)
-	MetricWalletActionsTotal           = "wallet_actions_total"
-	MetricWalletActionSuccessTotal     = "wallet_action_success_total"
-	MetricWalletActionFailedTotal      = "wallet_action_failed_total"
-	MetricWalletActionDurationSeconds  = "wallet_action_duration_seconds"
-	MetricWalletHeartbeatFailuresTotal = "wallet_heartbeat_failures_total"
+	MetricWalletActionsTotal                 = "wallet_actions_total"
+	MetricWalletActionSuccessTotal           = "wallet_action_success_total"
+	MetricWalletActionFailedTotal            = "wallet_action_failed_total"
+	MetricWalletActionDurationSeconds        = "wallet_action_duration_seconds"
+	MetricWalletHeartbeatFailuresTotal       = "wallet_heartbeat_failures_total"
+	MetricStuckWalletTransactionsTotal       = "stuck_wallet_transactions_total"
+	MetricUnmonitoredWalletTransactionsTotal = "unmonitored_wallet_transactions_total"
 
 	// Wallet Action Metrics (per-action type)
 	// These are generated dynamically using WalletActionMetricName helper function

pkg/clientinfo/rpc_health_test.go

@@ -59,7 +59,7 @@ func (f *fakeBitcoinChain) GetBlockHeader(blockHeight uint) (*bitcoin.BlockHeade
 func (f *fakeBitcoinChain) GetTransaction(transactionHash bitcoin.Hash) (*bitcoin.Transaction, error) {
 	panic("not needed in rpc_health tests")
 }
-func (f *fakeBitcoinChain) GetTransactionConfirmations(transactionHash bitcoin.Hash) (uint, error) {
+func (f *fakeBitcoinChain) GetTransactionConfirmations(_ context.Context, transactionHash bitcoin.Hash) (uint, error) {
 	panic("not needed in rpc_health tests")
 }
 func (f *fakeBitcoinChain) BroadcastTransaction(transaction *bitcoin.Transaction) error {

pkg/maintainer/btcdiff/bitcoin_chain_test.go

@@ -1,6 +1,7 @@
 package btcdiff
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/keep-network/keep-core/pkg/bitcoin"
@@ -26,6 +27,7 @@ func (lbc *localBitcoinChain) GetTransaction(
 // transaction with the given transaction hash. If the transaction with the
 // given hash was not found on the chain, this function returns an error.
 func (lbc *localBitcoinChain) GetTransactionConfirmations(
+	_ context.Context,
 	transactionHash bitcoin.Hash,
 ) (uint, error) {
 	panic("unsupported")

pkg/maintainer/spv/bitcoin_chain_test.go

@@ -2,6 +2,7 @@ package spv
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"math/big"
 	"sync"
@@ -72,7 +73,7 @@ func (lbc *localBitcoinChain) GetTransaction(transactionHash bitcoin.Hash) (
 	return nil, fmt.Errorf("transaction not found")
 }
 
-func (lbc *localBitcoinChain) GetTransactionConfirmations(transactionHash bitcoin.Hash) (
+func (lbc *localBitcoinChain) GetTransactionConfirmations(_ context.Context, transactionHash bitcoin.Hash) (
 	uint,
 	error,
 ) {

pkg/maintainer/spv/spv.go

@@ -337,6 +337,7 @@ func getProofInfo(
 	}
 
 	accumulatedConfirmations, err := btcChain.GetTransactionConfirmations(
+		context.Background(),
 		transactionHash,
 	)
 	if err != nil {

pkg/tbtc/bitcoin_chain_test.go

@@ -2,6 +2,7 @@ package tbtc
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"sync"
 
@@ -39,6 +40,7 @@ func (lbc *localBitcoinChain) GetTransaction(
 }
 
 func (lbc *localBitcoinChain) GetTransactionConfirmations(
+	_ context.Context,
 	transactionHash bitcoin.Hash,
 ) (uint, error) {
 	for index, transaction := range lbc.transactions {

pkg/tbtc/deposit_sweep.go

@@ -1,6 +1,7 @@
 package tbtc
 
 import (
+	"context"
 	"crypto/ecdsa"
 	"fmt"
 	"math/big"
@@ -105,6 +106,7 @@ func newDepositSweepAction(
 	proposalProcessingStartBlock uint64,
 	proposalExpiryBlock uint64,
 	waitForBlockFn waitForBlockFn,
+	transactionMonitor *transactionMonitor,
 ) *depositSweepAction {
 	transactionExecutor := newWalletTransactionExecutor(
 		btcChain,
@@ -113,6 +115,8 @@ func newDepositSweepAction(
 		waitForBlockFn,
 	)
 
+	transactionExecutor.setTransactionMonitor(transactionMonitor)
+
 	return &depositSweepAction{
 		logger:                           logger,
 		chain:                            chain,
@@ -340,6 +344,7 @@ func ValidateDepositSweepProposal(
 		)
 
 		confirmations, err := btcChain.GetTransactionConfirmations(
+			context.Background(),
 			depositKey.FundingTxHash,
 		)
 		if err != nil {

pkg/tbtc/deposit_sweep_test.go

@@ -205,6 +205,7 @@ func TestDepositSweepAction_Execute(t *testing.T) {
 				func(ctx context.Context, blockHeight uint64) error {
 					return nil
 				},
+				nil,
 			)
 
 			// Modify the default parameters of the action to make

pkg/tbtc/moved_funds_sweep.go

@@ -107,6 +107,7 @@ func newMovedFundsSweepAction(
 	proposalProcessingStartBlock uint64,
 	proposalExpiryBlock uint64,
 	waitForBlockFn waitForBlockFn,
+	transactionMonitor *transactionMonitor,
 ) *movedFundsSweepAction {
 	transactionExecutor := newWalletTransactionExecutor(
 		btcChain,
@@ -115,6 +116,8 @@ func newMovedFundsSweepAction(
 		waitForBlockFn,
 	)
 
+	transactionExecutor.setTransactionMonitor(transactionMonitor)
+
 	return &movedFundsSweepAction{
 		logger:                           logger,
 		chain:                            chain,

pkg/tbtc/moved_funds_sweep_test.go

@@ -112,6 +112,7 @@ func TestMovedFundsSweepAction_Execute(t *testing.T) {
 				func(ctx context.Context, blockHeight uint64) error {
 					return nil
 				},
+				nil,
 			)
 
 			// Modify the default parameters of the action to make

pkg/tbtc/moving_funds.go

@@ -103,6 +103,7 @@ func newMovingFundsAction(
 	proposalProcessingStartBlock uint64,
 	proposalExpiryBlock uint64,
 	waitForBlockFn waitForBlockFn,
+	transactionMonitor *transactionMonitor,
 ) *movingFundsAction {
 	transactionExecutor := newWalletTransactionExecutor(
 		btcChain,
@@ -111,6 +112,8 @@ func newMovingFundsAction(
 		waitForBlockFn,
 	)
 
+	transactionExecutor.setTransactionMonitor(transactionMonitor)
+
 	return &movingFundsAction{
 		logger:                           logger,
 		chain:                            chain,

pkg/tbtc/moving_funds_test.go

@@ -123,6 +123,7 @@ func TestMovingFundsAction_Execute(t *testing.T) {
 				func(ctx context.Context, blockHeight uint64) error {
 					return nil
 				},
+				nil,
 			)
 
 			// Modify the default parameters of the action to make