diff --git a/src/controller/admin/rpc_controller.go b/src/controller/admin/rpc_controller.go index 61431b1..e7c0f18 100644 --- a/src/controller/admin/rpc_controller.go +++ b/src/controller/admin/rpc_controller.go @@ -186,9 +186,9 @@ func (c *BaseAdminController) DeleteRpcNode(ctx echo.Context) error { return c.SucJson(ctx, nil) } -// HealthCheckRpcNode performs an on-demand probe and writes the result. -// For HTTP/WS/lite endpoints this is a TCP-level check against the configured -// URL host. TON lite rows usually point to a global.config.json HTTPS URL. +// HealthCheckRpcNode performs an on-demand capability probe and writes the +// result. EVM nodes are checked with the same log query/subscription required +// by payment recognition; other node types use a reachability probe. // @Summary Health check RPC node // @Description Perform an on-demand health probe on an RPC node // @Tags Admin RPC Nodes @@ -210,7 +210,7 @@ func (c *BaseAdminController) HealthCheckRpcNode(ctx echo.Context) error { if row.ID == 0 { return c.FailJson(ctx, constant.RpcNodeNotFoundErr) } - status, latency := task.ProbeNode(row.Url) + status, latency := task.ProbeRpcNode(*row) if err := data.UpdateRpcNodeHealth(id, status, latency); err != nil { return c.FailJson(ctx, err) } diff --git a/src/model/dao/mdb_sqlite_pure.go b/src/model/dao/mdb_sqlite_pure.go index 2f1e9ec..3554811 100644 --- a/src/model/dao/mdb_sqlite_pure.go +++ b/src/model/dao/mdb_sqlite_pure.go @@ -8,7 +8,10 @@ import ( ) func openDB(dsn string, cfg *gorm.Config) (*gorm.DB, error) { - db, err := gorm.Open(sqlite.Open(dsn+"?_journal_mode=WAL&_busy_timeout=5000"), cfg) + // modernc.org/sqlite applies connection-local PRAGMAs through repeated + // _pragma query parameters. The old _busy_timeout form was ignored by the + // pure-Go driver, so pooled connections failed immediately with SQLITE_BUSY. + db, err := gorm.Open(sqlite.Open(dsn+"?_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)&_pragma=busy_timeout(5000)"), cfg) if err != nil { return nil, err } diff --git a/src/model/dao/mdb_sqlite_pure_test.go b/src/model/dao/mdb_sqlite_pure_test.go new file mode 100644 index 0000000..ee32596 --- /dev/null +++ b/src/model/dao/mdb_sqlite_pure_test.go @@ -0,0 +1,47 @@ +//go:build !sqlite_cgo + +package dao + +import ( + "context" + "database/sql" + "path/filepath" + "testing" + + "gorm.io/gorm" +) + +func TestPureSQLiteBusyTimeoutAppliesToEveryConnection(t *testing.T) { + db, err := openDB(filepath.Join(t.TempDir(), "busy-timeout.db"), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + sqlDB, err := configureSQLite(db, 2) + if err != nil { + t.Fatalf("configure sqlite: %v", err) + } + defer sqlDB.Close() + sqlDB.SetMaxIdleConns(2) + + ctx := context.Background() + conn1, err := sqlDB.Conn(ctx) + if err != nil { + t.Fatalf("first connection: %v", err) + } + defer conn1.Close() + conn2, err := sqlDB.Conn(ctx) + if err != nil { + t.Fatalf("second connection: %v", err) + } + defer conn2.Close() + + for i, conn := range []*sql.Conn{conn1, conn2} { + var timeout int + if err := conn.QueryRowContext(ctx, "PRAGMA busy_timeout").Scan(&timeout); err != nil { + t.Fatalf("connection %d busy_timeout: %v", i+1, err) + } + if timeout != 5000 { + t.Fatalf("connection %d busy_timeout = %d, want 5000", i+1, timeout) + } + } +} diff --git a/src/model/dao/mdb_table_init.go b/src/model/dao/mdb_table_init.go index 2f294f0..adb4473 100644 --- a/src/model/dao/mdb_table_init.go +++ b/src/model/dao/mdb_table_init.go @@ -144,7 +144,8 @@ func defaultRpcNodes() []mdb.RpcNode { {Network: mdb.NetworkTron, Url: "https://api.trongrid.io", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkEthereum, Url: "wss://ethereum.publicnode.com", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkSolana, Url: "https://api.mainnet-beta.solana.com", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, - {Network: mdb.NetworkBsc, Url: "wss://bsc.drpc.org", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, + {Network: mdb.NetworkBsc, Url: "wss://bsc-rpc.publicnode.com", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, + {Network: mdb.NetworkBsc, Url: "https://bsc.rpc.blxrbdn.com", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkPolygon, Url: "wss://polygon-bor-rpc.publicnode.com", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkPlasma, Url: "wss://rpc.plasma.to", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkTon, Url: "https://ton-blockchain.github.io/global.config.json", Type: mdb.RpcNodeTypeLite, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, diff --git a/src/model/dao/runtime_sqlite.go b/src/model/dao/runtime_sqlite.go index 1f1577a..a01bef8 100644 --- a/src/model/dao/runtime_sqlite.go +++ b/src/model/dao/runtime_sqlite.go @@ -27,14 +27,10 @@ func RuntimeInit() error { return err } - concurrency := config.GetQueueConcurrency() - if concurrency < 2 { - concurrency = 2 - } - if concurrency > 16 { - concurrency = 16 - } - if _, err = configureSQLite(RuntimeDB, concurrency); err != nil { + // The runtime database is write-heavy (transaction locks and scan cursors). + // Serialize those writes through one connection; queue_concurrency controls + // workers, not the number of concurrent SQLite writers. + if _, err = configureSQLite(RuntimeDB, 1); err != nil { color.Red.Printf("[runtime_db] sqlite connDB err:%s", err.Error()) return err } diff --git a/src/mq/worker.go b/src/mq/worker.go index f4f7395..b383390 100644 --- a/src/mq/worker.go +++ b/src/mq/worker.go @@ -261,11 +261,15 @@ func isCallbackAck(body []byte) bool { } func cleanupExpiredTransactionLocks() { - if err := data.CleanupExpiredTransactionLocks(); err != nil { + if err := cleanupExpiredTransactionLocksWith(data.CleanupExpiredTransactionLocks); err != nil { log.Sugar.Errorf("[mq] cleanup expired transaction locks failed: %v", err) } } +func cleanupExpiredTransactionLocksWith(fn func() error) error { + return withSQLiteBusyRetry(fn) +} + func withSQLiteBusyRetry(fn func() error) error { var err error for attempt := 1; attempt <= sqliteBusyRetryAttempts; attempt++ { diff --git a/src/mq/worker_test.go b/src/mq/worker_test.go index b808680..c6fac66 100644 --- a/src/mq/worker_test.go +++ b/src/mq/worker_test.go @@ -2,6 +2,7 @@ package mq import ( "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -17,6 +18,23 @@ import ( "github.com/GMWalletApp/epusdt/util/sign" ) +func TestCleanupExpiredTransactionLocksRetriesSQLiteBusy(t *testing.T) { + attempts := 0 + err := cleanupExpiredTransactionLocksWith(func() error { + attempts++ + if attempts < sqliteBusyRetryAttempts { + return errors.New("database is locked (SQLITE_BUSY)") + } + return nil + }) + if err != nil { + t.Fatalf("cleanup retry: %v", err) + } + if attempts != sqliteBusyRetryAttempts { + t.Fatalf("attempts = %d, want %d", attempts, sqliteBusyRetryAttempts) + } +} + func TestProcessExpiredOrdersExpiresWaitingOrdersAndReleasesLocks(t *testing.T) { cleanup := testutil.SetupTestDatabases(t) defer cleanup() diff --git a/src/task/listen_bsc.go b/src/task/listen_bsc.go index ce11482..5df4645 100644 --- a/src/task/listen_bsc.go +++ b/src/task/listen_bsc.go @@ -12,7 +12,6 @@ import ( "github.com/GMWalletApp/epusdt/model/service" "github.com/GMWalletApp/epusdt/util/log" - "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" @@ -77,10 +76,7 @@ func runBscListener(contracts []common.Address) { } log.Sugar.Infof("[BSC-WS] connecting using WSS node %s watching %d contract(s)", data.RpcNodeLogLabel(wsNode), len(contracts)) - query := ethereum.FilterQuery{ - Addresses: contracts, - Topics: evmTransferTopics(recipientTopics), - } + query := evmLiveFilterQuery(contracts, recipientTopics) runEvmWsLogListener(ctx, mdb.NetworkBsc, "[BSC-WS]", wsNode, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { diff --git a/src/task/listen_chain_common.go b/src/task/listen_chain_common.go index 0b92598..d8dd5a1 100644 --- a/src/task/listen_chain_common.go +++ b/src/task/listen_chain_common.go @@ -2,6 +2,7 @@ package task import ( "context" + "math/big" "sort" "strings" "time" @@ -10,7 +11,9 @@ import ( "github.com/GMWalletApp/epusdt/model/mdb" "github.com/GMWalletApp/epusdt/util/log" + "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/rpc" ) const evmNodeDialTimeout = 10 * time.Second @@ -147,6 +150,17 @@ func evmTransferTopics(recipientTopics []common.Hash) [][]common.Hash { return [][]common.Hash{{transferEventHash}, nil, recipientTopics} } +// evmLiveFilterQuery starts a WebSocket subscription at the current head. +// go-ethereum otherwise serializes a nil FromBlock as 0x0, which makes public +// RPC providers reject the subscription as an oversized historical range. +func evmLiveFilterQuery(contracts []common.Address, recipientTopics []common.Hash) ethereum.FilterQuery { + return ethereum.FilterQuery{ + FromBlock: big.NewInt(int64(rpc.LatestBlockNumber)), + Addresses: contracts, + Topics: evmTransferTopics(recipientTopics), + } +} + // resolveChainWsURL picks a healthy WS endpoint from rpc_nodes for the // given network. If no enabled node is configured, the caller skips the // current listener run so admin-side disabled/deleted rows are respected. diff --git a/src/task/listen_chain_common_test.go b/src/task/listen_chain_common_test.go index d15d8aa..05ace82 100644 --- a/src/task/listen_chain_common_test.go +++ b/src/task/listen_chain_common_test.go @@ -11,6 +11,7 @@ import ( "github.com/GMWalletApp/epusdt/model/mdb" epLog "github.com/GMWalletApp/epusdt/util/log" "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/rpc" "go.uber.org/zap" ) @@ -175,6 +176,16 @@ func TestEvmHeaderBlockHeightRecordsRuntimeHeight(t *testing.T) { } } +func TestEvmLiveFilterQueryStartsAtLatest(t *testing.T) { + query := evmLiveFilterQuery(nil, nil) + if query.FromBlock == nil || !query.FromBlock.IsInt64() { + t.Fatalf("FromBlock = %v, want latest", query.FromBlock) + } + if got := query.FromBlock.Int64(); got != int64(rpc.LatestBlockNumber) { + t.Fatalf("FromBlock = %d, want %d (latest)", got, rpc.LatestBlockNumber) + } +} + func TestShouldRefreshEvmLatestHeaderAfterIdleInterval(t *testing.T) { lastUpdate := time.Now() diff --git a/src/task/listen_eth.go b/src/task/listen_eth.go index 8245929..3d01b9e 100644 --- a/src/task/listen_eth.go +++ b/src/task/listen_eth.go @@ -12,7 +12,6 @@ import ( "github.com/GMWalletApp/epusdt/model/service" "github.com/GMWalletApp/epusdt/util/log" - "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" @@ -82,10 +81,7 @@ func runEthereumListener(contracts []common.Address) { } log.Sugar.Infof("[ETH-WS] connecting using WSS node %s watching %d contract(s)", data.RpcNodeLogLabel(wsNode), len(contracts)) - query := ethereum.FilterQuery{ - Addresses: contracts, - Topics: evmTransferTopics(recipientTopics), - } + query := evmLiveFilterQuery(contracts, recipientTopics) runEvmWsLogListener(ctx, mdb.NetworkEthereum, "[ETH-WS]", wsNode, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { diff --git a/src/task/listen_plasma.go b/src/task/listen_plasma.go index 5697d18..5dc3e55 100644 --- a/src/task/listen_plasma.go +++ b/src/task/listen_plasma.go @@ -12,7 +12,6 @@ import ( "github.com/GMWalletApp/epusdt/model/service" "github.com/GMWalletApp/epusdt/util/log" - "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" @@ -76,10 +75,7 @@ func runPlasmaListener(contracts []common.Address) { } log.Sugar.Infof("[PLASMA-WS] connecting using WSS node %s watching %d contract(s)", data.RpcNodeLogLabel(wsNode), len(contracts)) - query := ethereum.FilterQuery{ - Addresses: contracts, - Topics: evmTransferTopics(recipientTopics), - } + query := evmLiveFilterQuery(contracts, recipientTopics) runEvmWsLogListener(ctx, mdb.NetworkPlasma, "[PLASMA-WS]", wsNode, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { diff --git a/src/task/listen_polygon.go b/src/task/listen_polygon.go index a6be41e..d2af502 100644 --- a/src/task/listen_polygon.go +++ b/src/task/listen_polygon.go @@ -12,7 +12,6 @@ import ( "github.com/GMWalletApp/epusdt/model/service" "github.com/GMWalletApp/epusdt/util/log" - "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" @@ -76,10 +75,7 @@ func runPolygonListener(contracts []common.Address) { } log.Sugar.Infof("[POLYGON-WS] connecting using WSS node %s watching %d contract(s)", data.RpcNodeLogLabel(wsNode), len(contracts)) - query := ethereum.FilterQuery{ - Addresses: contracts, - Topics: evmTransferTopics(recipientTopics), - } + query := evmLiveFilterQuery(contracts, recipientTopics) runEvmWsLogListener(ctx, mdb.NetworkPolygon, "[POLYGON-WS]", wsNode, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { diff --git a/src/task/rpc_health_job.go b/src/task/rpc_health_job.go index c133e2d..56d5059 100644 --- a/src/task/rpc_health_job.go +++ b/src/task/rpc_health_job.go @@ -1,6 +1,9 @@ package task import ( + "context" + "fmt" + "math/big" "net" "net/url" "strings" @@ -10,6 +13,10 @@ import ( "github.com/GMWalletApp/epusdt/model/data" "github.com/GMWalletApp/epusdt/model/mdb" "github.com/GMWalletApp/epusdt/util/log" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" ) const rpcProbeTimeout = 5 * time.Second @@ -38,7 +45,10 @@ func (r RpcHealthJob) Run() { wg.Add(1) go func(n mdb.RpcNode) { defer wg.Done() - status, latency := ProbeNode(n.Url) + status, latency, probeErr := probeRpcNode(n) + if status == mdb.RpcNodeStatusDown && n.Status != mdb.RpcNodeStatusDown { + log.Sugar.Errorf("[rpc-health] capability probe failed node=%s err=%v", data.RpcNodeLogLabel(n), probeErr) + } if err := data.UpdateRpcNodeHealth(n.ID, status, latency); err != nil { log.Sugar.Warnf("[rpc-health] update node %d err=%v", n.ID, err) } @@ -47,6 +57,119 @@ func (r RpcHealthJob) Run() { wg.Wait() } +// ProbeRpcNode verifies the capability required by the configured node. EVM +// HTTP nodes must serve both block height and the exact historical log filter +// used by the backfill scanner; EVM WS nodes must accept that log subscription. +// Other node types retain the legacy TCP reachability probe. +func ProbeRpcNode(node mdb.RpcNode) (string, int) { + status, latency, _ := probeRpcNode(node) + return status, latency +} + +func probeRpcNode(node mdb.RpcNode) (string, int, error) { + node.Network = strings.ToLower(strings.TrimSpace(node.Network)) + node.Type = strings.ToLower(strings.TrimSpace(node.Type)) + if isEvmRpcNetwork(node.Network) && (node.Type == mdb.RpcNodeTypeHttp || node.Type == mdb.RpcNodeTypeWs) { + start := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), rpcProbeTimeout) + defer cancel() + if err := probeEvmRpcNode(ctx, node); err != nil { + return mdb.RpcNodeStatusDown, -1, err + } + return mdb.RpcNodeStatusOk, int(time.Since(start).Milliseconds()), nil + } + + addr, err := ParseAddress(node.Url) + if err != nil { + return mdb.RpcNodeStatusDown, -1, err + } + dur, err := MeasureTCPDial(addr, rpcProbeTimeout) + if err != nil { + return mdb.RpcNodeStatusDown, -1, err + } + return mdb.RpcNodeStatusOk, int(dur.Milliseconds()), nil +} + +func isEvmRpcNetwork(network string) bool { + switch network { + case mdb.NetworkEthereum, mdb.NetworkBsc, mdb.NetworkPolygon, mdb.NetworkPlasma: + return true + default: + return false + } +} + +func probeEvmRpcNode(ctx context.Context, node mdb.RpcNode) error { + client, err := ethclient.DialContext(ctx, strings.TrimSpace(node.Url)) + if err != nil { + return fmt.Errorf("dial: %w", err) + } + defer client.Close() + + head, err := client.BlockNumber(ctx) + if err != nil { + return fmt.Errorf("block number: %w", err) + } + query, hasFilter, err := evmHealthFilterQuery(node.Network, node.Type, head) + if err != nil { + return err + } + if !hasFilter { + return nil + } + + if node.Type == mdb.RpcNodeTypeWs { + logsCh := make(chan types.Log) + sub, err := client.SubscribeFilterLogs(ctx, query, logsCh) + if err != nil { + return fmt.Errorf("subscribe logs: %w", err) + } + sub.Unsubscribe() + return nil + } + + if _, err := client.FilterLogs(ctx, query); err != nil { + return fmt.Errorf("filter logs: %w", err) + } + return nil +} + +func evmHealthFilterQuery(network, nodeType string, head uint64) (ethereum.FilterQuery, bool, error) { + contracts := loadChainTokenContracts(network, "") + recipients := loadEvmRecipientTopics(network, "") + if len(contracts) == 0 || len(recipients) == 0 { + return ethereum.FilterQuery{}, false, nil + } + + query := ethereum.FilterQuery{ + Addresses: contracts, + Topics: evmTransferTopics(recipients), + } + if nodeType != mdb.RpcNodeTypeHttp { + return evmLiveFilterQuery(contracts, recipients), true, nil + } + if head > uint64(1<<63-1) { + return ethereum.FilterQuery{}, false, fmt.Errorf("block number exceeds int64 range") + } + + headBlock := int64(head) + fromBlock := headBlock + cursor, err := data.GetEvmScanCursor(network) + if err != nil { + return ethereum.FilterQuery{}, false, fmt.Errorf("load scan cursor: %w", err) + } + if cursor.ID > 0 && cursor.LastBlock >= 0 && cursor.LastBlock < headBlock { + fromBlock = cursor.LastBlock + 1 + } + toBlock := fromBlock + evmBackfillBatchSize(network) - 1 + if toBlock > headBlock { + toBlock = headBlock + } + query.FromBlock = big.NewInt(fromBlock) + query.ToBlock = big.NewInt(toBlock) + return query, true, nil +} + // ProbeNode does a TCP dial to the RPC URL and returns (status, latencyMs). // Exported so the admin controller can reuse it without duplicating logic. func ProbeNode(rawURL string) (string, int) { diff --git a/src/task/rpc_health_job_test.go b/src/task/rpc_health_job_test.go index 7ae2f71..5a8c7df 100644 --- a/src/task/rpc_health_job_test.go +++ b/src/task/rpc_health_job_test.go @@ -1,12 +1,21 @@ package task import ( + "encoding/json" "net" + "net/http" + "net/http/httptest" "strconv" + "strings" + "sync/atomic" "testing" "time" + "github.com/GMWalletApp/epusdt/internal/testutil" + "github.com/GMWalletApp/epusdt/model/dao" + "github.com/GMWalletApp/epusdt/model/data" "github.com/GMWalletApp/epusdt/model/mdb" + "github.com/gorilla/websocket" ) // --------------- ParseAddress --------------- @@ -192,3 +201,199 @@ func TestProbeNode_InvalidURL(t *testing.T) { t.Fatalf("want -1, got %d", latency) } } + +func TestProbeRpcNode_EvmHTTPRequiresHistoricalLogs(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + seedEvmHealthProbeState(t) + + var sawLogs bool + server := newEvmHealthHTTPServer(t, func(method string, params json.RawMessage) (interface{}, *rpcHealthTestError) { + switch method { + case "eth_blockNumber": + return "0x100", nil + case "eth_getLogs": + sawLogs = true + var filters []map[string]interface{} + if err := json.Unmarshal(params, &filters); err != nil || len(filters) != 1 { + t.Fatalf("decode eth_getLogs params: %v params=%s", err, params) + } + if got := filters[0]["fromBlock"]; got != "0x65" { + t.Fatalf("fromBlock = %v, want 0x65", got) + } + if got := filters[0]["toBlock"]; got != "0x100" { + t.Fatalf("toBlock = %v, want 0x100", got) + } + return []interface{}{}, nil + default: + return nil, &rpcHealthTestError{Code: -32601, Message: "method not found"} + } + }) + defer server.Close() + + status, latency := ProbeRpcNode(mdb.RpcNode{Network: mdb.NetworkBsc, Type: mdb.RpcNodeTypeHttp, Url: server.URL}) + if status != mdb.RpcNodeStatusOk { + t.Fatalf("status = %s, want ok", status) + } + if latency < 0 { + t.Fatalf("latency = %d, want >= 0", latency) + } + if !sawLogs { + t.Fatal("probe did not call eth_getLogs") + } +} + +func TestProbeRpcNode_EvmHTTPRejectsBlockOnlyEndpoint(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + seedEvmHealthProbeState(t) + + server := newEvmHealthHTTPServer(t, func(method string, _ json.RawMessage) (interface{}, *rpcHealthTestError) { + if method == "eth_blockNumber" { + return "0x100", nil + } + return nil, &rpcHealthTestError{Code: -32005, Message: "eth_getLogs disabled"} + }) + defer server.Close() + + status, latency := ProbeRpcNode(mdb.RpcNode{Network: mdb.NetworkBsc, Type: mdb.RpcNodeTypeHttp, Url: server.URL}) + if status != mdb.RpcNodeStatusDown || latency != -1 { + t.Fatalf("probe = (%s, %d), want (down, -1)", status, latency) + } +} + +func TestProbeRpcNode_EvmWSRequiresLogSubscription(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + seedEvmHealthProbeState(t) + + server, sawSubscribe := newEvmHealthWSServer(t, false) + defer server.Close() + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + status, latency := ProbeRpcNode(mdb.RpcNode{Network: mdb.NetworkBsc, Type: mdb.RpcNodeTypeWs, Url: wsURL}) + if status != mdb.RpcNodeStatusOk || latency < 0 { + t.Fatalf("probe = (%s, %d), want ok", status, latency) + } + if !sawSubscribe.Load() { + t.Fatal("probe did not call eth_subscribe") + } +} + +func TestProbeRpcNode_EvmWSRejectsSubscriptionQuotaError(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + seedEvmHealthProbeState(t) + + server, _ := newEvmHealthWSServer(t, true) + defer server.Close() + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + + status, latency := ProbeRpcNode(mdb.RpcNode{Network: mdb.NetworkBsc, Type: mdb.RpcNodeTypeWs, Url: wsURL}) + if status != mdb.RpcNodeStatusDown || latency != -1 { + t.Fatalf("probe = (%s, %d), want (down, -1)", status, latency) + } +} + +type rpcHealthTestRequest struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` +} + +type rpcHealthTestError struct { + Code int `json:"code"` + Message string `json:"message"` +} + +type rpcHealthTestResponse struct { + JSONRPC string `json:"jsonrpc"` + ID json.RawMessage `json:"id"` + Result interface{} `json:"result,omitempty"` + Error *rpcHealthTestError `json:"error,omitempty"` +} + +func seedEvmHealthProbeState(t *testing.T) { + t.Helper() + if err := dao.Mdb.Create(&mdb.WalletAddress{ + Network: mdb.NetworkBsc, + Address: "0x1111111111111111111111111111111111111111", + Status: mdb.TokenStatusEnable, + }).Error; err != nil { + t.Fatalf("seed wallet: %v", err) + } + if err := data.UpsertEvmScanCursor(mdb.NetworkBsc, 100); err != nil { + t.Fatalf("seed cursor: %v", err) + } +} + +func newEvmHealthHTTPServer(t *testing.T, handle func(string, json.RawMessage) (interface{}, *rpcHealthTestError)) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer r.Body.Close() + var req rpcHealthTestRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("decode request: %v", err) + http.Error(w, "bad request", http.StatusBadRequest) + return + } + result, rpcErr := handle(req.Method, req.Params) + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(rpcHealthTestResponse{JSONRPC: "2.0", ID: req.ID, Result: result, Error: rpcErr}); err != nil { + t.Errorf("encode response: %v", err) + } + })) +} + +func newEvmHealthWSServer(t *testing.T, rejectSubscription bool) (*httptest.Server, *atomic.Bool) { + t.Helper() + sawSubscribe := new(atomic.Bool) + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrade websocket: %v", err) + return + } + defer conn.Close() + for { + _, payload, err := conn.ReadMessage() + if err != nil { + return + } + var req rpcHealthTestRequest + if err = json.Unmarshal(payload, &req); err != nil { + t.Errorf("decode websocket request: %v", err) + return + } + resp := rpcHealthTestResponse{JSONRPC: "2.0", ID: req.ID} + switch req.Method { + case "eth_blockNumber": + resp.Result = "0x100" + case "eth_subscribe": + sawSubscribe.Store(true) + if rejectSubscription { + resp.Error = &rpcHealthTestError{Code: 15, Message: "public endpoint rate limit"} + } else { + var params []json.RawMessage + var filter map[string]interface{} + if err = json.Unmarshal(req.Params, ¶ms); err != nil || len(params) != 2 { + resp.Error = &rpcHealthTestError{Code: -32602, Message: "invalid subscription params"} + } else if err = json.Unmarshal(params[1], &filter); err != nil || filter["fromBlock"] != "latest" { + resp.Error = &rpcHealthTestError{Code: -32602, Message: "exceed maximum block range"} + } else { + resp.Result = "0xsubscription" + } + } + case "eth_unsubscribe": + resp.Result = true + default: + resp.Error = &rpcHealthTestError{Code: -32601, Message: "method not found"} + } + if err = conn.WriteJSON(resp); err != nil { + return + } + } + })) + return server, sawSubscribe +}