diff --git a/apps/bitcoin/common.go b/apps/bitcoin/common.go index cab12c39..41044791 100644 --- a/apps/bitcoin/common.go +++ b/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: diff --git a/apps/bitcoin/coverage_test.go b/apps/bitcoin/coverage_test.go index a4e99969..8939f06b 100644 --- a/apps/bitcoin/coverage_test.go +++ b/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)) diff --git a/apps/bitcoin/rpc.go b/apps/bitcoin/rpc.go index 01f433e9..49578320 100644 --- a/apps/bitcoin/rpc.go +++ b/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" @@ -76,6 +78,9 @@ func RPCGetTransactionOutput(chain byte, rpc, hash string, index int64) (*RPCTra if err != nil { return nil, nil, err } + if tx == nil { + return nil, nil, fmt.Errorf("transaction not found: %s", hash) + } if int64(len(tx.Vout)) <= index { return nil, nil, nil } @@ -155,6 +160,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 +173,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 +188,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 +454,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) } diff --git a/apps/bitcoin/rpc_security_test.go b/apps/bitcoin/rpc_security_test.go index 3e69f1a2..53733845 100644 --- a/apps/bitcoin/rpc_security_test.go +++ b/apps/bitcoin/rpc_security_test.go @@ -3,6 +3,7 @@ package bitcoin import ( "encoding/hex" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -30,6 +31,56 @@ type bitcoinRPCTestResponse struct { 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.ErrorContains(t, err, "transaction not found: missing-hash") + 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") + tx, output, err := RPCGetTransactionOutput(ChainBitcoin, server.URL, "hash", 0) + require.Nil(t, tx) + require.Nil(t, output) + require.ErrorContains(t, err, "method not found") +} + func TestBitcoinRPCClientParsesResultsAndRejectsErrors(t *testing.T) { const ( blockHash = "000000000000000000000000000000000000000000000000000000000000002a" diff --git a/apps/ethereum/rpc.go b/apps/ethereum/rpc.go index 3f7fa6dd..9a94f54d 100644 --- a/apps/ethereum/rpc.go +++ b/apps/ethereum/rpc.go @@ -63,6 +63,7 @@ type RPCTransaction struct { type RPCBlockCallTrace struct { Result *RPCTransactionCallTrace `json:"result"` + TxHash string `json:"txHash"` } type RPCTransactionCallTrace struct { Calls []*RPCTransactionCallTrace `json:"calls"` @@ -245,6 +246,26 @@ func RPCDebugTraceBlockByNumber(rpc string, height int64) ([]*RPCBlockCallTrace, return txs, err } +func RPCDebugTraceBlockByHash(rpc, hash string) ([]*RPCBlockCallTrace, error) { + res, err := callEthereumRPCUntilSufficient(rpc, "debug_traceBlockByHash", []any{hash, map[string]any{"tracer": "callTracer"}}) + if err != nil { + return nil, err + } + var traces []*RPCBlockCallTrace + if err := json.Unmarshal(res, &traces); err != nil { + return nil, err + } + if traces == nil { + return nil, fmt.Errorf("missing block traces %s", hash) + } + for _, trace := range traces { + if trace == nil || trace.Result == nil { + return nil, fmt.Errorf("missing transaction trace in block %s", hash) + } + } + return traces, nil +} + func RPCGetAssetBalanceAtBlock(rpc, address, asset string, blockNumber uint64) (*big.Int, error) { if asset == EthereumEmptyAddress { return rpcGetEtherBalanceAtBlock(rpc, address, blockNumber) @@ -298,6 +319,17 @@ func rpcGetTokenBalanceAtBlock(rpc, address, tokenAddress string, blockNumber ui } func GetERC20TransferLogFromBlock(ctx context.Context, rpc string, chain, height int64) ([]*Transfer, error) { + return getERC20TransferLogs(ctx, rpc, chain, ethereum.FilterQuery{ + FromBlock: big.NewInt(height), ToBlock: big.NewInt(height), + }) +} + +func GetERC20TransferLogFromBlockHash(ctx context.Context, rpc string, chain int64, hash string) ([]*Transfer, error) { + blockHash := common.HexToHash(hash) + return getERC20TransferLogs(ctx, rpc, chain, ethereum.FilterQuery{BlockHash: &blockHash}) +} + +func getERC20TransferLogs(ctx context.Context, rpc string, chain int64, query ethereum.FilterQuery) ([]*Transfer, error) { client, err := ethclient.Dial(rpc) if err != nil { return nil, err @@ -306,11 +338,7 @@ func GetERC20TransferLogFromBlock(ctx context.Context, rpc string, chain, height logTransferSig := []byte("Transfer(address,address,uint256)") logTransferSigHash := crypto.Keccak256Hash(logTransferSig) - query := ethereum.FilterQuery{ - FromBlock: big.NewInt(height), - ToBlock: big.NewInt(height), - Topics: [][]common.Hash{{logTransferSigHash}}, - } + query.Topics = [][]common.Hash{{logTransferSigHash}} contractAbi, err := ga.JSON(strings.NewReader(abi.AssetABI)) if err != nil { log.Fatal(err) @@ -322,6 +350,9 @@ func GetERC20TransferLogFromBlock(ctx context.Context, rpc string, chain, height ts := []*Transfer{} for _, vLog := range logs { + if query.BlockHash != nil && (vLog.BlockHash != *query.BlockHash || vLog.Removed) { + return nil, fmt.Errorf("log does not belong to block %s", query.BlockHash.Hex()) + } switch { case len(vLog.Topics) == 3 && vLog.Topics[0].Hex() == logTransferSigHash.Hex() && len(vLog.Data) == 32: var event abi.AssetTransfer diff --git a/keeper/deposit.go b/keeper/deposit.go index d12147c6..b23a6013 100644 --- a/keeper/deposit.go +++ b/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) diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 63c7d0b0..69e287bb 100644 --- a/observer/bitcoin.go +++ b/observer/bitcoin.go @@ -222,6 +222,7 @@ func (node *Node) bitcoinWritePendingDeposit(ctx context.Context, receiver strin State: common.RequestStateInitial, Chain: chain, RequestId: id, + IsChange: change, CreatedAt: createdAt, UpdatedAt: createdAt, } @@ -380,9 +381,12 @@ 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 @@ -390,7 +394,14 @@ func (node *Node) bitcoinProcessTransaction(ctx context.Context, tx *bitcoin.RPC 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) diff --git a/observer/coverage_test.go b/observer/coverage_test.go index 74df631b..25f4a506 100644 --- a/observer/coverage_test.go +++ b/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} @@ -525,7 +525,7 @@ func TestCoverageObserverPureHelpersAndCheckpoints(t *testing.T) { require.Equal(t, "0102", views[0].Script) recoveryViews := node.viewRecoveries(ctx, []*Recovery{{Address: "address", State: common.RequestStateDone}}) require.Equal(t, "done", recoveryViews[0]["state"]) - depositViews := node.viewDeposits(ctx, []*Deposit{{TransactionHash: "hash", State: common.RequestStateInitial, Chain: common.SafeChainEthereum}}, nil) + depositViews := node.viewDeposits([]*Deposit{{TransactionHash: "hash", State: common.RequestStateInitial, Chain: common.SafeChainEthereum}}, nil) require.Equal(t, "pending", depositViews[0]["state"]) require.Equal(t, "hash", depositViews[0]["sent_hash"]) } diff --git a/observer/ethereum.go b/observer/ethereum.go index 5b0cd2de..70829f2c 100644 --- a/observer/ethereum.go +++ b/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) } diff --git a/observer/http.go b/observer/http.go index 17af49db..359098a3 100644 --- a/observer/http.go +++ b/observer/http.go @@ -104,6 +104,7 @@ func (node *Node) StartHTTP(version, readme string) { router.GET("/chains", node.httpListChains) router.GET("/signers", node.httpListSigners) router.GET("/keepers", node.httpListKeepers) + router.GET("/pending_deposits", node.httpListPendingDeposits) router.GET("/deposits", node.httpListDeposits) router.GET("/recoveries", node.httpListRecoveries) router.GET("/recoveries/:id", node.httpGetRecovery) @@ -260,6 +261,24 @@ func (node *Node) httpListNodes(w http.ResponseWriter, r *http.Request, typ stri common.RenderJSON(w, r, http.StatusOK, ns) } +func (node *Node) httpListPendingDeposits(w http.ResponseWriter, r *http.Request, params map[string]string) { + chain, _ := strconv.ParseInt(r.URL.Query().Get("chain"), 10, 32) + switch chain { + case common.SafeChainBitcoin, common.SafeChainLitecoin, common.SafeChainEthereum, common.SafeChainPolygon: + default: + common.RenderJSON(w, r, http.StatusBadRequest, map[string]any{"error": "chain"}) + return + } + holder := r.URL.Query().Get("holder") + offset, _ := strconv.ParseInt(r.URL.Query().Get("offset"), 10, 64) + deposits, err := node.listPendingDeposits(r.Context(), byte(chain), holder, offset) + if err != nil { + common.RenderError(w, r, err) + return + } + common.RenderJSON(w, r, http.StatusOK, deposits) +} + func (node *Node) httpListDeposits(w http.ResponseWriter, r *http.Request, params map[string]string) { holder := r.URL.Query().Get("holder") chain, _ := strconv.ParseInt(r.URL.Query().Get("chain"), 10, 32) @@ -268,7 +287,7 @@ func (node *Node) httpListDeposits(w http.ResponseWriter, r *http.Request, param return } offset, _ := strconv.ParseInt(r.URL.Query().Get("offset"), 10, 64) - deposits, err := node.store.ListDeposits(r.Context(), int(chain), holder, 0, offset) + deposits, err := node.store.listDeposits(r.Context(), int(chain), holder, common.RequestStateDone, offset, true) if err != nil { common.RenderError(w, r, err) return @@ -279,7 +298,7 @@ func (node *Node) httpListDeposits(w http.ResponseWriter, r *http.Request, param return } - common.RenderJSON(w, r, http.StatusOK, node.viewDeposits(r.Context(), deposits, sent)) + common.RenderJSON(w, r, http.StatusOK, node.viewDeposits(deposits, sent)) } func (node *Node) httpListRecoveries(w http.ResponseWriter, r *http.Request, params map[string]string) { @@ -847,7 +866,7 @@ func (node *Node) listPendingBitcoinUTXOsForHolder(ctx context.Context, holder s return node.keeperStore.ListPendingBitcoinUTXOsForHolder(ctx, holder) } -func (node *Node) viewDeposits(ctx context.Context, deposits []*Deposit, sent map[string]string) []map[string]any { +func (node *Node) viewDeposits(deposits []*Deposit, sent map[string]string) []map[string]any { view := make([]map[string]any, 0) for _, d := range deposits { state := "done" @@ -863,18 +882,13 @@ func (node *Node) viewDeposits(ctx context.Context, deposits []*Deposit, sent ma "receiver": d.Receiver, "sent_hash": sent[d.TransactionHash], "chain": d.Chain, - "change": false, + "change": d.IsChange, "state": state, "updated_at": d.UpdatedAt, "created_at": d.CreatedAt, } if dm["sent_hash"] == "" { dm["sent_hash"] = d.TransactionHash - } else { - switch d.Chain { - case common.SafeChainBitcoin, common.SafeChainLitecoin: - dm["change"] = node.bitcoinCheckDepositChange(ctx, d.TransactionHash, d.OutputIndex, sent[d.TransactionHash]) - } } view = append(view, dm) } diff --git a/observer/migration.go b/observer/migration.go new file mode 100644 index 00000000..ac16deee --- /dev/null +++ b/observer/migration.go @@ -0,0 +1,130 @@ +package observer + +import ( + "context" + "database/sql" + "fmt" + "time" + + "github.com/MixinNetwork/safe/common" +) + +const depositChangeMigrationKey = "migration-deposits-is-change" + +// Migrate prepares observer data before HTTP and background workers start. +func (node *Node) Migrate(ctx context.Context) error { + if err := node.store.migratePendingDepositCheckpoints(ctx); err != nil { + return err + } + return node.migrateDepositChanges(ctx) +} + +func (node *Node) migrateDepositChanges(ctx context.Context) error { + s := node.store + s.mutex.Lock() + defer s.mutex.Unlock() + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer common.Rollback(tx) + + exists, err := s.checkExistence(ctx, tx, "SELECT value FROM properties WHERE key=?", depositChangeMigrationKey) + if err != nil || exists { + return err + } + + _, err = tx.ExecContext(ctx, "ALTER TABLE deposits ADD COLUMN is_change BOOLEAN NOT NULL DEFAULT FALSE") + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, "CREATE INDEX IF NOT EXISTS deposits_by_chain_created ON deposits(chain, created_at)") + if err != nil { + return err + } + // EVM deposits retain the default false value. + for _, chain := range []byte{common.SafeChainBitcoin, common.SafeChainLitecoin} { + for offset := int64(0); ; { + batch, err := s.listDepositsByCreatedAt(ctx, tx, int(chain), "", 0, offset, false) + if err != nil { + return err + } + sent, err := s.queryDepositSentHashes(ctx, tx, batch) + if err != nil { + return err + } + for _, d := range batch { + change := node.bitcoinCheckDepositChange(ctx, d.TransactionHash, d.OutputIndex, sent[d.TransactionHash]) + err := s.execOne(ctx, tx, "UPDATE deposits SET is_change=? WHERE transaction_hash=? AND output_index=?", change, d.TransactionHash, d.OutputIndex) + if err != nil { + return fmt.Errorf("backfill deposit change %s:%d: %w", d.TransactionHash, d.OutputIndex, err) + } + } + if len(batch) < 100 { + break + } + offset = batch[len(batch)-1].CreatedAt.UnixNano() + 1 + } + } + + for _, query := range []string{ + "CREATE INDEX IF NOT EXISTS deposits_by_chain_state_change_created ON deposits(chain, state, is_change, created_at)", + "CREATE INDEX IF NOT EXISTS deposits_by_chain_state_change_updated ON deposits(chain, state, is_change, updated_at)", + "CREATE INDEX IF NOT EXISTS deposits_by_holder_chain_state_change_created ON deposits(holder, chain, state, is_change, created_at)", + "CREATE INDEX IF NOT EXISTS deposits_by_holder_chain_state_change_updated ON deposits(holder, chain, state, is_change, updated_at)", + } { + if _, err := tx.ExecContext(ctx, query); err != nil { + return err + } + } + if err := s.writeProperty(ctx, tx, depositChangeMigrationKey, "done"); err != nil { + return err + } + return tx.Commit() +} + +// Run before either scanner starts so the pending cursor inherits the old cursor once. +func (s *SQLite3Store) migratePendingDepositCheckpoints(ctx context.Context) error { + s.mutex.Lock() + defer s.mutex.Unlock() + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer common.Rollback(tx) + + ethereumExists, err := s.checkExistence(ctx, tx, "SELECT key FROM properties WHERE key=?", pendingDepositCheckpointKey(common.SafeChainEthereum)) + if err != nil { + return err + } + polygonExists, err := s.checkExistence(ctx, tx, "SELECT key FROM properties WHERE key=?", pendingDepositCheckpointKey(common.SafeChainPolygon)) + if err != nil { + return err + } + if ethereumExists && polygonExists { + return nil + } + + cols := []string{"key", "value", "created_at", "updated_at"} + now := time.Now().UTC() + for _, chain := range []byte{common.SafeChainEthereum, common.SafeChainPolygon} { + if (chain == common.SafeChainEthereum && ethereumExists) || (chain == common.SafeChainPolygon && polygonExists) { + continue + } + var value string + err := tx.QueryRowContext(ctx, "SELECT value FROM properties WHERE key=?", depositCheckpointKey(chain)).Scan(&value) + if err == sql.ErrNoRows { + continue + } else if err != nil { + return err + } + err = s.execOne(ctx, tx, buildInsertionSQL("properties", cols), pendingDepositCheckpointKey(chain), value, now, now) + if err != nil { + return err + } + } + + return tx.Commit() +} diff --git a/observer/node.go b/observer/node.go index b8874103..995ceae4 100644 --- a/observer/node.go +++ b/observer/node.go @@ -57,6 +57,11 @@ func NewNode(db *SQLite3Store, kd *store.SQLite3Store, conf *Configuration, keep } func (node *Node) Boot(ctx context.Context) { + err := node.Migrate(ctx) + if err != nil { + panic(err) + } + for _, chain := range []byte{ common.SafeChainBitcoin, common.SafeChainLitecoin, @@ -72,12 +77,14 @@ func (node *Node) Boot(ctx context.Context) { case common.SafeChainBitcoin, common.SafeChainLitecoin: go node.bitcoinNetworkInfoLoop(ctx, chain) go node.bitcoinRPCBlocksLoop(ctx, chain) + go node.bitcoinPendingDepositsLoop(ctx, chain) go node.bitcoinDepositConfirmLoop(ctx, chain) go node.bitcoinTransactionApprovalLoop(ctx, chain) go node.bitcoinTransactionSpendLoop(ctx, chain) case common.SafeChainPolygon, common.SafeChainEthereum: go node.ethereumNetworkInfoLoop(ctx, chain) go node.ethereumRPCBlocksLoop(ctx, chain) + go node.ethereumPendingDepositsLoop(ctx, chain) go node.ethereumDepositConfirmLoop(ctx, chain) go node.ethereumTransactionApprovalLoop(ctx, chain) go node.ethereumTransactionSpendLoop(ctx, chain) @@ -440,6 +447,18 @@ func (node *Node) readDepositCheckpoint(ctx context.Context, chain byte) (int64, return checkpoint, nil } +func (node *Node) readPendingDepositCheckpoint(ctx context.Context, chain byte) (int64, error) { + value, err := node.store.ReadProperty(ctx, pendingDepositCheckpointKey(chain)) + if err != nil || value == "" { + return depositCheckpointDefault(chain), err + } + checkpoint, err := strconv.ParseInt(value, 10, 64) + if err != nil { + return 0, err + } + return max(checkpoint, depositCheckpointDefault(chain)), nil +} + func (node *Node) getStuckTxHash(ctx context.Context, hash string) string { tx, err := node.keeperStore.ReadTransaction(ctx, hash) if err != nil { @@ -509,6 +528,15 @@ func depositCheckpointKey(chain byte) string { } } +func pendingDepositCheckpointKey(chain byte) string { + switch chain { + case common.SafeChainEthereum, common.SafeChainPolygon: + return fmt.Sprintf("ethereum-pending-deposit-checkpoint-%d", chain) + default: + panic(chain) + } +} + func (node *Node) safeTraceId(params ...string) string { traceId := common.UniqueId(node.conf.PrivateKey, node.conf.PrivateKey) for _, id := range params { diff --git a/observer/observer_test.go b/observer/observer_test.go index be120553..4066670b 100644 --- a/observer/observer_test.go +++ b/observer/observer_test.go @@ -65,7 +65,7 @@ func TestObserver(t *testing.T) { fvb, err := bitcoin.EstimateAvgFee(common.SafeChainBitcoin, node.conf.BitcoinRPC) require.Nil(err) - require.GreaterOrEqual(fvb, int64(5)) + require.GreaterOrEqual(fvb, int64(1)) require.Less(fvb, int64(500)) now := time.Now().UTC() diff --git a/observer/pending.go b/observer/pending.go new file mode 100644 index 00000000..5634d7cf --- /dev/null +++ b/observer/pending.go @@ -0,0 +1,427 @@ +package observer + +import ( + "context" + "database/sql" + "fmt" + "sort" + "time" + + "github.com/MixinNetwork/mixin/logger" + "github.com/MixinNetwork/safe/apps/bitcoin" + "github.com/MixinNetwork/safe/apps/ethereum" + "github.com/MixinNetwork/safe/common" + "github.com/shopspring/decimal" +) + +const bitcoinPendingDepositGracePeriod = time.Minute * 5 +const ethereumPendingDepositTTL = 24 * time.Hour + +func (node *Node) listPendingDeposits(ctx context.Context, chain byte, holder string, offset int64) ([]*PendingDeposit, error) { + deposits, err := node.store.ListPendingDeposits(ctx, int(chain), holder, offset) + if err != nil { + return nil, err + } + type depositKey struct { + hash string + index int64 + } + seen := make(map[depositKey]bool, len(deposits)) + for _, d := range deposits { + seen[depositKey{d.TransactionHash, d.OutputIndex}] = true + } + + // The first 100 rows of each source contain the first 100 of their deduplicated union. + initial, err := node.store.listDepositsByCreatedAt(ctx, nil, int(chain), holder, common.RequestStateInitial, offset, true) + if err != nil { + return nil, err + } + for _, d := range initial { + key := depositKey{d.TransactionHash, d.OutputIndex} + if seen[key] { + continue + } + deposits = append(deposits, &PendingDeposit{ + TransactionHash: d.TransactionHash, OutputIndex: d.OutputIndex, + AssetId: d.AssetId, AssetAddress: d.AssetAddress, Amount: d.Amount, + Receiver: d.Receiver, Sender: d.Sender, State: d.State, + Chain: d.Chain, Holder: d.Holder, CreatedAt: d.CreatedAt, + }) + seen[key] = true + } + sort.Slice(deposits, func(i, j int) bool { + a, b := deposits[i], deposits[j] + if !a.CreatedAt.Equal(b.CreatedAt) { + return a.CreatedAt.Before(b.CreatedAt) + } + if a.TransactionHash != b.TransactionHash { + return a.TransactionHash < b.TransactionHash + } + return a.OutputIndex < b.OutputIndex + }) + if len(deposits) > 100 { + deposits = deposits[:100] + } + return deposits, nil +} + +func (node *Node) bitcoinPendingDepositsLoop(ctx context.Context, chain byte) { + seen := make(map[string]bool) + for { + time.Sleep(5 * time.Second) + err := node.scanBitcoinPendingDeposits(ctx, chain, seen) + logger.Printf("node.scanBitcoinPendingDeposits(%d) => %v", chain, err) + } +} + +func (node *Node) scanBitcoinPendingDeposits(ctx context.Context, chain byte, seen map[string]bool) error { + rpc, _, _ := node.bitcoinParams(chain) + hashes, err := bitcoin.RPCGetRawMempoolHashes(rpc) + logger.Printf("bitcoin.RPCGetRawMempoolHashes(%d) => %d %v", chain, len(hashes), err) + if err != nil { + return fmt.Errorf("bitcoin.RPCGetRawMempoolHashes(%d) => %w", chain, err) + } + mempool := make(map[string]bool, len(hashes)) + for _, hash := range hashes { + mempool[hash] = true + } + for hash := range seen { + if !mempool[hash] { + delete(seen, hash) + } + } + + // Use persisted observations, including those recorded before a restart. + err = node.reconcileBitcoinPendingDeposits(ctx, chain, mempool, time.Now().UTC()) + if err != nil { + return err + } + + for _, hash := range hashes { + if seen[hash] { + continue + } + tx, err := bitcoin.RPCGetTransaction(chain, rpc, hash) + if err != nil { + return fmt.Errorf("bitcoin.RPCGetTransaction(%s) => %v %v", hash, tx, err) + } + if tx == nil { + logger.Printf("bitcoin.RPCGetTransaction(%s) => %v %v", hash, tx, err) + continue + } + err = node.bitcoinRecordPendingTransaction(ctx, tx, chain) + if err != nil { + return err + } + seen[hash] = true + } + return nil +} + +func (node *Node) reconcileBitcoinPendingDeposits(ctx context.Context, chain byte, mempool map[string]bool, now time.Time) error { + observations, err := node.store.listPendingDepositObservations(ctx, chain) + if err != nil { + return err + } + rpc, _, _ := node.bitcoinParams(chain) + // One RPC per transaction even when it has several Safe outputs. + lookups := make(map[string]bool) + for _, d := range observations { + if mempool[d.TransactionHash] { + if d.MissingSince.Valid { + err := node.store.updatePendingDepositMissingSince(ctx, chain, d, sql.NullTime{}) + if err != nil { + return err + } + } + continue + } + if !d.MissingSince.Valid { + err := node.store.updatePendingDepositMissingSince(ctx, chain, d, sql.NullTime{Time: now, Valid: true}) + if err != nil { + return err + } + continue + } + if now.Before(d.MissingSince.Time.Add(bitcoinPendingDepositGracePeriod)) { + continue + } + found, checked := lookups[d.TransactionHash] + if !checked { + tx, err := bitcoin.RPCGetTransaction(chain, rpc, d.TransactionHash) + logger.Printf("reconcile pending bitcoin transaction(%d, %s) => %v %v", chain, d.TransactionHash, tx, err) + if err != nil { + return fmt.Errorf("bitcoin.RPCGetTransaction(%s) => %w", d.TransactionHash, err) + } + // With the required txindex=1, only a definitive not-found response + // permits deletion. Mined transactions stay pending. + found = tx != nil + lookups[d.TransactionHash] = found + } + if found { + // A transaction may reappear after the snapshot. Any later + // disappearance must receive a fresh grace period. + err := node.store.updatePendingDepositMissingSince(ctx, chain, d, sql.NullTime{}) + if err != nil { + return err + } + } else { + err := node.store.deleteMissingPendingDeposit(ctx, chain, d) + if err != nil { + return err + } + } + } + return nil +} + +func (node *Node) bitcoinRecordPendingTransaction(ctx context.Context, tx *bitcoin.RPCTransaction, chain byte) error { + for _, index := range bitcoinDepositOutputIndexes(tx) { + out := tx.Vout[index] + err := node.bitcoinRecordPendingDeposit(ctx, out.ScriptPubKey.Address, tx, out.N, out.Value, chain) + if err != nil { + return err + } + } + return nil +} + +func (node *Node) bitcoinRecordPendingDeposit(ctx context.Context, receiver string, tx *bitcoin.RPCTransaction, index int64, value float64, chain byte) error { + if len(tx.Vin) == 0 { + return fmt.Errorf("missing transaction inputs %s", tx.TxId) + } + + rpc, chainId, params := node.bitcoinParams(chain) + amount := decimal.NewFromFloat(value) + minimum := decimal.RequireFromString(params.TransactionMinimum) + + sent, err := node.store.QueryDepositSentHashes(ctx, []*Deposit{{TransactionHash: tx.TxId}}) + logger.Verbosef("store.QueryDepositSentHashes(%s) => %v %v", tx.TxId, sent, err) + if err != nil { + return fmt.Errorf("store.QueryDepositSentHashes(%s) => %v", tx.TxId, err) + } + change := node.bitcoinCheckDepositChange(ctx, tx.TxId, index, sent[tx.TxId]) + if change || amount.Cmp(minimum) < 0 { + return nil + } + + old, _, err := node.keeperStore.ReadBitcoinUTXO(ctx, tx.TxId, int(index)) + logger.Verbosef("keeperStore.ReadBitcoinUTXO(%s, %d) => %v %v", tx.TxId, index, old, err) + if err != nil { + return fmt.Errorf("keeperStore.ReadBitcoinUTXO(%s, %d) => %v", tx.TxId, index, err) + } else if old != nil { + return nil + } + + safe, err := node.keeperStore.ReadSafeByAddress(ctx, receiver) + logger.Verbosef("keeperStore.ReadSafeByAddress(%s) => %v %v", receiver, safe, err) + if err != nil { + return fmt.Errorf("keeperStore.ReadSafeByAddress(%s) => %v", receiver, err) + } else if safe == nil || safe.Chain != chain || safe.State != common.RequestStateDone { + return nil + } + + c, err := node.keeperStore.ReadUnspentUtxoCountForSafe(ctx, receiver) + logger.Verbosef("keeperStore.ReadUnspentUtxoCountForSafe(%s) => %d %v", receiver, c, err) + if err != nil || c >= bitcoin.MaxUnspentUtxo/2 { + return err + } + + createdAt := time.Now().UTC() + deposit := &PendingDeposit{ + TransactionHash: tx.TxId, + OutputIndex: index, + AssetId: chainId, + Amount: amount.String(), + Receiver: receiver, + State: common.RequestStateInitial, + Chain: chain, + Holder: safe.Holder, + CreatedAt: createdAt, + } + + sender, err := bitcoin.RPCGetTransactionSender(chain, rpc, tx) + if err != nil { + return fmt.Errorf("bitcoin.RPCGetTransactionSender(%s) => %v", tx.TxId, err) + } + deposit.Sender = sender + + err = node.store.WritePendingDepositRecord(ctx, deposit) + if err != nil { + return fmt.Errorf("store.WritePendingDepositRecord(%v) => %v", deposit, err) + } + return nil +} + +func (node *Node) ethereumPendingDepositsLoop(ctx context.Context, chain byte) { + for { + time.Sleep(time.Second) + err := node.scanEthereumPendingDeposits(ctx, chain) + logger.Printf("node.scanEthereumPendingDeposits(%d) => %v", chain, err) + } +} + +func (node *Node) scanEthereumPendingDeposits(ctx context.Context, chain byte) error { + // Expire observations even when the head is unchanged or RPC is unavailable. + // Formal deposits are processed independently by the block loop. + err := node.store.deleteExpiredEthereumPendingDeposits(ctx, chain, time.Now().UTC().Add(-ethereumPendingDepositTTL)) + if err != nil { + return err + } + rpc, _, _ := node.ethereumParams(chain) + duration := 5 * time.Second + switch chain { + case common.SafeChainPolygon: + duration = 2 * time.Second + case common.SafeChainEthereum: + } + + checkpoint, err := node.readPendingDepositCheckpoint(ctx, chain) + if err != nil { + return err + } + height, err := ethereum.RPCGetBlockHeight(rpc) + if err != nil { + logger.Printf("ethereum.RPCGetBlockHeight(%d) => %v", chain, err) + return nil + } + logger.Printf("node.readPendingDepositCheckpoint(%d) => %d %d", chain, checkpoint, height) + if checkpoint >= height { + time.Sleep(duration) + return nil + } + end := min(height, checkpoint+node.getChainBlockBatch(chain)) + for current := checkpoint + 1; current <= end; current++ { + err := node.ethereumRecordPendingBlock(ctx, current, chain) + logger.Printf("node.ethereumRecordPendingBlock(%d, %d) => %v", chain, current, err) + if err != nil { + return err + } + err = node.store.WriteProperty(ctx, pendingDepositCheckpointKey(chain), fmt.Sprint(current)) + if err != nil { + return err + } + } + return nil +} + +func (node *Node) ethereumRecordPendingBlock(ctx context.Context, height int64, chain byte) error { + transfers, err := node.ethereumReadPendingBlockTransfers(ctx, height, chain) + if err != nil { + return err + } + for _, transfer := range transfers { + err := node.ethereumRecordPendingTransfer(ctx, transfer, chain) + if err != nil { + return err + } + } + return nil +} + +func (node *Node) ethereumReadPendingBlockTransfers(ctx context.Context, height int64, chain byte) ([]*ethereum.Transfer, error) { + rpc, assetID, _ := node.ethereumParams(chain) + block, err := ethereum.RPCGetBlockWithTransactions(rpc, height) + if err != nil { + return nil, err + } + if block.Hash == "" || int64(block.Height) != height { + return nil, fmt.Errorf("invalid pending block at %d", height) + } + // Bind execution results to the transaction list, even if this height reorgs. + traces, err := ethereum.RPCDebugTraceBlockByHash(rpc, block.Hash) + if err != nil { + return nil, err + } + if len(traces) > len(block.Tx) || (chain != common.SafeChainPolygon && len(traces) != len(block.Tx)) { + return nil, fmt.Errorf("pending block trace count mismatch at %d", height) + } + for i, trace := range traces { + if block.Tx[i] == nil || block.Tx[i].Hash == "" || (trace.TxHash != "" && trace.TxHash != block.Tx[i].Hash) { + return nil, fmt.Errorf("pending block trace transaction mismatch at %d", height) + } + } + logs, err := ethereum.GetERC20TransferLogFromBlockHash(ctx, rpc, int64(chain), block.Hash) + if err != nil { + return nil, err + } + // Reject a branch change observed during the reads before interpreting transfers. + current, err := ethereum.RPCGetBlockByHeight(rpc, height) + if err != nil { + return nil, err + } + if current.Hash != block.Hash || current.Height != block.Height { + return nil, fmt.Errorf("pending block changed while reading at %d", height) + } + return append(ethereum.LoopBlockTraces(chain, assetID, traces, block), logs...), nil +} + +func (node *Node) ethereumRecordPendingTransfer(ctx context.Context, transfer *ethereum.Transfer, chain byte) error { + if transfer.Value == nil || transfer.Value.Sign() <= 0 || transfer.Receiver == ethereum.EthereumEmptyAddress { + return nil + } + _, _, 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) + if err != nil { + return fmt.Errorf("keeperStore.ReadDeposit(%s, %d, %s, %s) => %v %v", transfer.Hash, transfer.Index, transfer.AssetId, transfer.Receiver, old, err) + } else if old != nil { + return nil + } + safe, err := node.keeperStore.ReadSafeByAddress(ctx, transfer.Receiver) + logger.Printf("keeperStore.ReadSafeByAddress(%s, %s) => %v %v", transfer.Hash, transfer.Receiver, safe, err) + if err != nil { + return fmt.Errorf("keeperStore.ReadSafeByAddress(%s) => %v", transfer.Receiver, err) + } else if safe == nil || safe.Chain != chain || safe.State != common.RequestStateDone { + return nil + } + + sbm, err := node.keeperStore.ReadPositiveEthereumTokenBalancesMap(ctx, safe.Address) + logger.Verbosef("keeperStore.ReadPositiveEthereumTokenBalancesMap(%s) => %d %v", safe.Address, len(sbm), err) + if err != nil || len(sbm) >= ethereum.MaxAssetsCount/2 { + return err + } + + asset, err := node.fetchMixinAsset(ctx, transfer.AssetId) + logger.Printf("node.fetchMixinAsset(%s, %s) => %v %v", transfer.Hash, transfer.AssetId, asset, err) + if err != nil || asset == nil { + return err + } + if !checkGoodAsset(asset) { + return nil + } + + amount, err := node.ethereumDepositAmount(transfer, chain) + if err != nil { + return nil + } + if amount.Cmp(minimum) < 0 { + return nil + } + priceUSD := decimal.RequireFromString(asset.PriceUSD) + if amount.Mul(priceUSD).Cmp(minimumEthereumDepositValueUSD) < 0 { + return nil + } + + createdAt := time.Now().UTC() + deposit := &PendingDeposit{ + TransactionHash: transfer.Hash, + OutputIndex: transfer.Index, + AssetId: transfer.AssetId, + AssetAddress: transfer.TokenAddress, + Amount: amount.String(), + Receiver: transfer.Receiver, + Sender: transfer.Sender, + State: common.RequestStateInitial, + Chain: chain, + Holder: safe.Holder, + CreatedAt: createdAt, + } + + err = node.store.WritePendingDepositRecord(ctx, deposit) + if err != nil { + return fmt.Errorf("store.WritePendingDepositRecord(%v) => %v", deposit, err) + } + return nil +} diff --git a/observer/pending_test.go b/observer/pending_test.go new file mode 100644 index 00000000..e713e37f --- /dev/null +++ b/observer/pending_test.go @@ -0,0 +1,746 @@ +package observer + +import ( + "database/sql" + "encoding/json" + "fmt" + "math" + "math/big" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/MixinNetwork/safe/apps/bitcoin" + "github.com/MixinNetwork/safe/apps/ethereum" + "github.com/MixinNetwork/safe/common" + ga "github.com/ethereum/go-ethereum/accounts/abi" + ec "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/stretchr/testify/require" +) + +func TestEthereumPendingExpiry(t *testing.T) { + for _, chain := range []byte{common.SafeChainEthereum, common.SafeChainPolygon} { + t.Run(fmt.Sprint(chain), func(t *testing.T) { + ctx := t.Context() + s := coverageObserverStore(t) + now := time.Now().UTC() + cutoff := now.Add(-ethereumPendingDepositTTL) + for i := range 102 { + require.NoError(t, s.WritePendingDepositRecord(ctx, &PendingDeposit{ + TransactionHash: fmt.Sprintf("orphan-%d", i), Chain: chain, + State: common.RequestStateInitial, CreatedAt: cutoff.Add(-time.Second), + })) + } + for _, d := range []*PendingDeposit{ + {TransactionHash: "boundary", Chain: chain, CreatedAt: cutoff}, + {TransactionHash: "recent", Chain: chain, CreatedAt: cutoff.Add(time.Second)}, + {TransactionHash: "confirmed", Chain: chain, CreatedAt: cutoff}, + } { + d.State = common.RequestStateInitial + require.NoError(t, s.WritePendingDepositRecord(ctx, d)) + } + for _, hash := range []string{"confirmed", "orphan-0"} { + require.NoError(t, s.WritePendingDepositIfNotExists(ctx, &Deposit{ + TransactionHash: hash, Chain: chain, RequestId: hash, + State: common.RequestStateInitial, CreatedAt: now, UpdatedAt: now, + })) + } + require.NoError(t, s.ConfirmPendingDeposit(ctx, "confirmed", 0, "confirmed", chain)) + for _, other := range []byte{common.SafeChainEthereum, common.SafeChainPolygon, common.SafeChainBitcoin, common.SafeChainLitecoin} { + if other == chain { + continue + } + require.NoError(t, s.WritePendingDepositRecord(ctx, &PendingDeposit{ + TransactionHash: "other-chain", Chain: other, + State: common.RequestStateInitial, CreatedAt: cutoff, + })) + } + require.Error(t, s.deleteExpiredEthereumPendingDeposits(ctx, common.SafeChainBitcoin, cutoff)) + require.NoError(t, s.deleteExpiredEthereumPendingDeposits(ctx, chain, cutoff)) + require.NoError(t, s.deleteExpiredEthereumPendingDeposits(ctx, chain, cutoff)) + rows, err := s.ListPendingDeposits(ctx, int(chain), "", 0) + require.NoError(t, err) + require.Len(t, rows, 1, "expired rows beyond the first 100 must also be removed") + require.Equal(t, "recent", rows[0].TransactionHash) + var count int + require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM pending_deposits").Scan(&count)) + require.Equal(t, 5, count, "keep recent, confirmed and all three other chains") + // The formal record survives expiry and can still finish confirmation. + formal, err := s.ListDeposits(ctx, int(chain), "", common.RequestStateInitial, 0) + require.NoError(t, err) + require.Len(t, formal, 1) + require.Equal(t, "orphan-0", formal[0].TransactionHash) + require.NoError(t, s.ConfirmPendingDeposit(ctx, "orphan-0", 0, "orphan-0", chain)) + formal, err = s.ListDeposits(ctx, int(chain), "", common.RequestStateDone, 0) + require.NoError(t, err) + require.Len(t, formal, 2) + }) + } +} + +func TestEthereumPendingExpiryRunsBeforeRPC(t *testing.T) { + for _, chain := range []byte{common.SafeChainEthereum, common.SafeChainPolygon} { + for _, mode := range []string{"idle", "rpc error", "cleanup error"} { + t.Run(fmt.Sprintf("%d/%s", chain, mode), func(t *testing.T) { + ctx := t.Context() + s := coverageObserverStore(t) + require.NoError(t, s.WritePendingDepositRecord(ctx, &PendingDeposit{ + TransactionHash: "expired", Chain: chain, State: common.RequestStateInitial, + CreatedAt: time.Now().UTC().Add(-ethereumPendingDepositTTL - time.Second), + })) + checkpoint := depositCheckpointDefault(chain) + require.NoError(t, s.WriteProperty(ctx, pendingDepositCheckpointKey(chain), fmt.Sprint(checkpoint))) + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + var q struct{ Method string } + require.NoError(t, json.NewDecoder(r.Body).Decode(&q)) + require.Equal(t, "eth_blockNumber", q.Method) + if mode == "rpc error" { + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{"code": -32000, "message": "backend unavailable"}, + })) + return + } + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"result": fmt.Sprintf("0x%x", checkpoint)})) + })) + defer server.Close() + node := &Node{store: s, conf: &Configuration{EthereumRPC: server.URL, PolygonRPC: server.URL}} + if mode == "cleanup error" { + _, err := s.db.Exec(`CREATE TRIGGER reject_expiry BEFORE DELETE ON pending_deposits + BEGIN SELECT RAISE(ABORT, 'injected expiry error'); END`) + require.NoError(t, err) + require.ErrorContains(t, node.scanEthereumPendingDeposits(ctx, chain), "injected expiry error") + require.Zero(t, calls.Load()) + rows, err := s.ListPendingDeposits(ctx, int(chain), "", 0) + require.NoError(t, err) + require.Len(t, rows, 1) + return + } + require.NoError(t, node.scanEthereumPendingDeposits(ctx, chain)) + require.EqualValues(t, 1, calls.Load()) + rows, err := s.ListPendingDeposits(ctx, int(chain), "", 0) + require.NoError(t, err) + require.Empty(t, rows) + cursor, err := node.readPendingDepositCheckpoint(ctx, chain) + require.NoError(t, err) + require.Equal(t, checkpoint, cursor, "expiry must not rewind or advance scanning") + }) + } + } +} + +func TestBitcoinPendingReconciliationDeletesMissing(t *testing.T) { + for _, chain := range []byte{common.SafeChainBitcoin, common.SafeChainLitecoin} { + for _, mode := range []string{"missing", "mined", "error"} { + t.Run(fmt.Sprintf("%d/%s", chain, mode), func(t *testing.T) { + ctx := t.Context() + s := coverageObserverStore(t) + now := time.Now().UTC() + // More than one public query page, with two outputs per transaction. + for i := range 102 { + require.NoError(t, s.WritePendingDepositRecord(ctx, &PendingDeposit{ + TransactionHash: fmt.Sprintf("replaced-%d", i/2), OutputIndex: int64(i % 2), + Chain: chain, State: common.RequestStateInitial, CreatedAt: now.Add(-time.Hour), + })) + } + var calls atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var q struct{ Method string } + require.NoError(t, json.NewDecoder(r.Body).Decode(&q)) + require.Equal(t, "getrawtransaction", q.Method) + calls.Add(1) + if mode == "mined" { + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{ + "result": map[string]any{"blockhash": "mined-block"}, + })) + return + } + code := -5 + if mode == "error" { + code = -32603 + } + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{"code": code, "message": "lookup failed"}, + })) + })) + defer server.Close() + node := &Node{store: s, conf: &Configuration{BitcoinRPC: server.URL, LitecoinRPC: server.URL}} + require.NoError(t, node.reconcileBitcoinPendingDeposits(ctx, chain, nil, now)) + require.NoError(t, node.reconcileBitcoinPendingDeposits(ctx, chain, nil, now.Add(bitcoinPendingDepositGracePeriod-time.Second))) + require.Zero(t, calls.Load(), "do not query or delete before grace expires") + + err := node.reconcileBitcoinPendingDeposits(ctx, chain, nil, now.Add(bitcoinPendingDepositGracePeriod)) + var count int + require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM pending_deposits").Scan(&count)) + switch mode { + case "missing": + require.NoError(t, err) + require.Zero(t, count, "remove rows, not just hide them with Failed state") + require.EqualValues(t, 51, calls.Load(), "one lookup per transaction") + // A rebroadcast becomes a new observation without a Failed revival path. + rebroadcast := &PendingDeposit{TransactionHash: "replaced-0", Chain: chain, + State: common.RequestStateInitial, CreatedAt: now} + require.NoError(t, s.WritePendingDepositRecord(ctx, rebroadcast)) + require.NoError(t, s.WritePendingDepositRecord(ctx, rebroadcast)) + rows, err := s.ListPendingDeposits(ctx, int(chain), "", 0) + require.NoError(t, err) + require.Len(t, rows, 1) + require.True(t, rows[0].CreatedAt.Equal(now)) + case "mined": + require.NoError(t, err) + require.Equal(t, 102, count) + require.EqualValues(t, 51, calls.Load()) + require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM pending_deposits WHERE missing_since IS NOT NULL").Scan(&count)) + require.Zero(t, count, "finding a transaction resets its disappearance timer") + case "error": + require.ErrorContains(t, err, "lookup failed") + require.Equal(t, 102, count) + require.EqualValues(t, 1, calls.Load(), "abort at the first RPC error") + } + }) + } + } +} + +func TestDeleteMissingPendingDepositGuards(t *testing.T) { + for _, mode := range []string{"missing", "refreshed", "formal", "confirmed"} { + t.Run(mode, func(t *testing.T) { + ctx := t.Context() + s := coverageObserverStore(t) + now := time.Now().UTC() + d := &PendingDeposit{TransactionHash: "same-hash", Chain: common.SafeChainBitcoin, + State: common.RequestStateInitial, CreatedAt: now} + require.NoError(t, s.WritePendingDepositRecord(ctx, d)) + otherOutput := *d + otherOutput.OutputIndex = 1 + require.NoError(t, s.WritePendingDepositRecord(ctx, &otherOutput)) + otherChain := *d + otherChain.Chain = common.SafeChainLitecoin + require.NoError(t, s.WritePendingDepositRecord(ctx, &otherChain)) + observation := pendingDepositObservation{TransactionHash: d.TransactionHash} + since := sql.NullTime{Time: now.Add(-2 * bitcoinPendingDepositGracePeriod), Valid: true} + require.NoError(t, s.updatePendingDepositMissingSince(ctx, d.Chain, observation, since)) + observation.MissingSince = since + if mode == "refreshed" { + require.NoError(t, s.updatePendingDepositMissingSince(ctx, d.Chain, observation, sql.NullTime{})) + } + if mode == "formal" || mode == "confirmed" { + formal := &Deposit{TransactionHash: d.TransactionHash, Chain: d.Chain, + State: common.RequestStateInitial, RequestId: "request", CreatedAt: now, UpdatedAt: now} + require.NoError(t, s.WritePendingDepositIfNotExists(ctx, formal)) + if mode == "confirmed" { + require.NoError(t, s.ConfirmPendingDeposit(ctx, d.TransactionHash, 0, formal.RequestId, d.Chain)) + } + } + require.NoError(t, s.deleteMissingPendingDeposit(ctx, d.Chain, observation)) + var count int + require.NoError(t, s.db.QueryRow("SELECT COUNT(*) FROM pending_deposits").Scan(&count)) + if mode == "missing" { + require.Equal(t, 2, count, "delete only the matching chain and output") + } else { + require.Equal(t, 3, count, "preserve refreshed, formal and confirmed observations") + } + }) + } +} + +func TestPendingDepositConfirmationAndHTTP(t *testing.T) { + f := newTransactionHTTPFixture(t) + ctx := t.Context() + now := time.Now().UTC().Add(-time.Minute) + safe, err := f.node.keeperStore.ReadSafeByAddress(ctx, f.address) + require.NoError(t, err) + d := &PendingDeposit{TransactionHash: "pending-hash", OutputIndex: 3, AssetId: common.SafeEthereumChainId, + AssetAddress: ethereum.EthereumEmptyAddress, Amount: "1.25", Receiver: f.address, + Sender: "sender-address", State: common.RequestStateInitial, Chain: common.SafeChainEthereum, Holder: safe.Holder, CreatedAt: now} + require.NoError(t, f.node.store.WritePendingDepositRecord(ctx, d)) + duplicate := *d + duplicate.CreatedAt = time.Now().UTC() + require.NoError(t, f.node.store.WritePendingDepositRecord(ctx, &duplicate)) + otherChain := *d + otherChain.Chain = common.SafeChainPolygon + require.NoError(t, f.node.store.WritePendingDepositRecord(ctx, &otherChain)) + deposit := &Deposit{TransactionHash: d.TransactionHash, OutputIndex: d.OutputIndex, AssetId: d.AssetId, + Amount: d.Amount, Receiver: d.Receiver, Chain: d.Chain, State: common.RequestStateInitial, + RequestId: "deposit-request", CreatedAt: now, UpdatedAt: now} + deposit.Holder = safe.Holder + require.NoError(t, f.node.store.WritePendingDepositIfNotExists(ctx, deposit)) + get := func(target string, pending bool) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, target, nil) + if pending { + f.node.httpListPendingDeposits(w, r, nil) + } else { + f.node.httpListDeposits(w, r, nil) + } + return w + } + chain := fmt.Sprint(d.Chain) + response := get("/pending_deposits?chain="+chain+"&holder="+safe.Holder, true) + require.Equal(t, http.StatusOK, response.Code) + var pending []*PendingDeposit + require.NoError(t, json.Unmarshal(response.Body.Bytes(), &pending)) + require.Len(t, pending, 1) + require.True(t, pending[0].CreatedAt.Equal(now), "duplicate scan preserves creation time") + require.Equal(t, d.Amount, pending[0].Amount) + require.Equal(t, d.AssetAddress, pending[0].AssetAddress) + require.Equal(t, d.Sender, pending[0].Sender) + require.Equal(t, d.Holder, pending[0].Holder) + require.Equal(t, common.RequestStateInitial, pending[0].State) + var fields []map[string]any + require.NoError(t, json.Unmarshal(response.Body.Bytes(), &fields)) + require.Equal(t, d.AssetAddress, fields[0]["asset_address"]) + require.Equal(t, d.Sender, fields[0]["sender"]) + require.Equal(t, d.Holder, fields[0]["holder"]) + for _, query := range []string{"&holder=unknown", + fmt.Sprintf("&offset=%d", now.Add(time.Second).UnixNano())} { + require.JSONEq(t, `[]`, get("/pending_deposits?chain="+chain+query, true).Body.String()) + } + for _, query := range []string{"", "?chain=999", "?chain=invalid"} { + require.Equal(t, http.StatusBadRequest, get("/pending_deposits"+query, true).Code) + } + // Formal Initial deposits fill gaps left by the early scanner. + _, err = f.node.store.db.ExecContext(ctx, "DELETE FROM pending_deposits WHERE chain=?", d.Chain) + require.NoError(t, err) + response = get("/pending_deposits?chain="+chain, true) + require.Equal(t, http.StatusOK, response.Code) + var initial []map[string]any + require.NoError(t, json.Unmarshal(response.Body.Bytes(), &initial)) + require.Len(t, initial, 1) + require.Equal(t, float64(common.RequestStateInitial), initial[0]["state"]) + require.Equal(t, d.TransactionHash, initial[0]["transaction_hash"]) + require.Equal(t, float64(d.OutputIndex), initial[0]["output_index"]) + require.JSONEq(t, `[]`, get("/deposits?chain="+chain, false).Body.String()) + require.NoError(t, f.node.store.WritePendingDepositRecord(ctx, d)) + require.NoError(t, f.node.store.ConfirmPendingDeposit(ctx, d.TransactionHash, d.OutputIndex, deposit.RequestId, d.Chain)) + // A slow or restarted scanner must not recreate a confirmed observation. + require.NoError(t, f.node.store.WritePendingDepositRecord(ctx, d)) + require.JSONEq(t, `[]`, get("/pending_deposits?chain="+chain, true).Body.String()) + var retained int + require.NoError(t, f.node.store.db.QueryRow("SELECT COUNT(*) FROM pending_deposits WHERE chain=?", d.Chain).Scan(&retained)) + require.Equal(t, 1, retained, "confirmed pending observations remain stored as history") + var state int + require.NoError(t, f.node.store.db.QueryRowContext(ctx, "SELECT state FROM pending_deposits WHERE chain=?", d.Chain).Scan(&state)) + require.Equal(t, common.RequestStateDone, state) + _, err = f.node.store.db.ExecContext(ctx, "DELETE FROM pending_deposits WHERE chain=?", d.Chain) + require.NoError(t, err) + require.NoError(t, f.node.store.WritePendingDepositRecord(ctx, d)) + require.NoError(t, f.node.store.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM pending_deposits WHERE chain=?", d.Chain).Scan(&retained)) + require.Zero(t, retained, "a late scan must not insert an already confirmed deposit") + response = get("/deposits?chain="+chain, false) + require.Equal(t, http.StatusOK, response.Code) + var confirmed []map[string]any + require.NoError(t, json.Unmarshal(response.Body.Bytes(), &confirmed)) + require.Len(t, confirmed, 1) + require.Equal(t, "done", confirmed[0]["state"]) + for _, key := range []string{"chain", "transaction_hash", "output_index"} { + require.Equal(t, initial[0][key], confirmed[0][key]) + } + remaining, err := f.node.store.ListPendingDeposits(ctx, int(otherChain.Chain), "", 0) + require.NoError(t, err) + require.Len(t, remaining, 1, "confirmation does not delete an observation on another chain") + require.Equal(t, common.RequestStateInitial, remaining[0].State) +} + +func TestBitcoinPendingMempoolExcludesChange(t *testing.T) { + for _, chain := range []byte{common.SafeChainBitcoin, common.SafeChainLitecoin} { + t.Run(fmt.Sprint(chain), func(t *testing.T) { + f := newTransactionHTTPFixture(t) + _, err := f.keeperDB.Exec("UPDATE safes SET chain=?", chain) + require.NoError(t, err) + _, err = f.keeperDB.Exec(`UPDATE transactions SET data='[{"receiver":"recipient","amount":"1"}]'`) + require.NoError(t, err) + _, err = f.node.store.db.Exec("UPDATE transactions SET spent_hash=?", "mempool-hash") + require.NoError(t, err) + rawCalls := 0 + senderCalls := 0 + missingCalls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var q struct { + Method string + Params []json.RawMessage + } + require.NoError(t, json.NewDecoder(r.Body).Decode(&q)) + var result any + switch q.Method { + case "getrawmempool": + result = []string{"missing-hash", "mempool-hash"} + case "getrawtransaction": + if string(q.Params[0]) == `"missing-hash"` { + missingCalls++ + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{ + "result": nil, + "error": map[string]any{"code": -5, "message": "No such mempool or blockchain transaction"}, + })) + return + } + if string(q.Params[0]) == `"previous-hash"` { + senderCalls++ + result = map[string]any{"txid": "previous-hash", "vout": []any{ + map[string]any{"n": 0, "value": 5, "scriptPubKey": map[string]any{"type": "witness_v0_keyhash", "address": "sender-address"}}, + }} + break + } + rawCalls++ + result = map[string]any{"txid": "mempool-hash", "vin": []any{map[string]any{"txid": "previous-hash", "vout": 0}}, "vout": []any{ + map[string]any{"n": 0, "value": 1.25, "scriptPubKey": map[string]any{"type": "witness_v0_scripthash", "address": f.address}}, + map[string]any{"n": 1, "value": 2, "scriptPubKey": map[string]any{"type": "witness_v0_scripthash", "address": f.address}}, + map[string]any{"n": 2, "value": 3, "scriptPubKey": map[string]any{"type": "witness_v0_scripthash", "address": "unknown"}}, + }} + default: + t.Errorf("unexpected RPC %s", q.Method) + } + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"result": result, "error": nil})) + })) + defer server.Close() + f.node.conf = &Configuration{ + BitcoinRPC: server.URL, LitecoinRPC: server.URL, + Bitcoin: ChainParams{TransactionMinimum: "1.25"}, Litecoin: ChainParams{TransactionMinimum: "1.25"}, + } + seen := make(map[string]bool) + require.NoError(t, f.node.scanBitcoinPendingDeposits(t.Context(), chain, seen)) + require.NoError(t, f.node.scanBitcoinPendingDeposits(t.Context(), chain, seen)) + require.Equal(t, 1, rawCalls, "unchanged mempool entries do not refetch transactions") + require.Equal(t, 1, senderCalls, "sender is resolved only for newly observed deposits") + require.Equal(t, 2, missingCalls, "missing transactions are skipped without being marked seen") + pending, err := f.node.store.ListPendingDeposits(t.Context(), int(chain), "", 0) + require.NoError(t, err) + require.Len(t, pending, 1) + require.Equal(t, int64(0), pending[0].OutputIndex) + require.Equal(t, "1.25", pending[0].Amount) + require.Empty(t, pending[0].AssetAddress, "Bitcoin native deposits have no token contract address") + require.Equal(t, "sender-address", pending[0].Sender) + safe, err := f.node.keeperStore.ReadSafeByAddress(t.Context(), f.address) + require.NoError(t, err) + require.Equal(t, safe.Holder, pending[0].Holder) + deposits, err := f.node.store.ListDeposits(t.Context(), int(chain), "", 0, 0) + require.NoError(t, err) + require.Empty(t, deposits, "mempool scanning does not enter the confirmation queue") + tx, err := bitcoin.RPCGetTransaction(chain, server.URL, "mempool-hash") + require.NoError(t, err) + for i := range 2 { + out := tx.Vout[i] + require.NoError(t, f.node.bitcoinWritePendingDeposit(t.Context(), f.address, tx, out.N, out.Value, chain)) + } + deposits, err = f.node.store.ListDeposits(t.Context(), int(chain), "", common.RequestStateInitial, 0) + require.NoError(t, err) + require.Len(t, deposits, 2) + for _, d := range deposits { + require.Equal(t, d.OutputIndex == 1, d.IsChange, "persist the output classification for confirmation and listing") + } + }) + } +} + +func TestBitcoinPendingPreservesRPCErrors(t *testing.T) { + f := newTransactionHTTPFixture(t) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var q struct{ Method string } + require.NoError(t, json.NewDecoder(r.Body).Decode(&q)) + if q.Method == "getrawmempool" { + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"result": []string{"mempool-hash"}})) + return + } + require.Equal(t, "getrawtransaction", q.Method) + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{"code": -32601, "message": "method not found"}, + })) + })) + t.Cleanup(server.Close) + f.node.conf = &Configuration{BitcoinRPC: server.URL} + seen := make(map[string]bool) + err := f.node.scanBitcoinPendingDeposits(t.Context(), common.SafeChainBitcoin, seen) + require.ErrorContains(t, err, "method not found") + require.Empty(t, seen) +} + +func TestBitcoinPendingUsesDepositAdmissionChecks(t *testing.T) { + for _, test := range []struct { + name string + amount float64 + credited bool + outputs int + }{ + {name: "below minimum", amount: 0.00009999}, + {name: "already credited by keeper", amount: 1, credited: true}, + {name: "safe reached output limit", amount: 1, outputs: bitcoin.MaxUnspentUtxo / 2}, + } { + t.Run(test.name, func(t *testing.T) { + f := newTransactionHTTPFixture(t) + ctx := t.Context() + chain := byte(common.SafeChainBitcoin) + _, err := f.keeperDB.ExecContext(ctx, "UPDATE safes SET chain=?", chain) + require.NoError(t, err) + f.node.conf = &Configuration{Bitcoin: ChainParams{TransactionMinimum: "0.0001"}} + tx := &bitcoin.RPCTransaction{} + require.NoError(t, json.Unmarshal([]byte(`{"txid":"pending-hash","vin":[{"txid":"previous-hash","vout":0}]}`), tx)) + now := time.Now().UTC() + if test.credited { + _, err := f.keeperDB.ExecContext(ctx, `INSERT INTO bitcoin_outputs + (transaction_hash, output_index, address, satoshi, script, sequence, chain, state, request_id, created_at, updated_at) + VALUES (?, 0, ?, 100000000, '', 0, ?, ?, 'credited-request', ?, ?)`, + tx.TxId, f.address, chain, common.RequestStateInitial, now, now) + require.NoError(t, err) + } + if test.outputs > 0 { + _, err := f.keeperDB.ExecContext(ctx, `WITH RECURSIVE outputs(n) AS + (VALUES(0) UNION ALL SELECT n+1 FROM outputs WHERE n+1 0 { + _, err := f.keeperDB.ExecContext(ctx, `WITH RECURSIVE assets(n) AS + (VALUES(0) UNION ALL SELECT n+1 FROM assets WHERE n+1=?", strings.Join(depositsCols, ",")) - params := []any{chain, time.Unix(0, offset)} + params := []any{chain, time.Unix(0, offset).UTC()} + if excludeChange { + query += " AND is_change=FALSE" + } if state > 0 { query += " AND state=?" params = append(params, state) @@ -382,7 +410,35 @@ func (s *SQLite3Store) ListDeposits(ctx context.Context, chain int, holder strin params = append(params, holder) } query += " ORDER BY updated_at ASC LIMIT 100" - rows, err := s.db.QueryContext(ctx, query, params...) + return s.queryDeposits(ctx, nil, query, params) +} + +func (s *SQLite3Store) listDepositsByCreatedAt(ctx context.Context, tx *sql.Tx, chain int, holder string, state int, offset int64, excludeChange bool) ([]*Deposit, error) { + query := fmt.Sprintf("SELECT %s FROM deposits WHERE chain=? AND created_at>=?", strings.Join(depositsCols, ",")) + params := []any{chain, time.Unix(0, offset).UTC()} + if state > 0 { + query += " AND state=?" + params = append(params, state) + } + if holder != "" { + query += " AND holder=?" + params = append(params, holder) + } + if excludeChange { + query += " AND is_change=FALSE" + } + query += " ORDER BY created_at ASC, transaction_hash ASC, CAST(output_index AS INTEGER) ASC LIMIT 100" + return s.queryDeposits(ctx, tx, query, params) +} + +func (s *SQLite3Store) queryDeposits(ctx context.Context, tx *sql.Tx, query string, params []any) ([]*Deposit, error) { + var rows *sql.Rows + var err error + if tx != nil { + rows, err = tx.QueryContext(ctx, query, params...) + } else { + rows, err = s.db.QueryContext(ctx, query, params...) + } if err != nil { return nil, err } @@ -391,13 +447,13 @@ func (s *SQLite3Store) ListDeposits(ctx context.Context, chain int, holder strin var deposits []*Deposit for rows.Next() { var d Deposit - err := rows.Scan(&d.TransactionHash, &d.OutputIndex, &d.AssetId, &d.AssetAddress, &d.Amount, &d.Receiver, &d.Sender, &d.State, &d.Chain, &d.Holder, &d.Category, &d.RequestId, &d.CreatedAt, &d.UpdatedAt) + err := rows.Scan(&d.TransactionHash, &d.OutputIndex, &d.AssetId, &d.AssetAddress, &d.Amount, &d.Receiver, &d.Sender, &d.State, &d.Chain, &d.Holder, &d.Category, &d.RequestId, &d.IsChange, &d.CreatedAt, &d.UpdatedAt) if err != nil { return nil, err } deposits = append(deposits, &d) } - return deposits, nil + return deposits, rows.Err() } func (s *SQLite3Store) CheckUnconfirmedDepositsForAssetAndHolder(ctx context.Context, holder, assetId string, offset time.Time) (bool, error) { @@ -414,10 +470,19 @@ func (s *SQLite3Store) CheckUnconfirmedDepositsForAssetAndHolder(ctx context.Con } func (s *SQLite3Store) QueryDepositSentHashes(ctx context.Context, deposits []*Deposit) (map[string]string, error) { + return s.queryDepositSentHashes(ctx, nil, deposits) +} + +func (s *SQLite3Store) queryDepositSentHashes(ctx context.Context, tx *sql.Tx, deposits []*Deposit) (map[string]string, error) { sent := make(map[string]string) for _, d := range deposits { query := "SELECT transaction_hash FROM transactions WHERE spent_hash=?" - row := s.db.QueryRowContext(ctx, query, d.TransactionHash) + var row *sql.Row + if tx != nil { + row = tx.QueryRowContext(ctx, query, d.TransactionHash) + } else { + row = s.db.QueryRowContext(ctx, query, d.TransactionHash) + } var hash string err := row.Scan(&hash) @@ -477,7 +542,7 @@ func (s *SQLite3Store) UpdateDepositRequestId(ctx context.Context, transactionHa return tx.Commit() } -func (s *SQLite3Store) ConfirmPendingDeposit(ctx context.Context, transactionHash string, outputIndex int64, rid string) error { +func (s *SQLite3Store) ConfirmPendingDeposit(ctx context.Context, transactionHash string, outputIndex int64, rid string, chain byte) error { s.mutex.Lock() defer s.mutex.Unlock() @@ -487,12 +552,18 @@ func (s *SQLite3Store) ConfirmPendingDeposit(ctx context.Context, transactionHas } defer common.Rollback(tx) - query := "UPDATE deposits SET state=?, updated_at=? WHERE transaction_hash=? AND output_index=? AND request_id=? AND state=?" - err = s.execOne(ctx, tx, query, common.RequestStateDone, time.Now().UTC(), transactionHash, outputIndex, rid, common.RequestStateInitial) + query := "UPDATE deposits SET state=?, updated_at=? WHERE transaction_hash=? AND output_index=? AND request_id=? AND chain=? AND state=?" + err = s.execOne(ctx, tx, query, common.RequestStateDone, time.Now().UTC(), transactionHash, outputIndex, rid, chain, common.RequestStateInitial) if err != nil { return fmt.Errorf("UPDATE deposits %v", err) } + query = "UPDATE pending_deposits SET state=?, missing_since=NULL WHERE transaction_hash=? AND output_index=? AND chain=? AND state=?" + _, err = tx.ExecContext(ctx, query, common.RequestStateDone, transactionHash, outputIndex, chain, common.RequestStateInitial) + if err != nil { + return fmt.Errorf("UPDATE pending_deposits %v", err) + } + return tx.Commit() } @@ -1072,3 +1143,124 @@ func (s *SQLite3Store) WriteCache(ctx context.Context, k, v string) error { func (s SQLite3Store) writeBlockCheckpoint(ctx context.Context, chain byte, checkpoint int64) error { return s.WriteProperty(ctx, depositCheckpointKey(chain), fmt.Sprint(checkpoint)) } + +func (s *SQLite3Store) WritePendingDepositRecord(ctx context.Context, d *PendingDeposit) error { + s.mutex.Lock() + defer s.mutex.Unlock() + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer common.Rollback(tx) + + if d.State != common.RequestStateInitial { + panic(d.State) + } + + existed, err := s.checkExistence(ctx, tx, "SELECT amount FROM pending_deposits WHERE chain=? AND transaction_hash=? AND output_index=?", d.Chain, d.TransactionHash, d.OutputIndex) + if err != nil || existed { + return err + } + + existed, err = s.checkExistence(ctx, tx, "SELECT amount FROM deposits WHERE transaction_hash=? AND output_index=? AND chain=? AND state=?", d.TransactionHash, d.OutputIndex, d.Chain, common.RequestStateDone) + if err != nil || existed { + return err + } + + err = s.execOne(ctx, tx, buildInsertionSQL("pending_deposits", pendingDepositsCols), d.values()...) + if err != nil { + return fmt.Errorf("INSERT pending_deposits %v", err) + } + + return tx.Commit() +} + +type pendingDepositObservation struct { + TransactionHash string + OutputIndex int64 + MissingSince sql.NullTime +} + +// Reconciliation must include records beyond the public API's first 100 rows. +func (s *SQLite3Store) listPendingDepositObservations(ctx context.Context, chain byte) ([]pendingDepositObservation, error) { + rows, err := s.db.QueryContext(ctx, "SELECT transaction_hash, output_index, missing_since FROM pending_deposits WHERE chain=? AND state=?", chain, common.RequestStateInitial) + if err != nil { + return nil, err + } + defer util.CloseOrPanic(rows) + + var observations []pendingDepositObservation + for rows.Next() { + var d pendingDepositObservation + if err := rows.Scan(&d.TransactionHash, &d.OutputIndex, &d.MissingSince); err != nil { + return nil, err + } + observations = append(observations, d) + } + return observations, rows.Err() +} + +func (s *SQLite3Store) updatePendingDepositMissingSince(ctx context.Context, chain byte, d pendingDepositObservation, since sql.NullTime) error { + s.mutex.Lock() + defer s.mutex.Unlock() + + _, err := s.db.ExecContext(ctx, "UPDATE pending_deposits SET missing_since=? WHERE chain=? AND transaction_hash=? AND output_index=? AND state=? AND missing_since IS ?", + since, chain, d.TransactionHash, d.OutputIndex, common.RequestStateInitial, d.MissingSince) + return err +} + +func (s *SQLite3Store) deleteMissingPendingDeposit(ctx context.Context, chain byte, d pendingDepositObservation) error { + s.mutex.Lock() + defer s.mutex.Unlock() + + // Do not delete an observation confirmed or refreshed during the RPC lookup, + // or one already picked up by the independent formal deposit scanner. + _, err := s.db.ExecContext(ctx, `DELETE FROM pending_deposits + WHERE chain=? AND transaction_hash=? AND output_index=? AND state=? AND missing_since=? + AND NOT EXISTS (SELECT 1 FROM deposits WHERE chain=? AND transaction_hash=? AND output_index=?)`, + chain, d.TransactionHash, d.OutputIndex, common.RequestStateInitial, d.MissingSince, + chain, d.TransactionHash, d.OutputIndex) + return err +} + +func (s *SQLite3Store) ListPendingDeposits(ctx context.Context, chain int, holder string, offset int64) ([]*PendingDeposit, error) { + query := fmt.Sprintf("SELECT %s FROM pending_deposits WHERE chain=? AND state=? AND created_at>=?", strings.Join(pendingDepositsCols, ",")) + params := []any{chain, common.RequestStateInitial, time.Unix(0, offset).UTC()} + if holder != "" { + query += " AND holder=?" + params = append(params, holder) + } + query += " ORDER BY created_at ASC, transaction_hash ASC, output_index ASC LIMIT 100" + rows, err := s.db.QueryContext(ctx, query, params...) + if err != nil { + return nil, err + } + defer util.CloseOrPanic(rows) + + deposits := make([]*PendingDeposit, 0) + for rows.Next() { + var d PendingDeposit + err := rows.Scan(&d.TransactionHash, &d.OutputIndex, &d.AssetId, &d.AssetAddress, &d.Amount, &d.Receiver, &d.Sender, &d.State, &d.Chain, &d.Holder, &d.CreatedAt) + if err != nil { + return nil, err + } + deposits = append(deposits, &d) + } + return deposits, rows.Err() +} + +func (s *SQLite3Store) deleteExpiredEthereumPendingDeposits(ctx context.Context, chain byte, cutoff time.Time) error { + switch chain { + case common.SafeChainEthereum, common.SafeChainPolygon: + default: + return fmt.Errorf("invalid EVM pending deposit chain %d", chain) + } + s.mutex.Lock() + defer s.mutex.Unlock() + + // Expiry only ends the observation; it does not invalidate a formal deposit. + _, err := s.db.ExecContext(ctx, "DELETE FROM pending_deposits WHERE chain=? AND state=? AND created_at<=?", + chain, common.RequestStateInitial, cutoff.UTC()) + return err +}