Improves/pending deposit

apps/bitcoin/common.go

@@ -106,7 +106,7 @@ func BlocksDuration(chain byte, count uint64) time.Duration {
 func CheckFeeRange(fvb int64, chain byte) bool {
 	switch chain {
 	case ChainBitcoin:
-		return fvb >= 2 && fvb <= 1000
+		return fvb >= 1 && fvb <= 1000
 	case ChainLitecoin:
 		return fvb >= 1 && fvb <= 20
 	default:

apps/bitcoin/coverage_test.go

@@ -128,7 +128,7 @@ func TestBitcoinAmountsTimingFeesAndMessages(t *testing.T) {
 	for _, fee := range []int64{2, 1_000} {
 		require.True(t, CheckFeeRange(fee, ChainBitcoin))
 	}
-	require.False(t, CheckFeeRange(1, ChainBitcoin))
+	require.True(t, CheckFeeRange(1, ChainBitcoin))
 	require.False(t, CheckFeeRange(1_001, ChainBitcoin))
 	for _, fee := range []int64{1, 20} {
 		require.True(t, CheckFeeRange(fee, ChainLitecoin))

apps/bitcoin/rpc.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"math/big"
@@ -14,6 +15,7 @@ import (
 
 	"github.com/MixinNetwork/mixin/logger"
 	"github.com/MixinNetwork/safe/util"
+	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/btcutil/v2"
 	"github.com/btcsuite/btcd/txscript/v2"
 	"github.com/shopspring/decimal"
@@ -73,7 +75,7 @@ func RPCGetTransactionOutput(chain byte, rpc, hash string, index int64) (*RPCTra
 		return nil, nil, fmt.Errorf("invalid output index %d", index)
 	}
 	tx, err := RPCGetTransaction(chain, rpc, hash)
-	if err != nil {
+	if err != nil || tx == nil {
 		return nil, nil, err
 	}
 	if int64(len(tx.Vout)) <= index {
@@ -155,6 +157,9 @@ func RPCGetTransactionSender(chain byte, rpc string, tx *RPCTransaction) (string
 	if err != nil {
 		return "", err
 	}
+	if itx == nil {
+		return "", fmt.Errorf("unknown transaction %s", tx.Vin[0].TxId)
+	}
 	index := tx.Vin[0].VOUT
 	if index < 0 || index >= int64(len(itx.Vout)) || itx.Vout[index] == nil || itx.Vout[index].ScriptPubKey == nil {
 		return "", fmt.Errorf("invalid previous output %s:%d", tx.Vin[0].TxId, index)
@@ -165,6 +170,10 @@ func RPCGetTransactionSender(chain byte, rpc string, tx *RPCTransaction) (string
 func RPCGetTransaction(chain byte, rpc, hash string) (*RPCTransaction, error) {
 	res, err := callBitcoinRPCUntilSufficient(rpc, "getrawtransaction", []any{hash, 1})
 	if err != nil {
+		var rpcErr *btcjson.RPCError
+		if errors.As(err, &rpcErr) && rpcErr.Code == btcjson.ErrRPCNoTxInfo {
+			return nil, nil
+		}
 		return nil, err
 	}
 	var tx RPCTransaction
@@ -176,13 +185,18 @@ func RPCGetTransaction(chain byte, rpc, hash string) (*RPCTransaction, error) {
 	return &tx, err
 }
 
-func RPCGetRawMempool(chain byte, rpc string) ([]*RPCTransaction, error) {
+func RPCGetRawMempoolHashes(rpc string) ([]string, error) {
 	res, err := callBitcoinRPCUntilSufficient(rpc, "getrawmempool", []any{})
 	if err != nil {
 		return nil, err
 	}
 	var txs []string
 	err = json.Unmarshal(res, &txs)
+	return txs, err
+}
+
+func RPCGetRawMempool(chain byte, rpc string) ([]*RPCTransaction, error) {
+	txs, err := RPCGetRawMempoolHashes(rpc)
 	if err != nil {
 		return nil, err
 	}
@@ -437,20 +451,20 @@ func callBitcoinRPC(rpc, method string, params []any) ([]byte, error) {
 		return nil, buildRPCError(rpc, method, params, err)
 	}
 	var result struct {
-		Data  any `json:"result"`
-		Error any `json:"error"`
+		Data  any               `json:"result"`
+		Error *btcjson.RPCError `json:"error"`
 	}
 	err = json.Unmarshal(body, &result)
 	if err != nil {
 		return nil, fmt.Errorf("%v (%s)", buildRPCError(rpc, method, params, err), string(body))
 	}
 	if result.Error != nil {
-		return nil, fmt.Errorf("%v (%s)", buildRPCError(rpc, method, params, err), string(body))
+		return nil, fmt.Errorf("%w (%s)", buildRPCError(rpc, method, params, result.Error), string(body))
 	}
 
 	return json.Marshal(result.Data)
 }
 
 func buildRPCError(rpc, method string, params []any, err error) error {
-	return fmt.Errorf("callBitcoinRPC(%s, %s, %v) => %v", rpc, method, params, err)
+	return fmt.Errorf("callBitcoinRPC(%s, %s, %v) => %w", rpc, method, params, err)
 }

apps/bitcoin/rpc_security_test.go

@@ -3,6 +3,7 @@
 import (
 	"encoding/hex"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -30,6 +31,52 @@
 	ID      json.RawMessage `json:"id"`
 }
 
+func TestBitcoinRPCGetTransactionNotFound(t *testing.T) {
+	for _, chain := range []byte{ChainBitcoin, ChainLitecoin} {
+		for _, status := range []int{http.StatusOK, http.StatusInternalServerError} {
+			t.Run(fmt.Sprintf("chain_%d_http_%d", chain, status), func(t *testing.T) {
+				server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					var request bitcoinRPCTestRequest
+					require.NoError(t, json.NewDecoder(r.Body).Decode(&request))
+					require.Equal(t, "getrawtransaction", request.Method)
+					w.Header().Set("Content-Type", "application/json")
+					w.WriteHeader(status)
+					require.NoError(t, json.NewEncoder(w).Encode(bitcoinRPCTestResponse{
+						JSONRPC: "2.0", ID: request.ID,
+						Error: map[string]any{"code": -5, "message": "No such mempool or blockchain transaction"},
+					}))
+				}))
+				t.Cleanup(server.Close)
+
+				tx, err := RPCGetTransaction(chain, server.URL, "missing-hash")
+				require.NoError(t, err)
+				require.Nil(t, tx)
+				tx, output, err := RPCGetTransactionOutput(chain, server.URL, "missing-hash", 0)
+				require.NoError(t, err)
+				require.Nil(t, tx)
+				require.Nil(t, output)
+				_, err = RPCGetTransactionSender(chain, server.URL, &RPCTransaction{
+					Vin: []*rpcIn{{TxId: "missing-parent", VOUT: 0}},
+				})
+				require.ErrorContains(t, err, "missing-parent")
+			})
+		}
+	}
+}
+
+func TestBitcoinRPCGetTransactionPreservesOtherErrors(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.NoError(t, json.NewEncoder(w).Encode(bitcoinRPCTestResponse{
+			JSONRPC: "2.0", Error: map[string]any{"code": -32601, "message": "method not found"},
+		}))
+	}))
+	t.Cleanup(server.Close)
+
+	tx, err := RPCGetTransaction(ChainBitcoin, server.URL, "hash")
+	require.Nil(t, tx)
+	require.ErrorContains(t, err, "method not found")
+}
+
 func TestBitcoinRPCClientParsesResultsAndRejectsErrors(t *testing.T) {
 	const (
 		blockHash       = "000000000000000000000000000000000000000000000000000000000000002a"
@@ -37,7 +84,7 @@
 	)
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		defer r.Body.Close()
 		var request bitcoinRPCTestRequest
 		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
 			t.Errorf("decode request: %v", err)

keeper/deposit.go

@@ -164,6 +164,9 @@ func (node *Node) doBitcoinHolderDeposit(ctx context.Context, req *common.Reques
 	if err != nil {
 		panic(fmt.Errorf("bitcoin.RPCTransaction(%s) => %v", deposit.Hash, err))
 	}
+	if btx == nil {
+		return node.failRequest(ctx, req, "")
+	}
 
 	amount := decimal.NewFromBigInt(deposit.Amount, -int32(asset.Decimals))
 	change, err := node.checkBitcoinChange(ctx, deposit, btx)

observer/bitcoin.go

@@ -380,17 +380,27 @@ func (node *Node) processBitcoinRPCBlock(ctx context.Context, chain byte, checkp
 	return node.store.WriteCache(ctx, key, "processed")
 }
 
-func (node *Node) bitcoinProcessTransaction(ctx context.Context, tx *bitcoin.RPCTransaction, chain byte) error {
-	for index := range tx.Vout {
-		out := tx.Vout[index]
+func bitcoinDepositOutputIndexes(tx *bitcoin.RPCTransaction) []int {
+	var indexes []int
+	for index, out := range tx.Vout {
+		if out.ScriptPubKey == nil {
+			continue
+		}
 		skt := out.ScriptPubKey.Type
 		if skt != bitcoin.ScriptPubKeyTypeWitnessScriptHash && skt != bitcoin.ScriptPubKeyTypeWitnessKeyHash {
 			continue
 		}
 		if out.N != int64(index) {
 			panic(tx.TxId)
 		}
+		indexes = append(indexes, index)
+	}
+	return indexes
+}
 
+func (node *Node) bitcoinProcessTransaction(ctx context.Context, tx *bitcoin.RPCTransaction, chain byte) error {
+	for _, index := range bitcoinDepositOutputIndexes(tx) {
+		out := tx.Vout[index]
 		err := node.bitcoinWritePendingDeposit(ctx, out.ScriptPubKey.Address, tx, out.N, out.Value, chain)
 		if err != nil {
 			panic(err)

observer/coverage_test.go

@@ -153,11 +153,11 @@ func TestCoverageObserverDepositsAndTransactions(t *testing.T) {
 
 	newRequestID := uuid.Must(uuid.NewV4()).String()
 	require.NoError(t, s.UpdateDepositRequestId(ctx, deposit.TransactionHash, deposit.OutputIndex, deposit.RequestId, newRequestID))
-	require.NoError(t, s.ConfirmPendingDeposit(ctx, deposit.TransactionHash, deposit.OutputIndex, newRequestID))
+	require.NoError(t, s.ConfirmPendingDeposit(ctx, deposit.TransactionHash, deposit.OutputIndex, newRequestID, deposit.Chain))
 	deposits, err = s.ListDeposits(ctx, int(deposit.Chain), deposit.Holder, common.RequestStateDone, 0)
 	require.NoError(t, err)
 	require.Len(t, deposits, 1)
-	require.Error(t, s.ConfirmPendingDeposit(ctx, deposit.TransactionHash, deposit.OutputIndex, newRequestID))
+	require.Error(t, s.ConfirmPendingDeposit(ctx, deposit.TransactionHash, deposit.OutputIndex, newRequestID, deposit.Chain))
 
 	sent := coverageObserverTransaction("sent", common.RequestStateDone, common.SafeChainBitcoin, now)
 	sent.SpentHash = sql.NullString{Valid: true, String: deposit.TransactionHash}

observer/ethereum.go

@@ -177,30 +177,34 @@ func (node *Node) processEthereumRPCBlock(ctx context.Context, num int64, chain
 }
 
 func (node *Node) doProcessEthereumRPCBlock(ctx context.Context, num int64, chain byte) error {
-	rpc, ethAssetId, _ := node.ethereumParams(chain)
-	blockTraces, err := ethereum.RPCDebugTraceBlockByNumber(rpc, num)
+	block, transfers, err := node.ethereumReadBlockTransfers(ctx, num, chain)
 	if err != nil {
 		return err
 	}
-	if len(blockTraces) == 0 {
-		return nil
+	return node.ethereumProcessBlock(ctx, chain, block, transfers)
+}
+
+func (node *Node) ethereumReadBlockTransfers(ctx context.Context, num int64, chain byte) (*ethereum.RPCBlockWithTransactions, []*ethereum.Transfer, error) {
+	rpc, ethAssetId, _ := node.ethereumParams(chain)
+	blockTraces, err := ethereum.RPCDebugTraceBlockByNumber(rpc, num)
+	if err != nil {
+		return nil, nil, err
 	}
 	block, err := ethereum.RPCGetBlockWithTransactions(rpc, num)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	erc20Transfers, err := ethereum.GetERC20TransferLogFromBlock(ctx, rpc, int64(chain), num)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 	transfers := ethereum.LoopBlockTraces(chain, ethAssetId, blockTraces, block)
 	transfers = append(transfers, erc20Transfers...)
-
-	return node.ethereumProcessBlock(ctx, chain, block, transfers)
+	return block, transfers, nil
 }
 
 func (node *Node) ethereumWritePendingDeposit(ctx context.Context, transfer *ethereum.Transfer, chain byte) error {
-	rpc, chainAssetId, params := node.ethereumParams(chain)
+	_, _, params := node.ethereumParams(chain)
 	minimum := decimal.RequireFromString(params.TransactionMinimum)
 	old, err := node.keeperStore.ReadDeposit(ctx, transfer.Hash, transfer.Index)
 	logger.Printf("keeperStore.ReadDeposit(%s, %d, %s, %s) => %v %v", transfer.Hash, transfer.Index, transfer.AssetId, transfer.Receiver, old, err)
@@ -232,19 +236,10 @@ func (node *Node) ethereumWritePendingDeposit(ctx context.Context, transfer *eth
 		return nil
 	}
 
-	var amount decimal.Decimal
-	switch transfer.AssetId {
-	case chainAssetId:
-		amount = decimal.NewFromBigInt(transfer.Value, -ethereum.ValuePrecision)
-	default:
-		asset, err := ethereum.FetchAsset(chain, rpc, transfer.TokenAddress)
-		logger.Printf("ethereum.FetchAsset(%s, %s) => %v %v", transfer.Hash, transfer.TokenAddress, asset, err)
-		if err != nil {
-			return nil
-		}
-		amount = decimal.NewFromBigInt(transfer.Value, -int32(asset.Decimals))
+	amount, err := node.ethereumDepositAmount(transfer, chain)
+	if err != nil {
+		return nil
 	}
-	amount = amount.RoundFloor(8)
 	if amount.Cmp(minimum) < 0 {
 		return nil
 	}
@@ -285,6 +280,20 @@ func (node *Node) ethereumWritePendingDeposit(ctx context.Context, transfer *eth
 	return nil
 }
 
+func (node *Node) ethereumDepositAmount(transfer *ethereum.Transfer, chain byte) (decimal.Decimal, error) {
+	rpc, chainAssetId, _ := node.ethereumParams(chain)
+	precision := int32(ethereum.ValuePrecision)
+	if transfer.AssetId != chainAssetId {
+		asset, err := ethereum.FetchAsset(chain, rpc, transfer.TokenAddress)
+		logger.Printf("ethereum.FetchAsset(%s, %s) => %v %v", transfer.Hash, transfer.TokenAddress, asset, err)
+		if err != nil {
+			return decimal.Zero, err
+		}
+		precision = int32(asset.Decimals)
+	}
+	return decimal.NewFromBigInt(transfer.Value, -precision).RoundFloor(8), nil
+}
+
 func (node *Node) ethereumConfirmPendingDeposit(ctx context.Context, deposit *Deposit) error {
 	rpc, ethereumAssetId, _ := node.ethereumParams(deposit.Chain)
 
@@ -378,7 +387,7 @@ func (node *Node) sendKeeperDepositTransaction(ctx context.Context, deposit *Dep
 	case common.RequestStateInitial:
 		return nil
 	case common.RequestStateDone:
-		err = node.store.ConfirmPendingDeposit(ctx, deposit.TransactionHash, deposit.OutputIndex, deposit.RequestId)
+		err = node.store.ConfirmPendingDeposit(ctx, deposit.TransactionHash, deposit.OutputIndex, deposit.RequestId, deposit.Chain)
 		if err != nil {
 			return fmt.Errorf("store.ConfirmPendingDeposit(%v) => %v", deposit, err)
 		}