fix: prevent SQLite lock churn and validate EVM RPC capabilities

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)
 	}

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
 	}

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)
+		}
+	}
+}

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},

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
 	}

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++ {

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()

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 {

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.

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()
 

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 {

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 {

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 {