From 3192cca6c3342b40fda000ab36544ce85a09f95b Mon Sep 17 00:00:00 2001 From: Kitipong Sirirueangsakul Date: Mon, 24 Aug 2026 01:50:39 +0700 Subject: [PATCH 1/2] Handle queued XRPL transactions --- .../relayertest/mocks/chain_xrpl_client.go | 8 +- relayer/chains/xrpl/client.go | 94 +++++++++++++++++-- relayer/chains/xrpl/client_test.go | 94 +++++++++++++++++++ relayer/chains/xrpl/provider.go | 5 +- relayer/chains/xrpl/provider_test.go | 2 +- 5 files changed, 188 insertions(+), 15 deletions(-) create mode 100644 relayer/chains/xrpl/client_test.go diff --git a/internal/relayertest/mocks/chain_xrpl_client.go b/internal/relayertest/mocks/chain_xrpl_client.go index 75b1b12f..b6b68aaf 100644 --- a/internal/relayertest/mocks/chain_xrpl_client.go +++ b/internal/relayertest/mocks/chain_xrpl_client.go @@ -60,18 +60,18 @@ func (mr *MockXRPLClientMockRecorder) Autofill(tx any) *gomock.Call { } // BroadcastTx mocks base method. -func (m *MockXRPLClient) BroadcastTx(txBlob string) (xrpl.TxResult, error) { +func (m *MockXRPLClient) BroadcastTx(ctx context.Context, txBlob string) (xrpl.TxResult, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "BroadcastTx", txBlob) + ret := m.ctrl.Call(m, "BroadcastTx", ctx, txBlob) ret0, _ := ret[0].(xrpl.TxResult) ret1, _ := ret[1].(error) return ret0, ret1 } // BroadcastTx indicates an expected call of BroadcastTx. -func (mr *MockXRPLClientMockRecorder) BroadcastTx(txBlob any) *gomock.Call { +func (mr *MockXRPLClientMockRecorder) BroadcastTx(ctx, txBlob any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BroadcastTx", reflect.TypeOf((*MockXRPLClient)(nil).BroadcastTx), txBlob) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BroadcastTx", reflect.TypeOf((*MockXRPLClient)(nil).BroadcastTx), ctx, txBlob) } // CheckAndConnect mocks base method. diff --git a/relayer/chains/xrpl/client.go b/relayer/chains/xrpl/client.go index fbc1b61b..ee7fde69 100644 --- a/relayer/chains/xrpl/client.go +++ b/relayer/chains/xrpl/client.go @@ -2,6 +2,7 @@ package xrpl import ( "context" + "errors" "fmt" "math/big" "net/http" @@ -23,6 +24,13 @@ import ( const RippleEpochOffset = 946684800 +const ( + queuedTransactionResult = "terQUEUED" + successfulTransactionResult = "tesSUCCESS" + transactionNotFoundError = "txnNotFound" + defaultTxPollingInterval = time.Second +) + // idleConnTimeout is shorter than the typical load-balancer idle timeout // (e.g. AWS ALB default 60s) to prevent "connection reset by peer" errors // from stale keep-alive connections being reused after the LB closes them. @@ -44,7 +52,7 @@ type Client interface { GetAccountSequenceNumber(account string) (uint32, error) GetBalance(account string) (*big.Int, error) Autofill(tx *transaction.FlatTransaction) error - BroadcastTx(txBlob string) (TxResult, error) + BroadcastTx(ctx context.Context, txBlob string) (TxResult, error) GetLedgerCloseTime(ledgerIndex common.LedgerIndex) (*time.Time, error) } @@ -55,6 +63,7 @@ type client struct { ChainName string Endpoints []string BlockConfirmation uint32 + TxPollingInterval time.Duration Log logger.Logger alert alert.Alert @@ -74,6 +83,7 @@ func NewClient(chainName string, cfg *XRPLChainProviderConfig, log logger.Logger ChainName: chainName, Endpoints: cfg.Endpoints, BlockConfirmation: 5, + TxPollingInterval: cfg.NonceInterval, Log: log.With("chain_name", chainName), alert: alert, clients: NewXRPLClients(), @@ -308,8 +318,10 @@ func (c *client) Autofill(tx *transaction.FlatTransaction) error { return nil } -// BroadcastTx submits a signed tx blob and returns its hash. -func (c *client) BroadcastTx(txBlob string) (TxResult, error) { +// BroadcastTx submits a signed tx blob and returns its validated result when it +// is queued. A queued transaction must not be rebuilt with a new sequence while +// its final result is still unknown. +func (c *client) BroadcastTx(ctx context.Context, txBlob string) (TxResult, error) { client, err := c.clients.GetSelectedClient() if err != nil { return TxResult{}, fmt.Errorf("failed to get client: %w", err) @@ -340,7 +352,17 @@ func (c *client) BroadcastTx(txBlob string) (TxResult, error) { }, fmt.Errorf("missing fee in submit response") } - if result.EngineResult != "tesSUCCESS" { + txResult := TxResult{ + TxHash: txHash, + Fee: fee, + LedgerIndex: result.ValidatedLedgerIndex, + } + + if result.EngineResult == queuedTransactionResult { + return c.waitForQueuedTransaction(ctx, txResult) + } + + if result.EngineResult != successfulTransactionResult { return TxResult{ TxHash: txHash, Fee: fee, @@ -352,11 +374,65 @@ func (c *client) BroadcastTx(txBlob string) (TxResult, error) { ) } - return TxResult{ - TxHash: txHash, - Fee: fee, - LedgerIndex: result.ValidatedLedgerIndex, - }, nil + return txResult, nil +} + +// waitForQueuedTransaction polls the original transaction hash until its +// validated result is known. Without LastLedgerSequence, txnNotFound and RPC +// errors cannot prove that a queued transaction will never validate, so they +// remain pending until the relayer context is canceled. +func (c *client) waitForQueuedTransaction( + ctx context.Context, + txResult TxResult, +) (TxResult, error) { + pollingInterval := c.TxPollingInterval + if pollingInterval <= 0 { + pollingInterval = defaultTxPollingInterval + } + for { + select { + case <-ctx.Done(): + return txResult, fmt.Errorf("waiting for queued transaction %s: %w", txResult.TxHash, ctx.Err()) + default: + } + + client, err := c.clients.GetSelectedClient() + if err == nil { + var res rpc.XRPLResponse + res, err = client.Request(&requests.TxRequest{Transaction: txResult.TxHash}) + if err == nil { + var txResponse requests.TxResponse + if err = res.GetResult(&txResponse); err == nil && txResponse.Validated { + txResult.LedgerIndex = txResponse.LedgerIndex + if txResponse.Meta.TransactionResult != successfulTransactionResult { + return txResult, fmt.Errorf( + "queued transaction %s validated with engine result %s", + txResult.TxHash, + txResponse.Meta.TransactionResult, + ) + } + + return txResult, nil + } + } + } + + if err != nil && !isTransactionNotFound(err) { + c.Log.Warn("Failed to query queued XRPL transaction", "tx_hash", txResult.TxHash, "err", err) + } + timer := time.NewTimer(pollingInterval) + select { + case <-ctx.Done(): + timer.Stop() + return txResult, fmt.Errorf("waiting for queued transaction %s: %w", txResult.TxHash, ctx.Err()) + case <-timer.C: + } + } +} + +func isTransactionNotFound(err error) bool { + var clientErr *rpc.ClientError + return errors.As(err, &clientErr) && clientErr.ErrorString == transactionNotFoundError } // GetLedgerCloseTime fetches the close time of the ledger with the given index. diff --git a/relayer/chains/xrpl/client_test.go b/relayer/chains/xrpl/client_test.go new file mode 100644 index 00000000..c5ab3ab9 --- /dev/null +++ b/relayer/chains/xrpl/client_test.go @@ -0,0 +1,94 @@ +package xrpl + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/Peersyst/xrpl-go/xrpl/rpc" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + + "github.com/bandprotocol/falcon/relayer/logger" +) + +func TestBroadcastTxWaitsForQueuedTransaction(t *testing.T) { + var submitRequests atomic.Int32 + var txRequests atomic.Int32 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request struct { + Method string `json:"method"` + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + w.Header().Set("Content-Type", "application/json") + switch request.Method { + case "submit": + submitRequests.Add(1) + _ = json.NewEncoder(w).Encode(map[string]any{ + "result": map[string]any{ + "engine_result": queuedTransactionResult, + "engine_result_message": "Held until escalated fee drops", + "tx_json": map[string]any{ + "hash": "HASH", + "Fee": "10", + }, + }, + }) + case "tx": + if txRequests.Add(1) == 1 { + _ = json.NewEncoder(w).Encode(map[string]any{ + "result": map[string]any{"error": "txnNotFound"}, + }) + return + } + + _ = json.NewEncoder(w).Encode(map[string]any{ + "result": map[string]any{ + "hash": "HASH", + "ledger_index": 123, + "meta": map[string]any{ + "TransactionResult": successfulTransactionResult, + }, + "validated": true, + }, + }) + default: + http.Error(w, "unexpected method", http.StatusBadRequest) + } + })) + defer server.Close() + + rpcConfig, err := rpc.NewClientConfig(server.URL) + require.NoError(t, err) + rpcClient := rpc.NewClient(rpcConfig) + clients := NewXRPLClients() + clients.SetClient(server.URL, rpcClient) + clients.SetSelectedEndpoint(server.URL) + + client := &client{ + ChainName: "xrpl-test", + TxPollingInterval: time.Millisecond, + Log: logger.NewZapLogWrapper(zap.NewNop().Sugar()), + clients: clients, + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + result, err := client.BroadcastTx(ctx, "signed-blob") + require.NoError(t, err) + require.Equal(t, "HASH", result.TxHash) + require.Equal(t, "10", result.Fee) + require.Equal(t, 123, result.LedgerIndex.Int()) + require.EqualValues(t, 1, submitRequests.Load()) + require.EqualValues(t, 2, txRequests.Load()) +} diff --git a/relayer/chains/xrpl/provider.go b/relayer/chains/xrpl/provider.go index 8e643eb6..c53058bb 100644 --- a/relayer/chains/xrpl/provider.go +++ b/relayer/chains/xrpl/provider.go @@ -200,10 +200,13 @@ SignerLoop: } } - txResult, err := cp.Client.BroadcastTx(txBlob) + txResult, err := cp.Client.BroadcastTx(ctx, txBlob) if err != nil { log.Error("Broadcast transaction error", "retry_count", retryCount, err) lastErr = err + if ctx.Err() != nil { + return fmt.Errorf("[XRPLProvider] broadcast interrupted: %w", err) + } // save failed tx in db cp.handleSaveTransaction( diff --git a/relayer/chains/xrpl/provider_test.go b/relayer/chains/xrpl/provider_test.go index e04b0b48..fbe90c20 100644 --- a/relayer/chains/xrpl/provider_test.go +++ b/relayer/chains/xrpl/provider_test.go @@ -129,7 +129,7 @@ func (s *XRPLProviderTestSuite) TestRelayPacket() { s.wallet.EXPECT().GetSigners().Return([]wallet.Signer{mockSigner}) s.client.EXPECT().CheckAndConnect(gomock.Any()).Return(nil) s.client.EXPECT().GetAccountSequenceNumber(mockSigner.GetAddress()).Return(uint32(10), nil) - s.client.EXPECT().BroadcastTx(gomock.Any()).Return( + s.client.EXPECT().BroadcastTx(gomock.Any(), gomock.Any()).Return( xrpl.TxResult{TxHash: "HASH", Fee: "100"}, nil, ) From 6cc68c67b6e2288fc526f17cf0b9c131a0d486fd Mon Sep 17 00:00:00 2001 From: Kitipong Sirirueangsakul Date: Mon, 24 Aug 2026 02:01:21 +0700 Subject: [PATCH 2/2] Fix XRPL test lock copy --- relayer/chains/xrpl/client_test.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/relayer/chains/xrpl/client_test.go b/relayer/chains/xrpl/client_test.go index c5ab3ab9..5b9f2333 100644 --- a/relayer/chains/xrpl/client_test.go +++ b/relayer/chains/xrpl/client_test.go @@ -70,16 +70,15 @@ func TestBroadcastTxWaitsForQueuedTransaction(t *testing.T) { rpcConfig, err := rpc.NewClientConfig(server.URL) require.NoError(t, err) rpcClient := rpc.NewClient(rpcConfig) - clients := NewXRPLClients() - clients.SetClient(server.URL, rpcClient) - clients.SetSelectedEndpoint(server.URL) client := &client{ ChainName: "xrpl-test", TxPollingInterval: time.Millisecond, Log: logger.NewZapLogWrapper(zap.NewNop().Sugar()), - clients: clients, + clients: NewXRPLClients(), } + client.clients.SetClient(server.URL, rpcClient) + client.clients.SetSelectedEndpoint(server.URL) ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel()