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.
2026-07-07 02:06:16 +00:00 · 2026-05-22 16:37:50 +08:00
parent 449d22a41b
commit dec1b97b9e
2 changed files with 159 additions and 1 deletions
@@ -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())
 	}
@@ -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)