feat: add in-memory RPC runtime stats SSE endpoint

src/model/data/rpc_runtime_stats.go

@@ -0,0 +1,108 @@
+package data
+
+import (
+	"strings"
+	"sync"
+	"time"
+)
+
+// RpcRuntimeStatsSnapshot is an in-process snapshot of RPC activity for one
+// chain. It is intentionally not persisted; restarting the process resets it.
+type RpcRuntimeStatsSnapshot struct {
+	Network           string
+	SuccessCount      int64
+	FailureCount      int64
+	LatestBlockHeight int64
+	LastSyncAt        time.Time
+}
+
+type rpcRuntimeStatsCounter struct {
+	successCount      int64
+	failureCount      int64
+	latestBlockHeight int64
+	lastSyncAt        time.Time
+}
+
+var gRpcRuntimeStats = struct {
+	sync.RWMutex
+	byNetwork map[string]*rpcRuntimeStatsCounter
+}{
+	byNetwork: make(map[string]*rpcRuntimeStatsCounter),
+}
+
+func normalizeRpcStatsNetwork(network string) string {
+	return strings.ToLower(strings.TrimSpace(network))
+}
+
+func rpcStatsCounterLocked(network string) *rpcRuntimeStatsCounter {
+	counter := gRpcRuntimeStats.byNetwork[network]
+	if counter == nil {
+		counter = &rpcRuntimeStatsCounter{}
+		gRpcRuntimeStats.byNetwork[network] = counter
+	}
+	return counter
+}
+
+// RecordRpcSuccess records one successful application-level RPC operation.
+func RecordRpcSuccess(network string) {
+	network = normalizeRpcStatsNetwork(network)
+	if network == "" {
+		return
+	}
+	gRpcRuntimeStats.Lock()
+	defer gRpcRuntimeStats.Unlock()
+	counter := rpcStatsCounterLocked(network)
+	counter.successCount++
+	counter.lastSyncAt = time.Now()
+}
+
+// RecordRpcFailure records one failed application-level RPC operation.
+func RecordRpcFailure(network string) {
+	network = normalizeRpcStatsNetwork(network)
+	if network == "" {
+		return
+	}
+	gRpcRuntimeStats.Lock()
+	defer gRpcRuntimeStats.Unlock()
+	counter := rpcStatsCounterLocked(network)
+	counter.failureCount++
+}
+
+// RecordRpcBlockHeight records the latest observed chain height.
+func RecordRpcBlockHeight(network string, height int64) {
+	network = normalizeRpcStatsNetwork(network)
+	if network == "" || height <= 0 {
+		return
+	}
+	gRpcRuntimeStats.Lock()
+	defer gRpcRuntimeStats.Unlock()
+	counter := rpcStatsCounterLocked(network)
+	counter.lastSyncAt = time.Now()
+	if height > counter.latestBlockHeight {
+		counter.latestBlockHeight = height
+	}
+}
+
+// SnapshotRpcRuntimeStats returns a copy of the current per-chain counters.
+func SnapshotRpcRuntimeStats() map[string]RpcRuntimeStatsSnapshot {
+	gRpcRuntimeStats.RLock()
+	defer gRpcRuntimeStats.RUnlock()
+	out := make(map[string]RpcRuntimeStatsSnapshot, len(gRpcRuntimeStats.byNetwork))
+	for network, counter := range gRpcRuntimeStats.byNetwork {
+		out[network] = RpcRuntimeStatsSnapshot{
+			Network:           network,
+			SuccessCount:      counter.successCount,
+			FailureCount:      counter.failureCount,
+			LatestBlockHeight: counter.latestBlockHeight,
+			LastSyncAt:        counter.lastSyncAt,
+		}
+	}
+	return out
+}
+
+// ResetRpcRuntimeStatsForTest clears process-local stats for isolated tests.
+func ResetRpcRuntimeStatsForTest() {
+	gRpcRuntimeStats.Lock()
+	defer gRpcRuntimeStats.Unlock()
+	gRpcRuntimeStats.byNetwork = make(map[string]*rpcRuntimeStatsCounter)
+}

src/model/data/rpc_runtime_stats_test.go

@@ -0,0 +1,81 @@
+package data
+
+import (
+	"testing"
+	"time"
+)
+
+func TestRpcRuntimeStatsRecordsCountsAndLatestHeight(t *testing.T) {
+	ResetRpcRuntimeStatsForTest()
+	t.Cleanup(ResetRpcRuntimeStatsForTest)
+
+	start := time.Now()
+	RecordRpcSuccess(" TRON ")
+	RecordRpcSuccess("tron")
+	RecordRpcFailure("TRON")
+	RecordRpcBlockHeight("Tron", 100)
+	RecordRpcBlockHeight("tron", 90)
+
+	stats := SnapshotRpcRuntimeStats()
+	got, ok := stats["tron"]
+	if !ok {
+		t.Fatalf("missing tron stats: %#v", stats)
+	}
+	if got.Network != "tron" {
+		t.Fatalf("network = %q, want tron", got.Network)
+	}
+	if got.SuccessCount != 2 {
+		t.Fatalf("success count = %d, want 2", got.SuccessCount)
+	}
+	if got.FailureCount != 1 {
+		t.Fatalf("failure count = %d, want 1", got.FailureCount)
+	}
+	if got.LatestBlockHeight != 100 {
+		t.Fatalf("latest block height = %d, want 100", got.LatestBlockHeight)
+	}
+	if got.LastSyncAt.Before(start) {
+		t.Fatalf("last sync at = %s, want after %s", got.LastSyncAt, start)
+	}
+}
+
+func TestRpcRuntimeStatsIgnoresEmptyNetworkAndInvalidHeight(t *testing.T) {
+	ResetRpcRuntimeStatsForTest()
+	t.Cleanup(ResetRpcRuntimeStatsForTest)
+
+	RecordRpcSuccess("")
+	RecordRpcFailure(" ")
+	RecordRpcBlockHeight("tron", 0)
+
+	stats := SnapshotRpcRuntimeStats()
+	if len(stats) != 0 {
+		t.Fatalf("stats = %#v, want empty", stats)
+	}
+}
+
+func TestRpcRuntimeStatsBlockHeightUpdatesLastSyncAt(t *testing.T) {
+	ResetRpcRuntimeStatsForTest()
+	t.Cleanup(ResetRpcRuntimeStatsForTest)
+
+	start := time.Now()
+	RecordRpcBlockHeight("ethereum", 123)
+
+	stats := SnapshotRpcRuntimeStats()
+	got := stats["ethereum"]
+	if got.SuccessCount != 0 {
+		t.Fatalf("success count = %d, want 0", got.SuccessCount)
+	}
+	if got.LastSyncAt.Before(start) {
+		t.Fatalf("last sync at = %s, want after %s", got.LastSyncAt, start)
+	}
+}
+
+func TestResetRpcRuntimeStatsForTest(t *testing.T) {
+	ResetRpcRuntimeStatsForTest()
+	RecordRpcSuccess("tron")
+	ResetRpcRuntimeStatsForTest()
+
+	stats := SnapshotRpcRuntimeStats()
+	if len(stats) != 0 {
+		t.Fatalf("stats = %#v, want empty after reset", stats)
+	}
+}

src/model/service/sol_task.go

@@ -325,7 +325,15 @@ func resolveSolanaRpcNode(excludeIDs ...uint64) (*mdb.RpcNode, error) {
 }
 
 // SolRetryClient 发送 Solana JSON-RPC 请求，自动重试
-func SolRetryClient(method string, params []interface{}) ([]byte, error) {
+func SolRetryClient(method string, params []interface{}) (body []byte, err error) {
+	defer func() {
+		if err != nil {
+			data.RecordRpcFailure(mdb.NetworkSolana)
+			return
+		}
+		data.RecordRpcSuccess(mdb.NetworkSolana)
+	}()
+
 	tried := make([]uint64, 0, 3)
 	var lastErr error
 	for attempts := 0; attempts < 3; attempts++ {
@@ -443,6 +451,32 @@ func SolGetSignaturesForAddress(address string, limit int, untilSig string, befo
 	return bodyData, nil
 }
 
+func SolGetBlockHeight() (height int64, err error) {
+	defer func() {
+		if err != nil {
+			data.RecordRpcFailure(mdb.NetworkSolana)
+			return
+		}
+		data.RecordRpcSuccess(mdb.NetworkSolana)
+	}()
+
+	rpcURL, err := resolveSolanaRpcURL()
+	if err != nil {
+		return 0, err
+	}
+	bodyData, err := solRetryClientWithURL(rpcURL, "getBlockHeight", []interface{}{
+		map[string]interface{}{"commitment": "finalized"},
+	})
+	if err != nil {
+		return 0, err
+	}
+	height = gjson.GetBytes(bodyData, "result").Int()
+	if height <= 0 {
+		return 0, fmt.Errorf("unexpected getBlockHeight response: %s", string(bodyData))
+	}
+	return height, nil
+}
+
 func SolGetTransaction(sig string) ([]byte, error) {
 	txData, err := SolRetryClient("getTransaction", []interface{}{
 		sig,

src/model/service/sol_task_test.go

@@ -102,6 +102,53 @@ func TestResolveSolanaRpcURLUsesGeneralWhenManualVerifyExists(t *testing.T) {
 	}
 }
 
+func TestSolGetBlockHeightDoesNotAffectRpcNodeFailover(t *testing.T) {
+	cleanup := testutil.SetupTestDatabases(t)
+	defer cleanup()
+	data.ResetRpcFailoverForTest()
+	data.ResetRpcRuntimeStatsForTest()
+	t.Cleanup(data.ResetRpcFailoverForTest)
+	t.Cleanup(data.ResetRpcRuntimeStatsForTest)
+
+	oldRetryCount := solRPCRetryCount
+	solRPCRetryCount = 0
+	t.Cleanup(func() {
+		solRPCRetryCount = oldRetryCount
+	})
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "temporary", http.StatusBadGateway)
+	}))
+	defer server.Close()
+
+	node := &mdb.RpcNode{
+		Network: mdb.NetworkSolana,
+		Url:     server.URL,
+		Type:    mdb.RpcNodeTypeHttp,
+		Weight:  1,
+		Enabled: true,
+		Purpose: mdb.RpcNodePurposeGeneral,
+		Status:  mdb.RpcNodeStatusOk,
+	}
+	if err := dao.Mdb.Create(node).Error; err != nil {
+		t.Fatalf("seed solana rpc_node: %v", err)
+	}
+
+	for i := 0; i < data.RpcFailoverThreshold+1; i++ {
+		if _, err := SolGetBlockHeight(); err == nil {
+			t.Fatalf("SolGetBlockHeight attempt %d unexpectedly succeeded", i+1)
+		}
+	}
+
+	if data.IsRpcNodeCoolingDown(node.ID) {
+		t.Fatal("SolGetBlockHeight failures should not affect rpc_node failover cooldown")
+	}
+	stats := data.SnapshotRpcRuntimeStats()
+	if got := stats[mdb.NetworkSolana].FailureCount; got != int64(data.RpcFailoverThreshold+1) {
+		t.Fatalf("solana runtime failure_count = %d, want %d", got, data.RpcFailoverThreshold+1)
+	}
+}
+
 func TestSolRetryClientSwitchesAfterFailureThreshold(t *testing.T) {
 	cleanup := testutil.SetupTestDatabases(t)
 	defer cleanup()