refactor(rpc): streamline RPC node selection and remove hardcoded defaults

This commit is contained in:
line-6000
2026-04-23 02:08:18 +08:00
parent 09514a6946
commit 91936d580f
13 changed files with 326 additions and 101 deletions
-14
View File
@@ -36,7 +36,6 @@ var (
TgManage int64 TgManage int64
UsdtRate float64 UsdtRate float64
RateApiUrl string RateApiUrl string
TRON_GRID_API_KEY string
BuildVersion = "0.0.0-dev" BuildVersion = "0.0.0-dev"
BuildCommit = "none" BuildCommit = "none"
BuildDate = "unknown" BuildDate = "unknown"
@@ -79,7 +78,6 @@ func Init() {
TgManage = viper.GetInt64("tg_manage") TgManage = viper.GetInt64("tg_manage")
RateApiUrl = GetRateApiUrl() RateApiUrl = GetRateApiUrl()
TRON_GRID_API_KEY = viper.GetString("tron_grid_api_key")
} }
func mustMkdir(path string) { func mustMkdir(path string) {
@@ -398,15 +396,3 @@ func GetCallbackRetryBaseDuration() time.Duration {
} }
return time.Duration(seconds) * time.Second 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"))
}
+12 -5
View File
@@ -47,8 +47,8 @@ func DeleteRpcNodeByID(id uint64) error {
// SelectRpcNode picks a healthy RPC endpoint for a (network, type) pair. // SelectRpcNode picks a healthy RPC endpoint for a (network, type) pair.
// Strategy: weighted random among rows where enabled=true AND status=ok. // Strategy: weighted random among rows where enabled=true AND status=ok.
// Falls back to any enabled row when none are healthy yet (bootstrap case // Falls back to enabled rows with status=unknown when no health check has
// where the health check hasn't run). Returns nil if nothing is defined. // run yet. Explicitly down rows are not selected.
func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) { func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) {
var rows []mdb.RpcNode var rows []mdb.RpcNode
err := dao.Mdb.Model(&mdb.RpcNode{}). err := dao.Mdb.Model(&mdb.RpcNode{}).
@@ -62,15 +62,22 @@ func SelectRpcNode(network, nodeType string) (*mdb.RpcNode, error) {
if len(rows) == 0 { if len(rows) == 0 {
return nil, nil return nil, nil
} }
healthy := rows[:0] healthy := make([]mdb.RpcNode, 0, len(rows))
bootstrap := make([]mdb.RpcNode, 0, len(rows))
for _, r := range rows { for _, r := range rows {
if r.Status == mdb.RpcNodeStatusOk { switch r.Status {
case mdb.RpcNodeStatusOk:
healthy = append(healthy, r) healthy = append(healthy, r)
case "", mdb.RpcNodeStatusUnknown:
bootstrap = append(bootstrap, r)
} }
} }
candidates := healthy candidates := healthy
if len(candidates) == 0 { 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 return pickWeighted(candidates), nil
} }
+101
View File
@@ -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)
}
}
+19 -1
View File
@@ -281,6 +281,21 @@ type solSignatureResult struct {
BlockTime *int64 `json:"blockTime"` 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 请求,自动重试 // SolRetryClient 发送 Solana JSON-RPC 请求,自动重试
func SolRetryClient(method string, params []interface{}) ([]byte, error) { func SolRetryClient(method string, params []interface{}) ([]byte, error) {
client := resty.New() client := resty.New()
@@ -297,7 +312,10 @@ func SolRetryClient(method string, params []interface{}) ([]byte, error) {
return false return false
}) })
rpcUrl := config.GetSolanaRpcUrl() rpcUrl, err := resolveSolanaRpcURL()
if err != nil {
return nil, err
}
resp, err := client.R(). resp, err := client.R().
SetHeader("Content-Type", "application/json"). SetHeader("Content-Type", "application/json").
+63 -11
View File
@@ -2,13 +2,59 @@ package service
import ( import (
"encoding/json" "encoding/json"
"fmt" "os"
"testing" "testing"
"github.com/assimon/luuu/internal/testutil"
"github.com/assimon/luuu/model/dao"
"github.com/assimon/luuu/model/mdb"
"github.com/tidwall/gjson" "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) { func TestSolClientHealthy(t *testing.T) {
requireSolanaIntegration(t)
bodyData, err := SolRetryClient("getHealth", nil) bodyData, err := SolRetryClient("getHealth", nil)
if err != nil { if err != nil {
t.Fatalf("SolRetryClient failed: %v", err) t.Fatalf("SolRetryClient failed: %v", err)
@@ -33,6 +79,8 @@ func TestSolClientHealthy(t *testing.T) {
} }
func TestSolClientGetSignaturesForAddress(t *testing.T) { func TestSolClientGetSignaturesForAddress(t *testing.T) {
requireSolanaIntegration(t)
// Example wallet address (replace with actual test address) // Example wallet address (replace with actual test address)
address := "2uFTf9TZ8gd7Kg6hkb79TxfaeNpaAgpJ8uVHguv2Yweu" address := "2uFTf9TZ8gd7Kg6hkb79TxfaeNpaAgpJ8uVHguv2Yweu"
@@ -57,6 +105,8 @@ func TestSolClientGetSignaturesForAddress(t *testing.T) {
} }
func TestSolClientGetTransaction(t *testing.T) { func TestSolClientGetTransaction(t *testing.T) {
requireSolanaIntegration(t)
// Example transaction signature (replace with actual test signature) // Example transaction signature (replace with actual test signature)
sig := "2aEoNykk4ZJ27C3y7EDJiQUc7GFnnsMe7ofFzB73swGL8kTxSBFCnwzWw3jzr3BND7k8hx15fZHUUAbG1XemNFe5" sig := "2aEoNykk4ZJ27C3y7EDJiQUc7GFnnsMe7ofFzB73swGL8kTxSBFCnwzWw3jzr3BND7k8hx15fZHUUAbG1XemNFe5"
@@ -64,7 +114,6 @@ func TestSolClientGetTransaction(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("SolRetryClient failed: %v", err) t.Fatalf("SolRetryClient failed: %v", err)
} }
fmt.Printf("%v\n", string(txData))
var result map[string]interface{} var result map[string]interface{}
err = json.Unmarshal(txData, &result) err = json.Unmarshal(txData, &result)
@@ -80,6 +129,15 @@ func TestSolClientGetTransaction(t *testing.T) {
t.Logf("Transaction Info for signature %s: %v", sig, txInfo) 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) { func TestFindATAAddress(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
@@ -219,9 +277,7 @@ func TestAdjustAmount(t *testing.T) {
} }
func TestParseTransferInfoFromInstruction_SplTransfer(t *testing.T) { func TestParseTransferInfoFromInstruction_SplTransfer(t *testing.T) {
if testing.Short() { requireSolanaIntegration(t)
t.Skip("Skipping integration test")
}
// SPL Token "transfer" (no mint in instruction, must look up from postTokenBalances) // SPL Token "transfer" (no mint in instruction, must look up from postTokenBalances)
sig := "3tZTwLrvmiZ59h4UzyMHPd7DPux7t9eXZgkUvEfquaoSuERrPSRNzWuSHKQM2fbiCWFDGNqoLpu2kLZnfoegVpqN" sig := "3tZTwLrvmiZ59h4UzyMHPd7DPux7t9eXZgkUvEfquaoSuERrPSRNzWuSHKQM2fbiCWFDGNqoLpu2kLZnfoegVpqN"
@@ -261,9 +317,7 @@ func TestParseTransferInfoFromInstruction_SplTransfer(t *testing.T) {
} }
func TestParseTransferInfoFromInstruction_TransferChecked(t *testing.T) { func TestParseTransferInfoFromInstruction_TransferChecked(t *testing.T) {
if testing.Short() { requireSolanaIntegration(t)
t.Skip("Skipping integration test")
}
// SPL Token "transferChecked" (has mint and tokenAmount in instruction) // SPL Token "transferChecked" (has mint and tokenAmount in instruction)
sig := "2aEoNykk4ZJ27C3y7EDJiQUc7GFnnsMe7ofFzB73swGL8kTxSBFCnwzWw3jzr3BND7k8hx15fZHUUAbG1XemNFe5" sig := "2aEoNykk4ZJ27C3y7EDJiQUc7GFnnsMe7ofFzB73swGL8kTxSBFCnwzWw3jzr3BND7k8hx15fZHUUAbG1XemNFe5"
@@ -306,9 +360,7 @@ func TestParseTransferInfoFromInstruction_TransferChecked(t *testing.T) {
} }
func TestParseTransferInfoFromInstruction_SystemTransfer(t *testing.T) { func TestParseTransferInfoFromInstruction_SystemTransfer(t *testing.T) {
if testing.Short() { requireSolanaIntegration(t)
t.Skip("Skipping integration test")
}
// System program SOL transfer // System program SOL transfer
sig := "5pNMonUBvLVpxXTmyd5CGVBs49W6781g2ACnrCXhbmtz58KENYA7HSqu6hQkQweg3qQboRd8WAscphNAtiq9UtZZ" sig := "5pNMonUBvLVpxXTmyd5CGVBs49W6781g2ACnrCXhbmtz58KENYA7HSqu6hQkQweg3qQboRd8WAscphNAtiq9UtZZ"
+22 -13
View File
@@ -9,7 +9,6 @@ import (
"strings" "strings"
"sync" "sync"
"github.com/assimon/luuu/config"
tron "github.com/assimon/luuu/crypto" tron "github.com/assimon/luuu/crypto"
"github.com/assimon/luuu/model/data" "github.com/assimon/luuu/model/data"
"github.com/assimon/luuu/model/mdb" "github.com/assimon/luuu/model/mdb"
@@ -26,19 +25,21 @@ import (
"github.com/tidwall/gjson" "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. // resolveTronNode returns (baseURL, apiKey) for the TRON HTTP RPC node.
// It reads the first healthy (or any enabled) row from the rpc_nodes table; // 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 func resolveTronNode() (string, string, error) {
// tron_grid_api_key value from .env (config.TRON_GRID_API_KEY).
func resolveTronNode() (string, string) {
node, err := data.SelectRpcNode(mdb.NetworkTron, mdb.RpcNodeTypeHttp) node, err := data.SelectRpcNode(mdb.NetworkTron, mdb.RpcNodeTypeHttp)
if err == nil && node != nil && node.ID > 0 { if err != nil {
return strings.TrimRight(node.Url, "/"), node.ApiKey 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) { func Trc20CallBack(address string, wg *sync.WaitGroup) {
@@ -79,7 +80,11 @@ func checkTrxTransfers(address string, wg *sync.WaitGroup) {
client := http_client.GetHttpClient() client := http_client.GetHttpClient()
startTime := carbon.Now().AddHours(-24).TimestampMilli() startTime := carbon.Now().AddHours(-24).TimestampMilli()
endTime := carbon.Now().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) url := fmt.Sprintf("%s/v1/accounts/%s/transactions", tronBaseURL, address)
resp, err := client.R().SetQueryParams(map[string]string{ resp, err := client.R().SetQueryParams(map[string]string{
@@ -217,7 +222,11 @@ func checkTrc20Transfers(address string, wg *sync.WaitGroup) {
client := http_client.GetHttpClient() client := http_client.GetHttpClient()
startTime := carbon.Now().AddHours(-24).TimestampMilli() startTime := carbon.Now().AddHours(-24).TimestampMilli()
endTime := carbon.Now().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) url := fmt.Sprintf("%s/v1/accounts/%s/transactions/trc20", tronBaseURL, address)
resp, err := client.R().SetQueryParams(map[string]string{ resp, err := client.R().SetQueryParams(map[string]string{
+11 -24
View File
@@ -3,26 +3,19 @@ package service
import ( import (
"testing" "testing"
"github.com/assimon/luuu/config"
"github.com/assimon/luuu/internal/testutil" "github.com/assimon/luuu/internal/testutil"
"github.com/assimon/luuu/model/dao" "github.com/assimon/luuu/model/dao"
"github.com/assimon/luuu/model/mdb" "github.com/assimon/luuu/model/mdb"
) )
// TestResolveTronNode_NoRow verifies that resolveTronNode falls back to the // TestResolveTronNode_NoRow verifies that resolveTronNode requires an enabled
// hard-coded default when the rpc_nodes table has no TRON row. // TRON row from rpc_nodes.
func TestResolveTronNode_NoRow(t *testing.T) { func TestResolveTronNode_NoRow(t *testing.T) {
cleanup := testutil.SetupTestDatabases(t) cleanup := testutil.SetupTestDatabases(t)
defer cleanup() defer cleanup()
config.TRON_GRID_API_KEY = "fallback-key" if gotURL, gotKey, err := resolveTronNode(); err == nil {
t.Fatalf("resolveTronNode() = (%q, %q, nil), want error", gotURL, gotKey)
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)
} }
} }
@@ -32,8 +25,6 @@ func TestResolveTronNode_WithRow(t *testing.T) {
cleanup := testutil.SetupTestDatabases(t) cleanup := testutil.SetupTestDatabases(t)
defer cleanup() defer cleanup()
config.TRON_GRID_API_KEY = "fallback-key"
// Insert an enabled TRON http node. // Insert an enabled TRON http node.
node := &mdb.RpcNode{ node := &mdb.RpcNode{
Network: mdb.NetworkTron, Network: mdb.NetworkTron,
@@ -48,7 +39,10 @@ func TestResolveTronNode_WithRow(t *testing.T) {
t.Fatalf("seed rpc_node: %v", err) 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" { if gotURL != "https://custom-tron.example.com" {
t.Errorf("url = %q, want https://custom-tron.example.com", gotURL) 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 // TestResolveTronNode_DisabledRow verifies that a disabled row is ignored.
// the fallback is used.
func TestResolveTronNode_DisabledRow(t *testing.T) { func TestResolveTronNode_DisabledRow(t *testing.T) {
cleanup := testutil.SetupTestDatabases(t) cleanup := testutil.SetupTestDatabases(t)
defer cleanup() defer cleanup()
config.TRON_GRID_API_KEY = "fallback-key"
node := &mdb.RpcNode{ node := &mdb.RpcNode{
Network: mdb.NetworkTron, Network: mdb.NetworkTron,
Url: "https://disabled-tron.example.com", Url: "https://disabled-tron.example.com",
@@ -82,11 +73,7 @@ func TestResolveTronNode_DisabledRow(t *testing.T) {
t.Fatalf("disable rpc_node: %v", err) t.Fatalf("disable rpc_node: %v", err)
} }
gotURL, gotKey := resolveTronNode() if gotURL, gotKey, err := resolveTronNode(); err == nil {
if gotURL != tronNodeDefault { t.Fatalf("resolveTronNode() = (%q, %q, nil), want error", gotURL, gotKey)
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)
} }
} }
+4 -3
View File
@@ -18,8 +18,6 @@ import (
"github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/ethclient"
) )
const bscDefaultWsURL = "wss://bsc.drpc.org"
type bscRecipientSnapshot struct { type bscRecipientSnapshot struct {
addrs map[string]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)) log.Sugar.Infof("[BSC-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts))
query := ethereum.FilterQuery{ query := ethereum.FilterQuery{
+15 -5
View File
@@ -90,12 +90,22 @@ func loadChainTokenContracts(network, logPrefix string) []common.Address {
} }
// resolveChainWsURL picks a healthy WS endpoint from rpc_nodes for the // resolveChainWsURL picks a healthy WS endpoint from rpc_nodes for the
// given network, falling back to the provided default (usually the // given network. If no enabled node is configured, the caller skips the
// public node URL) when no row is configured. // current listener run so admin-side disabled/deleted rows are respected.
func resolveChainWsURL(network, fallback string) string { func resolveChainWsURL(network, logPrefix string) (string, bool) {
node, err := data.SelectRpcNode(network, mdb.RpcNodeTypeWs) node, err := data.SelectRpcNode(network, mdb.RpcNodeTypeWs)
if err == nil && node != nil && node.ID > 0 { 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
} }
+67
View File
@@ -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)
}
}
+4 -19
View File
@@ -7,7 +7,6 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/assimon/luuu/config"
"github.com/assimon/luuu/model/data" "github.com/assimon/luuu/model/data"
"github.com/assimon/luuu/model/mdb" "github.com/assimon/luuu/model/mdb"
"github.com/assimon/luuu/model/service" "github.com/assimon/luuu/model/service"
@@ -22,22 +21,6 @@ import (
// Transfer 事件签名 — ERC-20 signature, same on every EVM chain. // Transfer 事件签名 — ERC-20 signature, same on every EVM chain.
var transferEventHash = common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") 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 { type ethRecipientSnapshot struct {
addrs map[string]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)) log.Sugar.Infof("[ETH-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts))
query := ethereum.FilterQuery{ query := ethereum.FilterQuery{
@@ -145,4 +131,3 @@ func isWatchedEthRecipient(to common.Address) bool {
_, ok := snap.addrs[strings.ToLower(to.Hex())] _, ok := snap.addrs[strings.ToLower(to.Hex())]
return ok return ok
} }
+4 -3
View File
@@ -18,8 +18,6 @@ import (
"github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/ethclient"
) )
const plasmaDefaultWsURL = "wss://rpc.plasma.to"
type plasmaRecipientSnapshot struct { type plasmaRecipientSnapshot struct {
addrs map[string]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)) log.Sugar.Infof("[PLASMA-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts))
query := ethereum.FilterQuery{ query := ethereum.FilterQuery{
+4 -3
View File
@@ -18,8 +18,6 @@ import (
"github.com/ethereum/go-ethereum/ethclient" "github.com/ethereum/go-ethereum/ethclient"
) )
const polygonDefaultWsURL = "wss://polygon-bor-rpc.publicnode.com"
type polygonRecipientSnapshot struct { type polygonRecipientSnapshot struct {
addrs map[string]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)) log.Sugar.Infof("[POLYGON-WS] connecting to %s watching %d contract(s)", wsURL, len(contracts))
query := ethereum.FilterQuery{ query := ethereum.FilterQuery{