From dec1b97b9e4b49adf0d50419ba145019b64fd58e Mon Sep 17 00:00:00 2001 From: line-6000 <154492442+line-6000@users.noreply.github.com> Date: Fri, 22 May 2026 16:37:50 +0800 Subject: [PATCH] fix: retry Solana signatures after transient processing errors Avoid marking a Solana signature as processed when order matching or OrderProcessing fails with a retryable error. Keep terminal outcomes idempotent while allowing later scans to retry transient failures. --- src/model/service/sol_task.go | 13 ++- src/model/service/sol_task_test.go | 147 +++++++++++++++++++++++++++++ 2 files changed, 159 insertions(+), 1 deletion(-) diff --git a/src/model/service/sol_task.go b/src/model/service/sol_task.go index c0c709d..5f86117 100644 --- a/src/model/service/sol_task.go +++ b/src/model/service/sol_task.go @@ -26,6 +26,8 @@ import ( // gProcessedSignatures 已处理签名缓存，避免重复调用 getTransaction var gProcessedSignatures sync.Map // sig -> unix timestamp +var processSolanaOrder = OrderProcessing + type TransferInfo struct { Source string // Source address (for SOL) or source ATA (for SPL tokens) Destination string // Destination address (for SOL) or destination ATA (for SPL tokens) @@ -150,6 +152,7 @@ func SolCallBack(address string, wg *sync.WaitGroup) { // Process each transaction signature for sigIdx, txSig := range result { sig := txSig.Signature + retrySignature := false // Skip failed transactions if txSig.Err != nil { @@ -213,6 +216,7 @@ func SolCallBack(address string, wg *sync.WaitGroup) { tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkSolana, address, token, amount) if err != nil { log.Sugar.Errorf("[SOL][%s] sig=%s query transaction_lock failed: %v", address, sig, err) + retrySignature = true continue } if tradeID == "" { @@ -226,6 +230,7 @@ func SolCallBack(address string, wg *sync.WaitGroup) { order, err := data.GetOrderInfoByTradeId(tradeID) if err != nil { log.Sugar.Errorf("[SOL][%s] sig=%s load order failed for trade_id=%s: %v", address, sig, tradeID, err) + retrySignature = true continue } log.Sugar.Infof("[SOL][%s] order loaded: trade_id=%s order_id=%s status=%d created_at_ms=%d", @@ -252,13 +257,14 @@ func SolCallBack(address string, wg *sync.WaitGroup) { } log.Sugar.Infof("[SOL][%s] calling OrderProcessing: trade_id=%s sig=%s token=%s amount=%.2f", address, tradeID, sig, token, amount) - err = OrderProcessing(req) + err = processSolanaOrder(req) if err != nil { if errors.Is(err, constant.OrderBlockAlreadyProcess) || errors.Is(err, constant.OrderStatusConflict) { log.Sugar.Infof("[SOL][%s] sig=%s already resolved: trade_id=%s reason=%v", address, sig, tradeID, err) continue } log.Sugar.Errorf("[SOL][%s] sig=%s OrderProcessing failed for trade_id=%s: %v", address, sig, tradeID, err) + retrySignature = true continue } @@ -268,6 +274,11 @@ func SolCallBack(address string, wg *sync.WaitGroup) { log.Sugar.Infof("[SOL][%s] payment fully processed: trade_id=%s sig=%s", address, tradeID, sig) } + if retrySignature { + log.Sugar.Debugf("[SOL][%s] sig=%s not marked processed because retryable processing failed", address, sig) + continue + } + // 标记已处理 gProcessedSignatures.Store(sig, time.Now().Unix()) } diff --git a/src/model/service/sol_task_test.go b/src/model/service/sol_task_test.go index 504c627..52ebddf 100644 --- a/src/model/service/sol_task_test.go +++ b/src/model/service/sol_task_test.go @@ -2,12 +2,19 @@ package service import ( "encoding/json" + "errors" + "net/http" + "net/http/httptest" "os" + "sync" "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/GMWalletApp/epusdt/model/request" "github.com/tidwall/gjson" ) @@ -52,6 +59,139 @@ func TestResolveSolanaRpcURLWithRow(t *testing.T) { } } +func TestSolCallBackDoesNotCacheSignatureAfterRetryableOrderProcessingError(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + clearProcessedSignatures() + t.Cleanup(clearProcessedSignatures) + + const ( + address = "2uFTf9TZ8gd7Kg6hkb79TxfaeNpaAgpJ8uVHguv2Yweu" + sig = "retryable-sol-signature" + tradeID = "sol-retryable-order-001" + amount = 1.23 + ) + blockTime := time.Now().Add(time.Minute).Unix() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var rpcReq struct { + Method string `json:"method"` + } + if err := json.NewDecoder(r.Body).Decode(&rpcReq); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + switch rpcReq.Method { + case "getSignaturesForAddress": + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "jsonrpc": "2.0", + "id": 1, + "result": []map[string]interface{}{ + { + "signature": sig, + "slot": 1, + "err": nil, + "blockTime": blockTime, + }, + }, + }) + case "getTransaction": + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "jsonrpc": "2.0", + "id": 1, + "result": map[string]interface{}{ + "blockTime": blockTime, + "meta": map[string]interface{}{"err": nil}, + "transaction": map[string]interface{}{ + "message": map[string]interface{}{ + "instructions": []map[string]interface{}{ + { + "programId": SystemProgramID, + "parsed": map[string]interface{}{ + "type": "transfer", + "info": map[string]interface{}{ + "source": "source-wallet", + "destination": address, + "lamports": 1_230_000_000, + }, + }, + }, + }, + }, + }, + }, + }) + default: + http.Error(w, "unexpected method "+rpcReq.Method, http.StatusBadRequest) + } + })) + defer server.Close() + + if err := dao.Mdb.Create(&mdb.RpcNode{ + Network: mdb.NetworkSolana, + Url: server.URL, + Type: mdb.RpcNodeTypeHttp, + Weight: 1, + Enabled: true, + Status: mdb.RpcNodeStatusOk, + }).Error; err != nil { + t.Fatalf("seed solana rpc_node: %v", err) + } + if err := dao.Mdb.Create(&mdb.ChainToken{ + Network: mdb.NetworkSolana, + Symbol: "SOL", + ContractAddress: "", + Decimals: SOL_Decimals, + Enabled: true, + }).Error; err != nil { + t.Fatalf("seed SOL chain token: %v", err) + } + order := &mdb.Orders{ + TradeId: tradeID, + OrderId: "merchant-sol-retryable-001", + Amount: 100, + Currency: "CNY", + ActualAmount: amount, + ReceiveAddress: address, + Token: "SOL", + Network: mdb.NetworkSolana, + Status: mdb.StatusWaitPay, + } + if err := dao.Mdb.Create(order).Error; err != nil { + t.Fatalf("seed order: %v", err) + } + if err := data.LockTransaction(mdb.NetworkSolana, address, "SOL", tradeID, amount, time.Hour); err != nil { + t.Fatalf("lock transaction: %v", err) + } + + oldProcessSolanaOrder := processSolanaOrder + calls := 0 + processSolanaOrder = func(req *request.OrderProcessingRequest) error { + calls++ + if req.TradeId != tradeID { + t.Fatalf("order processing trade_id = %q, want %q", req.TradeId, tradeID) + } + return errors.New("temporary database error") + } + t.Cleanup(func() { + processSolanaOrder = oldProcessSolanaOrder + }) + + var wg sync.WaitGroup + wg.Add(1) + SolCallBack(address, &wg) + wg.Wait() + + if calls != 1 { + t.Fatalf("order processing calls = %d, want 1", calls) + } + if _, ok := gProcessedSignatures.Load(sig); ok { + t.Fatalf("signature %s was cached after retryable order processing error", sig) + } +} + func TestSolClientHealthy(t *testing.T) { requireSolanaIntegration(t) @@ -78,6 +218,13 @@ func TestSolClientHealthy(t *testing.T) { } } +func clearProcessedSignatures() { + gProcessedSignatures.Range(func(key, value interface{}) bool { + gProcessedSignatures.Delete(key) + return true + }) +} + func TestSolClientGetSignaturesForAddress(t *testing.T) { requireSolanaIntegration(t)