From 91936d580f9c793bb503f0114cca46ca0321cab1 Mon Sep 17 00:00:00 2001 From: line-6000 <154492442+line-6000@users.noreply.github.com> Date: Thu, 23 Apr 2026 02:08:18 +0800 Subject: [PATCH] refactor(rpc): streamline RPC node selection and remove hardcoded defaults --- src/config/config.go | 14 ---- src/model/data/rpc_node_data.go | 17 +++-- src/model/data/rpc_node_data_test.go | 101 +++++++++++++++++++++++++++ src/model/service/sol_task.go | 20 +++++- src/model/service/sol_task_test.go | 74 +++++++++++++++++--- src/model/service/task_service.go | 35 ++++++---- src/model/service/task_tron_test.go | 35 +++------- src/task/listen_bsc.go | 7 +- src/task/listen_chain_common.go | 20 ++++-- src/task/listen_chain_common_test.go | 67 ++++++++++++++++++ src/task/listen_eth.go | 23 ++---- src/task/listen_plasma.go | 7 +- src/task/listen_polygon.go | 7 +- 13 files changed, 326 insertions(+), 101 deletions(-) create mode 100644 src/model/data/rpc_node_data_test.go create mode 100644 src/task/listen_chain_common_test.go diff --git a/src/config/config.go b/src/config/config.go index 8d1d8be..1bf2a39 100644 --- a/src/config/config.go +++ b/src/config/config.go @@ -36,7 +36,6 @@ var ( TgManage int64 UsdtRate float64 RateApiUrl string - TRON_GRID_API_KEY string BuildVersion = "0.0.0-dev" BuildCommit = "none" BuildDate = "unknown" @@ -79,7 +78,6 @@ func Init() { TgManage = viper.GetInt64("tg_manage") RateApiUrl = GetRateApiUrl() - TRON_GRID_API_KEY = viper.GetString("tron_grid_api_key") } func mustMkdir(path string) { @@ -398,15 +396,3 @@ func GetCallbackRetryBaseDuration() time.Duration { } return time.Duration(seconds) * time.Second } - -func GetSolanaRpcUrl() string { - rpcUrl := viper.GetString("solana_rpc_url") - if rpcUrl == "" { - return "https://api.mainnet-beta.solana.com" - } - return rpcUrl -} - -func GetEthereumWsUrl() string { - return strings.TrimSpace(viper.GetString("ethereum_ws_url")) -} diff --git a/src/model/data/rpc_node_data.go b/src/model/data/rpc_node_data.go index 701b105..dafd2e1 100644 --- a/src/model/data/rpc_node_data.go +++ b/src/model/data/rpc_node_data.go @@ -47,8 +47,8 @@ func DeleteRpcNodeByID(id uint64) error { // SelectRpcNode picks a healthy RPC endpoint for a (network, type) pair. // Strategy: weighted random among rows where enabled=true AND status=ok. -// Falls back to any enabled row when none are healthy yet (bootstrap case -// where the health check hasn't run). Returns nil if nothing is defined. +// Falls back to enabled rows with status=unknown when no health check has +// run yet. Explicitly down rows are not selected. func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) { var rows []mdb.RpcNode err := dao.Mdb.Model(&mdb.RpcNode{}). @@ -62,15 +62,22 @@ func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) { if len(rows) == 0 { return nil, nil } - healthy := rows[:0] + healthy := make([]mdb.RpcNode, 0, len(rows)) + bootstrap := make([]mdb.RpcNode, 0, len(rows)) for _, r := range rows { - if r.Status == mdb.RpcNodeStatusOk { + switch r.Status { + case mdb.RpcNodeStatusOk: healthy = append(healthy, r) + case "", mdb.RpcNodeStatusUnknown: + bootstrap = append(bootstrap, r) } } candidates := healthy if len(candidates) == 0 { - candidates = rows // bootstrap: no health check has run yet + candidates = bootstrap + } + if len(candidates) == 0 { + return nil, nil } return pickWeighted(candidates), nil } diff --git a/src/model/data/rpc_node_data_test.go b/src/model/data/rpc_node_data_test.go new file mode 100644 index 0000000..d5638ae --- /dev/null +++ b/src/model/data/rpc_node_data_test.go @@ -0,0 +1,101 @@ +package data + +import ( + "testing" + + "github.com/assimon/luuu/internal/testutil" + "github.com/assimon/luuu/model/dao" + "github.com/assimon/luuu/model/mdb" +) + +func TestSelectRpcNodeUsesHealthyRow(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: "https://unknown.example.com", + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusUnknown, + }).Error; err != nil { + t.Fatalf("seed unknown rpc_node: %v", err) + } + if err := dao.Mdb.Create(&mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: "https://ok.example.com", + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusOk, + }).Error; err != nil { + t.Fatalf("seed ok rpc_node: %v", err) + } + + got, err := SelectRpcNode(mdb.NetworkSolana, mdb.RpcNodeTypeHttp) + if err != nil { + t.Fatalf("SelectRpcNode(): %v", err) + } + if got == nil || got.Url != "https://ok.example.com" { + t.Fatalf("SelectRpcNode() = %#v, want ok row", got) + } +} + +func TestSelectRpcNodeFallsBackToUnknownOnly(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: "https://unknown.example.com", + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusUnknown, + }).Error; err != nil { + t.Fatalf("seed unknown rpc_node: %v", err) + } + if err := dao.Mdb.Create(&mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: "https://down.example.com", + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusDown, + }).Error; err != nil { + t.Fatalf("seed down rpc_node: %v", err) + } + + got, err := SelectRpcNode(mdb.NetworkSolana, mdb.RpcNodeTypeHttp) + if err != nil { + t.Fatalf("SelectRpcNode(): %v", err) + } + if got == nil || got.Url != "https://unknown.example.com" { + t.Fatalf("SelectRpcNode() = %#v, want unknown row", got) + } +} + +func TestSelectRpcNodeIgnoresDownRows(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: "https://down.example.com", + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusDown, + }).Error; err != nil { + t.Fatalf("seed down rpc_node: %v", err) + } + + got, err := SelectRpcNode(mdb.NetworkSolana, mdb.RpcNodeTypeHttp) + if err != nil { + t.Fatalf("SelectRpcNode(): %v", err) + } + if got != nil { + t.Fatalf("SelectRpcNode() = %#v, want nil", got) + } +} diff --git a/src/model/service/sol_task.go b/src/model/service/sol_task.go index 68a996c..d32b57d 100644 --- a/src/model/service/sol_task.go +++ b/src/model/service/sol_task.go @@ -281,6 +281,21 @@ type solSignatureResult struct { BlockTime *int64 `json:"blockTime"` } +func resolveSolanaRpcURL() (string, error) { + node, err := data.SelectRpcNode(mdb.NetworkSolana, mdb.RpcNodeTypeHttp) + if err != nil { + return "", err + } + if node == nil || node.ID == 0 { + return "", fmt.Errorf("no enabled %s %s RPC node configured in rpc_nodes", mdb.NetworkSolana, mdb.RpcNodeTypeHttp) + } + rpcURL := strings.TrimSpace(node.Url) + if rpcURL == "" { + return "", fmt.Errorf("rpc_nodes id=%d has empty url", node.ID) + } + return rpcURL, nil +} + // SolRetryClient 发送 Solana JSON-RPC 请求，自动重试 func SolRetryClient(method string, params []interface{}) ([]byte, error) { client := resty.New() @@ -297,7 +312,10 @@ func SolRetryClient(method string, params []interface{}) ([]byte, error) { return false }) - rpcUrl := config.GetSolanaRpcUrl() + rpcUrl, err := resolveSolanaRpcURL() + if err != nil { + return nil, err + } resp, err := client.R(). SetHeader("Content-Type", "application/json"). diff --git a/src/model/service/sol_task_test.go b/src/model/service/sol_task_test.go index 924a2a6..cdb3359 100644 --- a/src/model/service/sol_task_test.go +++ b/src/model/service/sol_task_test.go @@ -2,13 +2,59 @@ package service import ( "encoding/json" - "fmt" + "os" "testing" + "github.com/assimon/luuu/internal/testutil" + "github.com/assimon/luuu/model/dao" + "github.com/assimon/luuu/model/mdb" "github.com/tidwall/gjson" ) +func setupSolanaRPCNode(t *testing.T, url string) func() { + t.Helper() + + cleanup := testutil.SetupTestDatabases(t) + node := &mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: url, + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusOk, + } + if err := dao.Mdb.Create(node).Error; err != nil { + cleanup() + t.Fatalf("seed solana rpc_node: %v", err) + } + return cleanup +} + +func TestResolveSolanaRpcURLRequiresRpcNode(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if got, err := resolveSolanaRpcURL(); err == nil { + t.Fatalf("resolveSolanaRpcURL() = %q, nil; want error", got) + } +} + +func TestResolveSolanaRpcURLWithRow(t *testing.T) { + cleanup := setupSolanaRPCNode(t, " https://solana.example.com ") + defer cleanup() + + got, err := resolveSolanaRpcURL() + if err != nil { + t.Fatalf("resolveSolanaRpcURL(): %v", err) + } + if got != "https://solana.example.com" { + t.Fatalf("resolveSolanaRpcURL() = %q, want https://solana.example.com", got) + } +} + func TestSolClientHealthy(t *testing.T) { + requireSolanaIntegration(t) + bodyData, err := SolRetryClient("getHealth", nil) if err != nil { t.Fatalf("SolRetryClient failed: %v", err) @@ -33,6 +79,8 @@ func TestSolClientHealthy(t *testing.T) { } func TestSolClientGetSignaturesForAddress(t *testing.T) { + requireSolanaIntegration(t) + // Example wallet address (replace with actual test address) address := "2uFTf9TZ8gd7Kg6hkb79TxfaeNpaAgpJ8uVHguv2Yweu" @@ -57,6 +105,8 @@ func TestSolClientGetSignaturesForAddress(t *testing.T) { } func TestSolClientGetTransaction(t *testing.T) { + requireSolanaIntegration(t) + // Example transaction signature (replace with actual test signature) sig := "2aEoNykk4ZJ27C3y7EDJiQUc7GFnnsMe7ofFzB73swGL8kTxSBFCnwzWw3jzr3BND7k8hx15fZHUUAbG1XemNFe5" @@ -64,7 +114,6 @@ func TestSolClientGetTransaction(t *testing.T) { if err != nil { t.Fatalf("SolRetryClient failed: %v", err) } - fmt.Printf("%v\n", string(txData)) var result map[string]interface{} err = json.Unmarshal(txData, &result) @@ -80,6 +129,15 @@ func TestSolClientGetTransaction(t *testing.T) { t.Logf("Transaction Info for signature %s: %v", sig, txInfo) } +func requireSolanaIntegration(t *testing.T) { + t.Helper() + if testing.Short() || os.Getenv("RUN_SOLANA_INTEGRATION_TESTS") == "" { + t.Skip("set RUN_SOLANA_INTEGRATION_TESTS=1 to run Solana public RPC integration tests") + } + cleanup := setupSolanaRPCNode(t, "https://api.mainnet-beta.solana.com") + t.Cleanup(cleanup) +} + func TestFindATAAddress(t *testing.T) { tests := []struct { name string @@ -219,9 +277,7 @@ func TestAdjustAmount(t *testing.T) { } func TestParseTransferInfoFromInstruction_SplTransfer(t *testing.T) { - if testing.Short() { - t.Skip("Skipping integration test") - } + requireSolanaIntegration(t) // SPL Token "transfer" (no mint in instruction, must look up from postTokenBalances) sig := "3tZTwLrvmiZ59h4UzyMHPd7DPux7t9eXZgkUvEfquaoSuERrPSRNzWuSHKQM2fbiCWFDGNqoLpu2kLZnfoegVpqN" @@ -261,9 +317,7 @@ func TestParseTransferInfoFromInstruction_SplTransfer(t *testing.T) { } func TestParseTransferInfoFromInstruction_TransferChecked(t *testing.T) { - if testing.Short() { - t.Skip("Skipping integration test") - } + requireSolanaIntegration(t) // SPL Token "transferChecked" (has mint and tokenAmount in instruction) sig := "2aEoNykk4ZJ27C3y7EDJiQUc7GFnnsMe7ofFzB73swGL8kTxSBFCnwzWw3jzr3BND7k8hx15fZHUUAbG1XemNFe5" @@ -306,9 +360,7 @@ func TestParseTransferInfoFromInstruction_TransferChecked(t *testing.T) { } func TestParseTransferInfoFromInstruction_SystemTransfer(t *testing.T) { - if testing.Short() { - t.Skip("Skipping integration test") - } + requireSolanaIntegration(t) // System program SOL transfer sig := "5pNMonUBvLVpxXTmyd5CGVBs49W6781g2ACnrCXhbmtz58KENYA7HSqu6hQkQweg3qQboRd8WAscphNAtiq9UtZZ" diff --git a/src/model/service/task_service.go b/src/model/service/task_service.go index 71eeb85..9fbda51 100644 --- a/src/model/service/task_service.go +++ b/src/model/service/task_service.go @@ -9,7 +9,6 @@ import ( "strings" "sync" - "github.com/assimon/luuu/config" tron "github.com/assimon/luuu/crypto" "github.com/assimon/luuu/model/data" "github.com/assimon/luuu/model/mdb" @@ -26,19 +25,21 @@ import ( "github.com/tidwall/gjson" ) -// tronNodeDefault is the fallback when no rpc_nodes row exists for TRON. -const tronNodeDefault = "https://api.trongrid.io" - // resolveTronNode returns (baseURL, apiKey) for the TRON HTTP RPC node. -// It reads the first healthy (or any enabled) row from the rpc_nodes table; -// if none exists it falls back to the hard-coded default URL and the -// tron_grid_api_key value from .env (config.TRON_GRID_API_KEY). -func resolveTronNode() (string, string) { +// It reads the first healthy (or any enabled) row from the rpc_nodes table. +func resolveTronNode() (string, string, error) { node, err := data.SelectRpcNode(mdb.NetworkTron, mdb.RpcNodeTypeHttp) - if err == nil && node != nil && node.ID > 0 { - return strings.TrimRight(node.Url, "/"), node.ApiKey + if err != nil { + return "", "", err } - return tronNodeDefault, config.TRON_GRID_API_KEY + if node == nil || node.ID == 0 { + return "", "", fmt.Errorf("no enabled %s %s RPC node configured in rpc_nodes", mdb.NetworkTron, mdb.RpcNodeTypeHttp) + } + rpcURL := strings.TrimRight(strings.TrimSpace(node.Url), "/") + if rpcURL == "" { + return "", "", fmt.Errorf("rpc_nodes id=%d has empty url", node.ID) + } + return rpcURL, node.ApiKey, nil } func Trc20CallBack(address string, wg *sync.WaitGroup) { @@ -79,7 +80,11 @@ func checkTrxTransfers(address string, wg *sync.WaitGroup) { client := http_client.GetHttpClient() startTime := carbon.Now().AddHours(-24).TimestampMilli() endTime := carbon.Now().TimestampMilli() - tronBaseURL, tronAPIKey := resolveTronNode() + tronBaseURL, tronAPIKey, err := resolveTronNode() + if err != nil { + log.Sugar.Errorf("[TRX][%s] resolve rpc_nodes err=%v", address, err) + return + } url := fmt.Sprintf("%s/v1/accounts/%s/transactions", tronBaseURL, address) resp, err := client.R().SetQueryParams(map[string]string{ @@ -217,7 +222,11 @@ func checkTrc20Transfers(address string, wg *sync.WaitGroup) { client := http_client.GetHttpClient() startTime := carbon.Now().AddHours(-24).TimestampMilli() endTime := carbon.Now().TimestampMilli() - tronBaseURL, tronAPIKey := resolveTronNode() + tronBaseURL, tronAPIKey, err := resolveTronNode() + if err != nil { + log.Sugar.Errorf("[TRC20][%s] resolve rpc_nodes err=%v", address, err) + return + } url := fmt.Sprintf("%s/v1/accounts/%s/transactions/trc20", tronBaseURL, address) resp, err := client.R().SetQueryParams(map[string]string{ diff --git a/src/model/service/task_tron_test.go b/src/model/service/task_tron_test.go index b526b58..ff39291 100644 --- a/src/model/service/task_tron_test.go +++ b/src/model/service/task_tron_test.go @@ -3,26 +3,19 @@ package service import ( "testing" - "github.com/assimon/luuu/config" "github.com/assimon/luuu/internal/testutil" "github.com/assimon/luuu/model/dao" "github.com/assimon/luuu/model/mdb" ) -// TestResolveTronNode_NoRow verifies that resolveTronNode falls back to the -// hard-coded default when the rpc_nodes table has no TRON row. +// TestResolveTronNode_NoRow verifies that resolveTronNode requires an enabled +// TRON row from rpc_nodes. func TestResolveTronNode_NoRow(t *testing.T) { cleanup := testutil.SetupTestDatabases(t) defer cleanup() - config.TRON_GRID_API_KEY = "fallback-key" - - gotURL, gotKey := resolveTronNode() - if gotURL != tronNodeDefault { - t.Errorf("url = %q, want %q", gotURL, tronNodeDefault) - } - if gotKey != "fallback-key" { - t.Errorf("apiKey = %q, want \"fallback-key\"", gotKey) + if gotURL, gotKey, err := resolveTronNode(); err == nil { + t.Fatalf("resolveTronNode() = (%q, %q, nil), want error", gotURL, gotKey) } } @@ -32,8 +25,6 @@ func TestResolveTronNode_WithRow(t *testing.T) { cleanup := testutil.SetupTestDatabases(t) defer cleanup() - config.TRON_GRID_API_KEY = "fallback-key" - // Insert an enabled TRON http node. node := &mdb.RpcNode{ Network: mdb.NetworkTron, @@ -48,7 +39,10 @@ func TestResolveTronNode_WithRow(t *testing.T) { t.Fatalf("seed rpc_node: %v", err) } - gotURL, gotKey := resolveTronNode() + gotURL, gotKey, err := resolveTronNode() + if err != nil { + t.Fatalf("resolveTronNode(): %v", err) + } if gotURL != "https://custom-tron.example.com" { t.Errorf("url = %q, want https://custom-tron.example.com", gotURL) } @@ -57,14 +51,11 @@ func TestResolveTronNode_WithRow(t *testing.T) { } } -// TestResolveTronNode_DisabledRow verifies that a disabled row is ignored and -// the fallback is used. +// TestResolveTronNode_DisabledRow verifies that a disabled row is ignored. func TestResolveTronNode_DisabledRow(t *testing.T) { cleanup := testutil.SetupTestDatabases(t) defer cleanup() - config.TRON_GRID_API_KEY = "fallback-key" - node := &mdb.RpcNode{ Network: mdb.NetworkTron, Url: "https://disabled-tron.example.com", @@ -82,11 +73,7 @@ func TestResolveTronNode_DisabledRow(t *testing.T) { t.Fatalf("disable rpc_node: %v", err) } - gotURL, gotKey := resolveTronNode() - if gotURL != tronNodeDefault { - t.Errorf("url = %q, want %q (disabled row should be ignored)", gotURL, tronNodeDefault) - } - if gotKey != "fallback-key" { - t.Errorf("apiKey = %q, want \"fallback-key\"", gotKey) + if gotURL, gotKey, err := resolveTronNode(); err == nil { + t.Fatalf("resolveTronNode() = (%q, %q, nil), want error", gotURL, gotKey) } } diff --git a/src/task/listen_bsc.go b/src/task/listen_bsc.go index 85aac3f..1589638 100644 --- a/src/task/listen_bsc.go +++ b/src/task/listen_bsc.go @@ -18,8 +18,6 @@ import ( "github.com/ethereum/go-ethereum/ethclient" ) -const bscDefaultWsURL = "wss://bsc.drpc.org" - type bscRecipientSnapshot struct { addrs map[string]struct{} } @@ -68,7 +66,10 @@ func runBscListener(contracts []common.Address) { } }() - wsURL := resolveChainWsURL(mdb.NetworkBsc, bscDefaultWsURL) + wsURL, ok := resolveChainWsURL(mdb.NetworkBsc, "[BSC-WS]") + if !ok { + return + } log.Sugar.Infof("[BSC-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts)) query := ethereum.FilterQuery{ diff --git a/src/task/listen_chain_common.go b/src/task/listen_chain_common.go index d8d3050..f7032b5 100644 --- a/src/task/listen_chain_common.go +++ b/src/task/listen_chain_common.go @@ -90,12 +90,22 @@ func loadChainTokenContracts(network, logPrefix string) []common.Address { } // resolveChainWsURL picks a healthy WS endpoint from rpc_nodes for the -// given network, falling back to the provided default (usually the -// public node URL) when no row is configured. -func resolveChainWsURL(network, fallback string) string { +// given network. If no enabled node is configured, the caller skips the +// current listener run so admin-side disabled/deleted rows are respected. +func resolveChainWsURL(network, logPrefix string) (string, bool) { node, err := data.SelectRpcNode(network, mdb.RpcNodeTypeWs) if err == nil && node != nil && node.ID > 0 { - return node.Url + rpcURL := strings.TrimSpace(node.Url) + if rpcURL != "" { + return rpcURL, true + } + log.Sugar.Errorf("%s rpc_nodes id=%d has empty url", logPrefix, node.ID) + return "", false } - return fallback + if err != nil { + log.Sugar.Errorf("%s resolve rpc_nodes err=%v", logPrefix, err) + } else { + log.Sugar.Warnf("%s no enabled %s WS RPC node configured in rpc_nodes", logPrefix, network) + } + return "", false } diff --git a/src/task/listen_chain_common_test.go b/src/task/listen_chain_common_test.go new file mode 100644 index 0000000..50071c7 --- /dev/null +++ b/src/task/listen_chain_common_test.go @@ -0,0 +1,67 @@ +package task + +import ( + "testing" + + "github.com/assimon/luuu/internal/testutil" + "github.com/assimon/luuu/model/dao" + "github.com/assimon/luuu/model/mdb" +) + +func TestResolveChainWsURLRequiresEnabledRpcNode(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if got, ok := resolveChainWsURL(mdb.NetworkEthereum, "[TEST]"); ok { + t.Fatalf("resolveChainWsURL() = (%q, true), want false", got) + } +} + +func TestResolveChainWsURLWithRow(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + node := &mdb.RpcNode{ + Network: mdb.NetworkEthereum, + Url: " wss://ethereum.example.com ", + Type: mdb.RpcNodeTypeWs, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusOk, + } + if err := dao.Mdb.Create(node).Error; err != nil { + t.Fatalf("seed rpc_node: %v", err) + } + + got, ok := resolveChainWsURL(mdb.NetworkEthereum, "[TEST]") + if !ok { + t.Fatalf("resolveChainWsURL() ok=false, want true") + } + if got != "wss://ethereum.example.com" { + t.Fatalf("resolveChainWsURL() = %q, want wss://ethereum.example.com", got) + } +} + +func TestResolveChainWsURLDisabledRow(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + node := &mdb.RpcNode{ + Network: mdb.NetworkEthereum, + Url: "wss://disabled.example.com", + Type: mdb.RpcNodeTypeWs, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusOk, + } + if err := dao.Mdb.Create(node).Error; err != nil { + t.Fatalf("seed rpc_node: %v", err) + } + if err := dao.Mdb.Model(node).Update("enabled", false).Error; err != nil { + t.Fatalf("disable rpc_node: %v", err) + } + + if got, ok := resolveChainWsURL(mdb.NetworkEthereum, "[TEST]"); ok { + t.Fatalf("resolveChainWsURL() = (%q, true), want false", got) + } +} diff --git a/src/task/listen_eth.go b/src/task/listen_eth.go index d89bd90..969103b 100644 --- a/src/task/listen_eth.go +++ b/src/task/listen_eth.go @@ -7,7 +7,6 @@ import ( "sync/atomic" "time" - "github.com/assimon/luuu/config" "github.com/assimon/luuu/model/data" "github.com/assimon/luuu/model/mdb" "github.com/assimon/luuu/model/service" @@ -22,22 +21,6 @@ import ( // Transfer 事件签名 — ERC-20 signature, same on every EVM chain. var transferEventHash = common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") -const ethereumDefaultWsURL = "wss://ethereum.publicnode.com" - -// resolveEthereumWsURL picks a healthy WS endpoint. Falls back to the -// legacy .env value (if set) before the shared default — keeps existing -// deployments working until the admin adds an rpc_nodes row. -func resolveEthereumWsURL() string { - node, err := data.SelectRpcNode(mdb.NetworkEthereum, mdb.RpcNodeTypeWs) - if err == nil && node != nil && node.ID > 0 { - return node.Url - } - if u := config.GetEthereumWsUrl(); u != "" { - return u - } - return ethereumDefaultWsURL -} - type ethRecipientSnapshot struct { addrs map[string]struct{} } @@ -88,7 +71,10 @@ func runEthereumListener(contracts []common.Address) { } }() - wsURL := resolveEthereumWsURL() + wsURL, ok := resolveChainWsURL(mdb.NetworkEthereum, "[ETH-WS]") + if !ok { + return + } log.Sugar.Infof("[ETH-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts)) query := ethereum.FilterQuery{ @@ -145,4 +131,3 @@ func isWatchedEthRecipient(to common.Address) bool { _, ok := snap.addrs[strings.ToLower(to.Hex())] return ok } - diff --git a/src/task/listen_plasma.go b/src/task/listen_plasma.go index a7a63b0..7c0c71b 100644 --- a/src/task/listen_plasma.go +++ b/src/task/listen_plasma.go @@ -18,8 +18,6 @@ import ( "github.com/ethereum/go-ethereum/ethclient" ) -const plasmaDefaultWsURL = "wss://rpc.plasma.to" - type plasmaRecipientSnapshot struct { addrs map[string]struct{} } @@ -67,7 +65,10 @@ func runPlasmaListener(contracts []common.Address) { } }() - wsURL := resolveChainWsURL(mdb.NetworkPlasma, plasmaDefaultWsURL) + wsURL, ok := resolveChainWsURL(mdb.NetworkPlasma, "[PLASMA-WS]") + if !ok { + return + } log.Sugar.Infof("[PLASMA-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts)) query := ethereum.FilterQuery{ diff --git a/src/task/listen_polygon.go b/src/task/listen_polygon.go index e416dc4..b165bd2 100644 --- a/src/task/listen_polygon.go +++ b/src/task/listen_polygon.go @@ -18,8 +18,6 @@ import ( "github.com/ethereum/go-ethereum/ethclient" ) -const polygonDefaultWsURL = "wss://polygon-bor-rpc.publicnode.com" - type polygonRecipientSnapshot struct { addrs map[string]struct{} } @@ -67,7 +65,10 @@ func runPolygonListener(contracts []common.Address) { } }() - wsURL := resolveChainWsURL(mdb.NetworkPolygon, polygonDefaultWsURL) + wsURL, ok := resolveChainWsURL(mdb.NetworkPolygon, "[POLYGON-WS]") + if !ok { + return + } log.Sugar.Infof("[POLYGON-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts)) query := ethereum.FilterQuery{