From 9debef02422bd4ece64aea6bfd650b795c360980 Mon Sep 17 00:00:00 2001 From: hundredark Date: Sat, 10 Oct 2026 17:40:34 +0800 Subject: [PATCH 01/14] scan the latest blocks but restrict the pending deposits with confirmations --- observer/bitcoin.go | 13 +++++-------- observer/ethereum.go | 13 +++++-------- 2 files changed, 10 insertions(+), 16 deletions(-) diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 5d7ff4f2..41a900f6 100644 --- a/observer/bitcoin.go +++ b/observer/bitcoin.go @@ -49,11 +49,6 @@ func (node *Node) bitcoinNetworkInfoLoop(ctx context.Context, chain byte) { if err != nil { continue } - delay := node.getChainFinalizationDelay(chain) - if delay > height || delay < 1 { - panic(delay) - } - height = height + 1 - delay info, err := node.keeperStore.ReadLatestNetworkInfo(ctx, chain, time.Now()) if err != nil { panic(err) @@ -279,6 +274,9 @@ func (node *Node) bitcoinConfirmPendingDeposit(ctx context.Context, deposit *Dep if info.Height < output.Height { confirmations = 0 } + if confirmations < uint64(node.getChainFinalizationDelay(deposit.Chain)) { + return nil + } if !bitcoin.CheckFinalization(confirmations, output.Coinbase) { return nil } @@ -323,8 +321,7 @@ func (node *Node) bitcoinRPCBlocksLoop(ctx context.Context, chain byte) { continue } logger.Printf("node.bitcoinReadDepositCheckpoint(%d) => %d %d", chain, checkpoint, height) - delay := node.getChainFinalizationDelay(chain) - if checkpoint+delay > height+1 { + if checkpoint >= height { time.Sleep(duration) continue } @@ -333,7 +330,7 @@ func (node *Node) bitcoinRPCBlocksLoop(ctx context.Context, chain byte) { var wg sync.WaitGroup for range batch { ckpt := checkpoint + 1 - if ckpt+delay > int64(height)+1 { + if ckpt > height { break } wg.Add(1) diff --git a/observer/ethereum.go b/observer/ethereum.go index 231c0f5c..370cc3b2 100644 --- a/observer/ethereum.go +++ b/observer/ethereum.go @@ -119,11 +119,6 @@ func (node *Node) ethereumNetworkInfoLoop(ctx context.Context, chain byte) { logger.Printf("ethereum.RPCGetBlockHeight(%d) => %v", chain, err) continue } - delay := node.getChainFinalizationDelay(chain) - if delay > height || delay < 1 { - panic(delay) - } - height = height + 1 - delay info, err := node.keeperStore.ReadLatestNetworkInfo(ctx, chain, time.Now()) if err != nil { panic(err) @@ -335,6 +330,9 @@ func (node *Node) ethereumConfirmPendingDeposit(ctx context.Context, deposit *De if info.Height < etx.BlockHeight { confirmations = 0 } + if confirmations < uint64(node.getChainFinalizationDelay(deposit.Chain)) { + return nil + } if !ethereum.CheckFinalization(confirmations, deposit.Chain) { return nil } @@ -429,8 +427,7 @@ func (node *Node) ethereumRPCBlocksLoop(ctx context.Context, chain byte) { continue } logger.Printf("node.ethereumReadDepositCheckpoint(%d) => %d %d", chain, checkpoint, height) - delay := node.getChainFinalizationDelay(chain) - if checkpoint+delay > height+1 { + if checkpoint >= height { time.Sleep(duration) continue } @@ -439,7 +436,7 @@ func (node *Node) ethereumRPCBlocksLoop(ctx context.Context, chain byte) { var wg sync.WaitGroup for range batch { ckpt := checkpoint + 1 - if ckpt+delay > int64(height)+1 { + if ckpt > height { break } wg.Add(1) From 78118566bb3bc38a7f9341eb59eee1a9b6486696 Mon Sep 17 00:00:00 2001 From: hundredark Date: Sat, 10 Oct 2026 17:56:14 +0800 Subject: [PATCH 02/14] revert network info loop --- observer/bitcoin.go | 5 +++++ observer/ethereum.go | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 41a900f6..6397287b 100644 --- a/observer/bitcoin.go +++ b/observer/bitcoin.go @@ -49,6 +49,11 @@ func (node *Node) bitcoinNetworkInfoLoop(ctx context.Context, chain byte) { if err != nil { continue } + delay := node.getChainFinalizationDelay(chain) + if delay > height || delay < 1 { + panic(delay) + } + height = height + 1 - delay info, err := node.keeperStore.ReadLatestNetworkInfo(ctx, chain, time.Now()) if err != nil { panic(err) diff --git a/observer/ethereum.go b/observer/ethereum.go index 370cc3b2..fef6096f 100644 --- a/observer/ethereum.go +++ b/observer/ethereum.go @@ -119,6 +119,11 @@ func (node *Node) ethereumNetworkInfoLoop(ctx context.Context, chain byte) { logger.Printf("ethereum.RPCGetBlockHeight(%d) => %v", chain, err) continue } + delay := node.getChainFinalizationDelay(chain) + if delay > height || delay < 1 { + panic(delay) + } + height = height + 1 - delay info, err := node.keeperStore.ReadLatestNetworkInfo(ctx, chain, time.Now()) if err != nil { panic(err) From 7712e1c93001b6683b0e86d7211f69f15a29cfc4 Mon Sep 17 00:00:00 2001 From: hundredark Date: Sat, 10 Oct 2026 18:03:56 +0800 Subject: [PATCH 03/14] remove duplicate confirmation check --- observer/bitcoin.go | 3 --- observer/ethereum.go | 3 --- 2 files changed, 6 deletions(-) diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 6397287b..7c301736 100644 --- a/observer/bitcoin.go +++ b/observer/bitcoin.go @@ -279,9 +279,6 @@ func (node *Node) bitcoinConfirmPendingDeposit(ctx context.Context, deposit *Dep if info.Height < output.Height { confirmations = 0 } - if confirmations < uint64(node.getChainFinalizationDelay(deposit.Chain)) { - return nil - } if !bitcoin.CheckFinalization(confirmations, output.Coinbase) { return nil } diff --git a/observer/ethereum.go b/observer/ethereum.go index fef6096f..ad0e4feb 100644 --- a/observer/ethereum.go +++ b/observer/ethereum.go @@ -335,9 +335,6 @@ func (node *Node) ethereumConfirmPendingDeposit(ctx context.Context, deposit *De if info.Height < etx.BlockHeight { confirmations = 0 } - if confirmations < uint64(node.getChainFinalizationDelay(deposit.Chain)) { - return nil - } if !ethereum.CheckFinalization(confirmations, deposit.Chain) { return nil } From 1548064b2b0f22c821555f7eb19ee4f439325839 Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 09:50:42 +0800 Subject: [PATCH 04/14] Revert pending deposit changes to match main --- observer/bitcoin.go | 5 +++-- observer/ethereum.go | 5 +++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 7c301736..5d7ff4f2 100644 --- a/observer/bitcoin.go +++ b/observer/bitcoin.go @@ -323,7 +323,8 @@ func (node *Node) bitcoinRPCBlocksLoop(ctx context.Context, chain byte) { continue } logger.Printf("node.bitcoinReadDepositCheckpoint(%d) => %d %d", chain, checkpoint, height) - if checkpoint >= height { + delay := node.getChainFinalizationDelay(chain) + if checkpoint+delay > height+1 { time.Sleep(duration) continue } @@ -332,7 +333,7 @@ func (node *Node) bitcoinRPCBlocksLoop(ctx context.Context, chain byte) { var wg sync.WaitGroup for range batch { ckpt := checkpoint + 1 - if ckpt > height { + if ckpt+delay > int64(height)+1 { break } wg.Add(1) diff --git a/observer/ethereum.go b/observer/ethereum.go index ad0e4feb..231c0f5c 100644 --- a/observer/ethereum.go +++ b/observer/ethereum.go @@ -429,7 +429,8 @@ func (node *Node) ethereumRPCBlocksLoop(ctx context.Context, chain byte) { continue } logger.Printf("node.ethereumReadDepositCheckpoint(%d) => %d %d", chain, checkpoint, height) - if checkpoint >= height { + delay := node.getChainFinalizationDelay(chain) + if checkpoint+delay > height+1 { time.Sleep(duration) continue } @@ -438,7 +439,7 @@ func (node *Node) ethereumRPCBlocksLoop(ctx context.Context, chain byte) { var wg sync.WaitGroup for range batch { ckpt := checkpoint + 1 - if ckpt > height { + if ckpt+delay > int64(height)+1 { break } wg.Add(1) From abbe4afddf65e9ed0c690022606bdcaf8e996426 Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 13:12:22 +0800 Subject: [PATCH 05/14] add pending deposit loop --- apps/bitcoin/rpc.go | 7 +- observer/bitcoin.go | 16 +- observer/coverage_test.go | 4 +- observer/ethereum.go | 53 +++-- observer/http.go | 21 +- observer/migration.go | 54 +++++ observer/node.go | 28 +++ observer/pending.go | 270 +++++++++++++++++++++++++ observer/pending_test.go | 410 ++++++++++++++++++++++++++++++++++++++ observer/schema.sql | 19 ++ observer/store.go | 90 ++++++++- 11 files changed, 940 insertions(+), 32 deletions(-) create mode 100644 observer/migration.go create mode 100644 observer/pending.go create mode 100644 observer/pending_test.go diff --git a/apps/bitcoin/rpc.go b/apps/bitcoin/rpc.go index a677117d..58a4d5b3 100644 --- a/apps/bitcoin/rpc.go +++ b/apps/bitcoin/rpc.go @@ -163,13 +163,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 } diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 5d7ff4f2..b38f81d8 100644 --- a/observer/bitcoin.go +++ b/observer/bitcoin.go @@ -380,9 +380,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 +393,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..59c2e6a4 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} diff --git a/observer/ethereum.go b/observer/ethereum.go index 231c0f5c..fdc08e8f 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 19630fe2..7627109a 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.store.ListPendingDeposits(r.Context(), int(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) if err != nil { common.RenderError(w, r, err) return diff --git a/observer/migration.go b/observer/migration.go new file mode 100644 index 00000000..e62500db --- /dev/null +++ b/observer/migration.go @@ -0,0 +1,54 @@ +package observer + +import ( + "context" + "database/sql" + "time" + + "github.com/MixinNetwork/safe/common" +) + +// 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..8089a15f 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.store.migratePendingDepositCheckpoints(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/pending.go b/observer/pending.go new file mode 100644 index 00000000..93cec9f0 --- /dev/null +++ b/observer/pending.go @@ -0,0 +1,270 @@ +package observer + +import ( + "context" + "fmt" + "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" +) + +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) + if err != nil { + panic(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 { + panic(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) + } + } + + for _, hash := range hashes { + if seen[hash] { + continue + } + tx, err := bitcoin.RPCGetTransaction(chain, rpc, hash) + if err != nil || tx == nil { + return fmt.Errorf("bitcoin.RPCGetTransaction(%s) => %v %v", hash, tx, err) + } + err = node.bitcoinRecordPendingTransaction(ctx, tx, chain) + if err != nil { + return err + } + seen[hash] = true + } + 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) + if err != nil { + panic(err) + } + } +} + +func (node *Node) scanEthereumPendingDeposits(ctx context.Context, chain byte) error { + 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.ethereumReadBlockTransfers(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) 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..6651a1cb --- /dev/null +++ b/observer/pending_test.go @@ -0,0 +1,410 @@ +package observer + +import ( + "encoding/json" + "fmt" + "math" + "math/big" + "net/http" + "net/http/httptest" + "strings" + "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 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) + } + require.JSONEq(t, `[]`, get("/deposits?chain="+chain, false).Body.String()) + 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"]) + 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 + 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{"mempool-hash"} + case "getrawtransaction": + 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") + 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") + }) + } +} + +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(pendingDepositsCols, ",")) + params := []any{chain, common.RequestStateInitial, time.Unix(0, offset)} + 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() +} From 606bbc2f3149660f6c1c4eabef9e925bd5fc21fd Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 13:23:54 +0800 Subject: [PATCH 06/14] fix tx could be nil --- apps/bitcoin/rpc.go | 19 +++++++++---- apps/bitcoin/rpc_security_test.go | 47 +++++++++++++++++++++++++++++++ keeper/deposit.go | 3 ++ observer/pending.go | 6 +++- observer/pending_test.go | 34 +++++++++++++++++++++- 5 files changed, 102 insertions(+), 7 deletions(-) diff --git a/apps/bitcoin/rpc.go b/apps/bitcoin/rpc.go index 58a4d5b3..b3c360d0 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" @@ -70,7 +72,7 @@ type RPCBlockWithTransactions struct { func RPCGetTransactionOutput(chain byte, rpc, hash string, index int64) (*RPCTransaction, *Output, error) { tx, err := RPCGetTransaction(chain, rpc, hash) - if err != nil { + if err != nil || tx == nil { return nil, nil, err } if int64(len(tx.Vout)) <= index { @@ -146,12 +148,19 @@ 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) + } return itx.Vout[tx.Vin[0].VOUT].ScriptPubKey.Address, nil } 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 @@ -419,20 +428,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..23acb4d3 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,52 @@ 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.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" diff --git a/keeper/deposit.go b/keeper/deposit.go index 40f12793..59aaa9b8 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/pending.go b/observer/pending.go index 93cec9f0..ea6aa808 100644 --- a/observer/pending.go +++ b/observer/pending.go @@ -46,9 +46,13 @@ func (node *Node) scanBitcoinPendingDeposits(ctx context.Context, chain byte, se continue } tx, err := bitcoin.RPCGetTransaction(chain, rpc, hash) - if err != nil || tx == nil { + 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 diff --git a/observer/pending_test.go b/observer/pending_test.go index 6651a1cb..53a12ade 100644 --- a/observer/pending_test.go +++ b/observer/pending_test.go @@ -116,6 +116,7 @@ func TestBitcoinPendingMempoolExcludesChange(t *testing.T) { 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 @@ -125,8 +126,16 @@ func TestBitcoinPendingMempoolExcludesChange(t *testing.T) { var result any switch q.Method { case "getrawmempool": - result = []string{"mempool-hash"} + 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{ @@ -155,6 +164,7 @@ func TestBitcoinPendingMempoolExcludesChange(t *testing.T) { 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) @@ -172,6 +182,28 @@ func TestBitcoinPendingMempoolExcludesChange(t *testing.T) { } } +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 From f5fd586dbfefb33a2d448e86248c3257427d1ca8 Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 14:00:17 +0800 Subject: [PATCH 07/14] slight fixes --- observer/http.go | 37 +++++++++++++++++++++++++++++++++++++ observer/pending_test.go | 15 +++++++++++++++ observer/store.go | 4 ++-- 3 files changed, 54 insertions(+), 2 deletions(-) diff --git a/observer/http.go b/observer/http.go index 7627109a..9b3818c0 100644 --- a/observer/http.go +++ b/observer/http.go @@ -12,6 +12,7 @@ import ( "fmt" "math/big" "net/http" + "sort" "strconv" "strings" "time" @@ -276,6 +277,42 @@ func (node *Node) httpListPendingDeposits(w http.ResponseWriter, r *http.Request common.RenderError(w, r, err) return } + initial, err := node.store.ListDeposits(r.Context(), int(chain), holder, common.RequestStateInitial, offset) + if err != nil { + common.RenderError(w, r, err) + return + } + 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 + } + 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 + }) common.RenderJSON(w, r, http.StatusOK, deposits) } diff --git a/observer/pending_test.go b/observer/pending_test.go index 53a12ade..d3175f2a 100644 --- a/observer/pending_test.go +++ b/observer/pending_test.go @@ -76,7 +76,19 @@ func TestPendingDepositConfirmationAndHTTP(t *testing.T) { 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)) @@ -98,6 +110,9 @@ func TestPendingDepositConfirmationAndHTTP(t *testing.T) { 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") diff --git a/observer/store.go b/observer/store.go index 29aab211..125acb95 100644 --- a/observer/store.go +++ b/observer/store.go @@ -392,7 +392,7 @@ func (s *SQLite3Store) ListProposedAccountsWithSig(ctx context.Context) ([]*Acco func (s *SQLite3Store) ListDeposits(ctx context.Context, chain int, holder string, state int, offset int64) ([]*Deposit, error) { query := fmt.Sprintf("SELECT %s FROM deposits WHERE chain=? AND updated_at>=?", strings.Join(depositsCols, ",")) - params := []any{chain, time.Unix(0, offset)} + params := []any{chain, time.Unix(0, offset).UTC()} if state > 0 { query += " AND state=?" params = append(params, state) @@ -1133,7 +1133,7 @@ func (s *SQLite3Store) WritePendingDepositRecord(ctx context.Context, d *Pending 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)} + params := []any{chain, common.RequestStateInitial, time.Unix(0, offset).UTC()} if holder != "" { query += " AND holder=?" params = append(params, holder) From 3e27e7d972ace97894802b75de8bd22c124f892b Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 14:16:59 +0800 Subject: [PATCH 08/14] fix test --- apps/bitcoin/common.go | 2 +- observer/observer_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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/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() From ae1dbff464affb3eb46a259d43594356475186a8 Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 14:22:03 +0800 Subject: [PATCH 09/14] update test --- apps/bitcoin/coverage_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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)) From a3cef61552648ec3153d9550e548bab8d32cc71b Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 16:36:46 +0800 Subject: [PATCH 10/14] improve bitcoin pending deposits --- observer/pending.go | 71 ++++++++++++++++++++++-- observer/pending_test.go | 117 +++++++++++++++++++++++++++++++++++++++ observer/schema.sql | 1 + observer/store.go | 50 ++++++++++++++++- 4 files changed, 234 insertions(+), 5 deletions(-) diff --git a/observer/pending.go b/observer/pending.go index ea6aa808..a5d755b1 100644 --- a/observer/pending.go +++ b/observer/pending.go @@ -2,6 +2,7 @@ package observer import ( "context" + "database/sql" "fmt" "time" @@ -12,15 +13,14 @@ import ( "github.com/shopspring/decimal" ) +const bitcoinPendingDepositGracePeriod = time.Minute * 5 + 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) - if err != nil { - panic(err) - } } } @@ -29,7 +29,7 @@ func (node *Node) scanBitcoinPendingDeposits(ctx context.Context, chain byte, se hashes, err := bitcoin.RPCGetRawMempoolHashes(rpc) logger.Printf("bitcoin.RPCGetRawMempoolHashes(%d) => %d %v", chain, len(hashes), err) if err != nil { - panic(err) + return fmt.Errorf("bitcoin.RPCGetRawMempoolHashes(%d) => %w", chain, err) } mempool := make(map[string]bool, len(hashes)) for _, hash := range hashes { @@ -41,6 +41,12 @@ func (node *Node) scanBitcoinPendingDeposits(ctx context.Context, chain byte, se } } + // 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 @@ -62,6 +68,63 @@ func (node *Node) scanBitcoinPendingDeposits(ctx context.Context, chain byte, se 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] diff --git a/observer/pending_test.go b/observer/pending_test.go index d3175f2a..a06da100 100644 --- a/observer/pending_test.go +++ b/observer/pending_test.go @@ -1,6 +1,7 @@ package observer import ( + "database/sql" "encoding/json" "fmt" "math" @@ -8,6 +9,7 @@ import ( "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "time" @@ -21,6 +23,121 @@ import ( "github.com/stretchr/testify/require" ) +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() diff --git a/observer/schema.sql b/observer/schema.sql index e39d0655..602ddae3 100644 --- a/observer/schema.sql +++ b/observer/schema.sql @@ -73,6 +73,7 @@ CREATE TABLE IF NOT EXISTS pending_deposits ( chain INTEGER NOT NULL, holder VARCHAR NOT NULL, created_at TIMESTAMP NOT NULL, + missing_since TIMESTAMP, PRIMARY KEY ('chain', 'transaction_hash', 'output_index') ); diff --git a/observer/store.go b/observer/store.go index 125acb95..0353053f 100644 --- a/observer/store.go +++ b/observer/store.go @@ -513,7 +513,7 @@ func (s *SQLite3Store) ConfirmPendingDeposit(ctx context.Context, transactionHas return fmt.Errorf("UPDATE deposits %v", err) } - query = "UPDATE pending_deposits SET state=? WHERE transaction_hash=? AND output_index=? AND chain=? AND state=?" + 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) @@ -1131,6 +1131,54 @@ func (s *SQLite3Store) WritePendingDepositRecord(ctx context.Context, d *Pending 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()} From 0ea053eb547bc9e9a6e5db892674360091233e7c Mon Sep 17 00:00:00 2001 From: hundredark Date: Tue, 20 Oct 2026 17:26:59 +0800 Subject: [PATCH 11/14] improve evm pending deposits --- observer/pending.go | 7 +++ observer/pending_test.go | 112 +++++++++++++++++++++++++++++++++++++++ observer/store.go | 15 ++++++ 3 files changed, 134 insertions(+) diff --git a/observer/pending.go b/observer/pending.go index a5d755b1..9df23730 100644 --- a/observer/pending.go +++ b/observer/pending.go @@ -14,6 +14,7 @@ import ( ) const bitcoinPendingDepositGracePeriod = time.Minute * 5 +const ethereumPendingDepositTTL = 24 * time.Hour func (node *Node) bitcoinPendingDepositsLoop(ctx context.Context, chain byte) { seen := make(map[string]bool) @@ -215,6 +216,12 @@ func (node *Node) ethereumPendingDepositsLoop(ctx context.Context, chain byte) { } 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 { diff --git a/observer/pending_test.go b/observer/pending_test.go index a06da100..19ed5767 100644 --- a/observer/pending_test.go +++ b/observer/pending_test.go @@ -23,6 +23,118 @@ import ( "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"} { diff --git a/observer/store.go b/observer/store.go index 0353053f..4abebe53 100644 --- a/observer/store.go +++ b/observer/store.go @@ -1204,3 +1204,18 @@ func (s *SQLite3Store) ListPendingDeposits(ctx context.Context, chain int, holde } 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 +} From 90a475745b6222825fd2d681c7c64d5f9f604e38 Mon Sep 17 00:00:00 2001 From: hundredark Date: Fri, 23 Oct 2026 14:45:04 +0800 Subject: [PATCH 12/14] fix issues --- apps/bitcoin/rpc.go | 5 +- apps/ethereum/rpc.go | 41 ++++++++++++-- observer/http.go | 39 +------------ observer/pending.go | 115 +++++++++++++++++++++++++++++++++++++-- observer/pending_test.go | 56 +++++++++++++++++-- observer/store.go | 32 +++++++++++ 6 files changed, 236 insertions(+), 52 deletions(-) diff --git a/apps/bitcoin/rpc.go b/apps/bitcoin/rpc.go index 49f9e847..49578320 100644 --- a/apps/bitcoin/rpc.go +++ b/apps/bitcoin/rpc.go @@ -75,9 +75,12 @@ 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 || tx == nil { + 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 } 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/observer/http.go b/observer/http.go index 95a916b9..3b44f2ff 100644 --- a/observer/http.go +++ b/observer/http.go @@ -12,7 +12,6 @@ import ( "fmt" "math/big" "net/http" - "sort" "strconv" "strings" "time" @@ -272,47 +271,11 @@ func (node *Node) httpListPendingDeposits(w http.ResponseWriter, r *http.Request } holder := r.URL.Query().Get("holder") offset, _ := strconv.ParseInt(r.URL.Query().Get("offset"), 10, 64) - deposits, err := node.store.ListPendingDeposits(r.Context(), int(chain), holder, offset) + deposits, err := node.listPendingDeposits(r.Context(), byte(chain), holder, offset) if err != nil { common.RenderError(w, r, err) return } - initial, err := node.store.ListDeposits(r.Context(), int(chain), holder, common.RequestStateInitial, offset) - if err != nil { - common.RenderError(w, r, err) - return - } - 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 - } - 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 - }) common.RenderJSON(w, r, http.StatusOK, deposits) } diff --git a/observer/pending.go b/observer/pending.go index 9df23730..9c4a8bd9 100644 --- a/observer/pending.go +++ b/observer/pending.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "sort" "time" "github.com/MixinNetwork/mixin/logger" @@ -16,6 +17,78 @@ import ( 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 + } + + var after *Deposit + for added := 0; added < 100; { + initial, err := node.store.listInitialDepositsByCreatedAt(ctx, int(chain), holder, offset, after) + if err != nil { + return nil, err + } + var sent map[string]string + switch chain { + case common.SafeChainBitcoin, common.SafeChainLitecoin: + sent, err = node.store.QueryDepositSentHashes(ctx, initial) + if err != nil { + return nil, err + } + } + for _, d := range initial { + key := depositKey{d.TransactionHash, d.OutputIndex} + if seen[key] { + continue + } + switch chain { + case common.SafeChainBitcoin, common.SafeChainLitecoin: + if node.bitcoinCheckDepositChange(ctx, d.TransactionHash, d.OutputIndex, sent[d.TransactionHash]) { + 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 + added++ + if added == 100 { + break + } + } + if len(initial) < 100 { + break + } + after = initial[len(initial)-1] + } + 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 { @@ -209,9 +282,6 @@ func (node *Node) ethereumPendingDepositsLoop(ctx context.Context, chain byte) { time.Sleep(time.Second) err := node.scanEthereumPendingDeposits(ctx, chain) logger.Printf("node.scanEthereumPendingDeposits(%d) => %v", chain, err) - if err != nil { - panic(err) - } } } @@ -260,7 +330,7 @@ func (node *Node) scanEthereumPendingDeposits(ctx context.Context, chain byte) e } func (node *Node) ethereumRecordPendingBlock(ctx context.Context, height int64, chain byte) error { - _, transfers, err := node.ethereumReadBlockTransfers(ctx, height, chain) + transfers, err := node.ethereumReadPendingBlockTransfers(ctx, height, chain) if err != nil { return err } @@ -273,6 +343,43 @@ func (node *Node) ethereumRecordPendingBlock(ctx context.Context, height int64, 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 diff --git a/observer/pending_test.go b/observer/pending_test.go index 19ed5767..54b92ab7 100644 --- a/observer/pending_test.go +++ b/observer/pending_test.go @@ -584,6 +584,7 @@ func TestEthereumPendingScansLatestBlockAndRetries(t *testing.T) { }, Data: ec.LeftPadBytes(big.NewInt(1234567890123456789).Bytes(), 32), BlockNumber: uint64(height), TxHash: ec.HexToHash(hash), BlockHash: blockHash, Index: 2} failLogs := true + readFailure := "" server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/network/assets/") { id := strings.TrimPrefix(r.URL.Path, "/network/assets/") @@ -605,21 +606,55 @@ func TestEthereumPendingScansLatestBlockAndRetries(t *testing.T) { "error": map[string]any{"code": -32000, "message": "backend unavailable"}})) return } + if q.Method == "eth_getBlockByNumber" && string(q.Params[1]) == "false" && readFailure == "header error" { + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": q.ID, + "error": map[string]any{"code": -32000, "message": "header unavailable"}})) + return + } var result any switch q.Method { case "eth_blockNumber": result = fmt.Sprintf("0x%x", height) - case "debug_traceBlockByNumber": - require.JSONEq(t, fmt.Sprintf(`"0x%x"`, height), string(q.Params[0])) - result = []*ethereum.RPCBlockCallTrace{{Result: ðereum.RPCTransactionCallTrace{ + case "debug_traceBlockByHash": + require.JSONEq(t, fmt.Sprintf("%q", blockHash.Hex()), string(q.Params[0])) + traces := []*ethereum.RPCBlockCallTrace{{TxHash: hash, Result: ðereum.RPCTransactionCallTrace{ Type: "CALL", To: f.address, From: "0x2222222222222222222222222222222222222222", Value: "0xde0b6b3a7640000", Calls: []*ethereum.RPCTransactionCallTrace{{Type: "CALL", To: f.address, Value: "0x1", Error: "execution reverted"}}, }}} + switch readFailure { + case "trace hash": + traces[0].TxHash = blockHash.Hex() + case "trace count": + traces = append(traces, traces[0]) + case "null traces": + traces = nil + case "missing trace result": + traces[0].Result = nil + } + result = traces case "eth_getBlockByNumber": result = map[string]any{"number": fmt.Sprintf("0x%x", height), "hash": blockHash.Hex(), "transactions": []any{map[string]any{"hash": hash}}} + if string(q.Params[1]) == "false" { + currentHash := blockHash.Hex() + if readFailure == "reorg" { + currentHash = "0x" + strings.Repeat("c", 64) + } + result = map[string]any{"number": fmt.Sprintf("0x%x", height), "hash": currentHash, + "timestamp": "0x1", "transactions": []string{hash}} + } case "eth_getLogs": - result = []types.Log{log} + var filter map[string]any + require.NoError(t, json.Unmarshal(q.Params[0], &filter)) + require.Equal(t, blockHash.Hex(), filter["blockHash"]) + require.NotContains(t, filter, "fromBlock") + require.NotContains(t, filter, "toBlock") + entry := log + if readFailure == "log hash" { + entry.BlockHash = ec.HexToHash(hash) + } + entry.Removed = readFailure == "removed log" + result = []types.Log{entry} case "eth_call": var call struct{ Input, Data string } require.NoError(t, json.Unmarshal(q.Params[0], &call)) @@ -653,6 +688,19 @@ func TestEthereumPendingScansLatestBlockAndRetries(t *testing.T) { require.NoError(t, err) require.Equal(t, height-1, checkpoint, "failed blocks do not advance the cursor") failLogs = false + for _, failure := range []string{"trace hash", "trace count", "null traces", "missing trace result", "log hash", "removed log", "reorg", "header error"} { + t.Run(failure, func(t *testing.T) { + readFailure = failure + require.Error(t, f.node.scanEthereumPendingDeposits(ctx, chain)) + checkpoint, err := f.node.readPendingDepositCheckpoint(ctx, chain) + require.NoError(t, err) + require.Equal(t, height-1, checkpoint) + pending, err := f.node.store.ListPendingDeposits(ctx, int(chain), "", 0) + require.NoError(t, err) + require.Empty(t, pending, "do not write observations from inconsistent block reads") + }) + } + readFailure = "" err = f.node.scanEthereumPendingDeposits(ctx, chain) require.NoError(t, err) pending, err := f.node.store.ListPendingDeposits(ctx, int(chain), "", 0) diff --git a/observer/store.go b/observer/store.go index 4abebe53..8eaafead 100644 --- a/observer/store.go +++ b/observer/store.go @@ -1205,6 +1205,38 @@ func (s *SQLite3Store) ListPendingDeposits(ctx context.Context, chain int, holde return deposits, rows.Err() } +// The pending API uses creation time; ListDeposits keeps its update-time cursor. +// after continues an internal batch when change filtering removes candidate rows. +func (s *SQLite3Store) listInitialDepositsByCreatedAt(ctx context.Context, chain int, holder string, offset int64, after *Deposit) ([]*Deposit, error) { + query := fmt.Sprintf("SELECT %s FROM deposits WHERE chain=? AND state=? AND created_at>=?", strings.Join(depositsCols, ",")) + params := []any{chain, common.RequestStateInitial, time.Unix(0, offset).UTC()} + if holder != "" { + query += " AND holder=?" + params = append(params, holder) + } + if after != nil { + query += " AND (created_at, transaction_hash, CAST(output_index AS INTEGER)) > (?, ?, ?)" + params = append(params, after.CreatedAt, after.TransactionHash, after.OutputIndex) + } + // output_index is stored as VARCHAR in deposits, but the API sorts it numerically. + query += " ORDER BY created_at ASC, transaction_hash ASC, CAST(output_index AS INTEGER) ASC LIMIT 100" + rows, err := s.db.QueryContext(ctx, query, params...) + if err != nil { + return nil, err + } + defer util.CloseOrPanic(rows) + 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) + 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: From fba4deba777e234776f193a59aaa77b0e5addf6d Mon Sep 17 00:00:00 2001 From: hundredark Date: Fri, 23 Oct 2026 14:50:29 +0800 Subject: [PATCH 13/14] fix test --- apps/bitcoin/rpc_security_test.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/bitcoin/rpc_security_test.go b/apps/bitcoin/rpc_security_test.go index 23acb4d3..53733845 100644 --- a/apps/bitcoin/rpc_security_test.go +++ b/apps/bitcoin/rpc_security_test.go @@ -52,7 +52,7 @@ func TestBitcoinRPCGetTransactionNotFound(t *testing.T) { require.NoError(t, err) require.Nil(t, tx) tx, output, err := RPCGetTransactionOutput(chain, server.URL, "missing-hash", 0) - require.NoError(t, err) + require.ErrorContains(t, err, "transaction not found: missing-hash") require.Nil(t, tx) require.Nil(t, output) _, err = RPCGetTransactionSender(chain, server.URL, &RPCTransaction{ @@ -75,6 +75,10 @@ func TestBitcoinRPCGetTransactionPreservesOtherErrors(t *testing.T) { 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) { From b4c69856043ee0caa8639d3769bab71b91380a9c Mon Sep 17 00:00:00 2001 From: hundredark Date: Fri, 23 Oct 2026 17:08:16 +0800 Subject: [PATCH 14/14] improve deposit listing --- observer/bitcoin.go | 1 + observer/coverage_test.go | 2 +- observer/http.go | 13 ++---- observer/migration.go | 76 +++++++++++++++++++++++++++++++++ observer/node.go | 2 +- observer/pending.go | 56 +++++++----------------- observer/pending_test.go | 12 ++++++ observer/schema.sql | 7 +++ observer/store.go | 89 ++++++++++++++++++++++----------------- 9 files changed, 169 insertions(+), 89 deletions(-) diff --git a/observer/bitcoin.go b/observer/bitcoin.go index 63080930..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, } diff --git a/observer/coverage_test.go b/observer/coverage_test.go index 59c2e6a4..25f4a506 100644 --- a/observer/coverage_test.go +++ b/observer/coverage_test.go @@ -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/http.go b/observer/http.go index 3b44f2ff..359098a3 100644 --- a/observer/http.go +++ b/observer/http.go @@ -287,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, common.RequestStateDone, offset) + deposits, err := node.store.listDeposits(r.Context(), int(chain), holder, common.RequestStateDone, offset, true) if err != nil { common.RenderError(w, r, err) return @@ -298,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) { @@ -866,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" @@ -882,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 index e62500db..ac16deee 100644 --- a/observer/migration.go +++ b/observer/migration.go @@ -3,11 +3,87 @@ 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() diff --git a/observer/node.go b/observer/node.go index 8089a15f..995ceae4 100644 --- a/observer/node.go +++ b/observer/node.go @@ -57,7 +57,7 @@ func NewNode(db *SQLite3Store, kd *store.SQLite3Store, conf *Configuration, keep } func (node *Node) Boot(ctx context.Context) { - err := node.store.migratePendingDepositCheckpoints(ctx) + err := node.Migrate(ctx) if err != nil { panic(err) } diff --git a/observer/pending.go b/observer/pending.go index 9c4a8bd9..5634d7cf 100644 --- a/observer/pending.go +++ b/observer/pending.go @@ -31,47 +31,23 @@ func (node *Node) listPendingDeposits(ctx context.Context, chain byte, holder st seen[depositKey{d.TransactionHash, d.OutputIndex}] = true } - var after *Deposit - for added := 0; added < 100; { - initial, err := node.store.listInitialDepositsByCreatedAt(ctx, int(chain), holder, offset, after) - if err != nil { - return nil, err - } - var sent map[string]string - switch chain { - case common.SafeChainBitcoin, common.SafeChainLitecoin: - sent, err = node.store.QueryDepositSentHashes(ctx, initial) - if err != nil { - return nil, err - } - } - for _, d := range initial { - key := depositKey{d.TransactionHash, d.OutputIndex} - if seen[key] { - continue - } - switch chain { - case common.SafeChainBitcoin, common.SafeChainLitecoin: - if node.bitcoinCheckDepositChange(ctx, d.TransactionHash, d.OutputIndex, sent[d.TransactionHash]) { - 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 - added++ - if added == 100 { - break - } - } - if len(initial) < 100 { - break + // 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 } - after = initial[len(initial)-1] + 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] diff --git a/observer/pending_test.go b/observer/pending_test.go index 54b92ab7..e713e37f 100644 --- a/observer/pending_test.go +++ b/observer/pending_test.go @@ -422,6 +422,18 @@ func TestBitcoinPendingMempoolExcludesChange(t *testing.T) { 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") + } }) } } diff --git a/observer/schema.sql b/observer/schema.sql index 602ddae3..bf5d936c 100644 --- a/observer/schema.sql +++ b/observer/schema.sql @@ -94,6 +94,7 @@ CREATE TABLE IF NOT EXISTS deposits ( holder VARCHAR NOT NULL, category INTEGER NOT NULL, request_id VARCHAR NOT NULL, + is_change BOOLEAN NOT NULL DEFAULT FALSE, created_at TIMESTAMP NOT NULL, updated_at TIMESTAMP NOT NULL, PRIMARY KEY ('transaction_hash', 'output_index') @@ -103,6 +104,11 @@ CREATE INDEX IF NOT EXISTS deposits_by_chain_state_created ON deposits(chain, st CREATE INDEX IF NOT EXISTS deposits_by_holder_asset_state_created ON deposits(holder, asset_id, state, created_at); CREATE INDEX IF NOT EXISTS deposits_by_chain_state_updated ON deposits(chain, state, updated_at); CREATE INDEX IF NOT EXISTS deposits_by_holder_asset_state_updated ON deposits(holder, asset_id, state, updated_at); +-- CREATE INDEX IF NOT EXISTS deposits_by_chain_created ON deposits(chain, created_at); +-- 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); @@ -122,6 +128,7 @@ CREATE TABLE IF NOT EXISTS transactions ( ); CREATE INDEX IF NOT EXISTS transactions_by_chain_state_created ON transactions(chain, state, created_at); +CREATE INDEX IF NOT EXISTS transactions_by_spent_hash ON transactions(spent_hash); diff --git a/observer/store.go b/observer/store.go index 8eaafead..42fdb8d0 100644 --- a/observer/store.go +++ b/observer/store.go @@ -52,6 +52,7 @@ type Deposit struct { Holder string Category byte RequestId string + IsChange bool CreatedAt time.Time UpdatedAt time.Time } @@ -128,12 +129,12 @@ var accountCols = []string{"address", "created_at", "signature", "approved_at", var assetCols = []string{"asset_id", "mixin_id", "asset_key", "symbol", "name", "decimals", "chain", "created_at"} -var depositsCols = []string{"transaction_hash", "output_index", "asset_id", "asset_address", "amount", "receiver", "sender", "state", "chain", "holder", "category", "request_id", "created_at", "updated_at"} +var depositsCols = []string{"transaction_hash", "output_index", "asset_id", "asset_address", "amount", "receiver", "sender", "state", "chain", "holder", "category", "request_id", "is_change", "created_at", "updated_at"} var pendingDepositsCols = []string{"transaction_hash", "output_index", "asset_id", "asset_address", "amount", "receiver", "sender", "state", "chain", "holder", "created_at"} func (d *Deposit) values() []any { - return []any{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} + return []any{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} } func (d *PendingDeposit) values() []any { @@ -391,8 +392,15 @@ func (s *SQLite3Store) ListProposedAccountsWithSig(ctx context.Context) ([]*Acco } func (s *SQLite3Store) ListDeposits(ctx context.Context, chain int, holder string, state int, offset int64) ([]*Deposit, error) { + return s.listDeposits(ctx, chain, holder, state, offset, false) +} + +func (s *SQLite3Store) listDeposits(ctx context.Context, chain int, holder string, state int, offset int64, excludeChange bool) ([]*Deposit, error) { query := fmt.Sprintf("SELECT %s FROM deposits WHERE chain=? AND updated_at>=?", strings.Join(depositsCols, ",")) 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) @@ -402,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 } @@ -411,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) { @@ -434,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) @@ -1205,38 +1250,6 @@ func (s *SQLite3Store) ListPendingDeposits(ctx context.Context, chain int, holde return deposits, rows.Err() } -// The pending API uses creation time; ListDeposits keeps its update-time cursor. -// after continues an internal batch when change filtering removes candidate rows. -func (s *SQLite3Store) listInitialDepositsByCreatedAt(ctx context.Context, chain int, holder string, offset int64, after *Deposit) ([]*Deposit, error) { - query := fmt.Sprintf("SELECT %s FROM deposits WHERE chain=? AND state=? AND created_at>=?", strings.Join(depositsCols, ",")) - params := []any{chain, common.RequestStateInitial, time.Unix(0, offset).UTC()} - if holder != "" { - query += " AND holder=?" - params = append(params, holder) - } - if after != nil { - query += " AND (created_at, transaction_hash, CAST(output_index AS INTEGER)) > (?, ?, ?)" - params = append(params, after.CreatedAt, after.TransactionHash, after.OutputIndex) - } - // output_index is stored as VARCHAR in deposits, but the API sorts it numerically. - query += " ORDER BY created_at ASC, transaction_hash ASC, CAST(output_index AS INTEGER) ASC LIMIT 100" - rows, err := s.db.QueryContext(ctx, query, params...) - if err != nil { - return nil, err - } - defer util.CloseOrPanic(rows) - 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) - 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: