RFQ API: add GET /ack endpoint

services/rfq/api/client/client.go

@@ -34,6 +34,7 @@
 	GetAllQuotes(ctx context.Context) ([]*model.GetQuoteResponse, error)
 	GetSpecificQuote(ctx context.Context, q *model.GetQuoteSpecificRequest) ([]*model.GetQuoteResponse, error)
 	GetQuoteByRelayerAddress(ctx context.Context, relayerAddr string) ([]*model.GetQuoteResponse, error)
+	GetRelayAck(ctx context.Context, txID string) (*model.GetRelayAckResponse, error)
 	resty() *resty.Client
 }
 
@@ -188,3 +189,24 @@
 
 	return quotes, nil
 }
+
+func (c *unauthenticatedClient) GetRelayAck(ctx context.Context, txID string) (*model.GetRelayAckResponse, error) {
+	var ack *model.GetRelayAckResponse
+	resp, err := c.rClient.R().
+		SetContext(ctx).
+		SetQueryParams(map[string]string{
+			"id": txID,
+		}).
+		SetResult(&ack).
+		Get(rest.AckRoute)
+
+	if err != nil {
+		return nil, fmt.Errorf("error from server: %s %w", resp.Status(), err)
+	}
+
+	if resp.IsError() {
+		return nil, fmt.Errorf("error from server: %s", resp.Status())
+	}
+
+	return ack, nil
+}

services/rfq/api/config/config.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/jftuga/ellipsis"
 	"gopkg.in/yaml.v2"
@@ -21,8 +22,19 @@ type Config struct {
 	Database   DatabaseConfig `yaml:"database"`
 	OmniRPCURL string         `yaml:"omnirpc_url"`
 	// bridges is a map of chainid->address
-	Bridges map[uint32]string `yaml:"bridges"`
-	Port    string            `yaml:"port"`
+	Bridges         map[uint32]string `yaml:"bridges"`
+	Port            string            `yaml:"port"`
+	RelayAckTimeout time.Duration     `yaml:"relay_ack_timeout"`
+}
+
+const defaultRelayAckTimeout = 5 * time.Second
+
+// GetRelayAckTimeout returns the relay ack timeout.
+func (c Config) GetRelayAckTimeout() time.Duration {
+	if c.RelayAckTimeout == 0 {
+		return defaultRelayAckTimeout
+	}
+	return c.RelayAckTimeout
 }
 
 // LoadConfig loads the config from the given path.

services/rfq/api/model/response.go

@@ -25,3 +25,11 @@ type GetQuoteResponse struct {
 	// UpdatedAt is the time that the quote was last upserted
 	UpdatedAt string `json:"updated_at"`
 }
+
+// GetRelayAckResponse contains the schema for a GET /relay/ack response.
+type GetRelayAckResponse struct {
+	// TxID is the transaction ID
+	TransactionID string `json:"tx_id"`
+	// ShouldRelay is a boolean indicating whether the transaction should be relayed
+	ShouldRelay bool `json:"should_relay"`
+}

services/rfq/api/rest/server.go

@@ -5,6 +5,7 @@
 	"context"
 	"fmt"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/ipfs/go-log"
@@ -25,6 +26,7 @@
 	"github.com/synapsecns/sanguine/services/rfq/api/docs"
 	"github.com/synapsecns/sanguine/services/rfq/api/model"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
+	"github.com/synapsecns/sanguine/services/rfq/relayer/relapi"
 )
 
 // QuoterAPIServer is a struct that holds the configuration, database connection, gin engine, RPC client, metrics handler, and fast bridge contracts.
@@ -37,6 +39,11 @@
 	handler             metrics.Handler
 	fastBridgeContracts map[uint32]*fastbridge.FastBridge
 	roleCache           map[uint32]*ttlcache.Cache[string, bool]
+	// relayAckCache contains a set of transactionID values that reflect
+	// transactions that have been acked for relay
+	relayAckCache *ttlcache.Cache[string, bool]
+	// ackMux is a mutex used to ensure that only one transaction id can be acked at a time.
+	ackMux sync.Mutex
 }
 
 // NewAPI holds the configuration, database connection, gin engine, RPC client, metrics handler, and fast bridge contracts.
@@ -80,27 +87,40 @@
 			ttlcache.WithTTL[string, bool](cacheInterval),
 		)
 		roleCache := roles[chainID]
-
 		go roleCache.Start()
 		go func() {
 			<-ctx.Done()
 			roleCache.Stop()
 		}()
 	}
 
+	// create the relay ack cache
+	relayAckCache := ttlcache.New[string, bool](
+		ttlcache.WithTTL[string, bool](cfg.GetRelayAckTimeout()),
+		ttlcache.WithDisableTouchOnHit[string, bool](),
+	)
+	go relayAckCache.Start()
+	go func() {
+		<-ctx.Done()
+		relayAckCache.Stop()
+	}()
+
 	return &QuoterAPIServer{
 		cfg:                 cfg,
 		db:                  store,
 		omnirpcClient:       omniRPCClient,
 		handler:             handler,
 		fastBridgeContracts: bridges,
 		roleCache:           roles,
+		relayAckCache:       relayAckCache,
+		ackMux:              sync.Mutex{},
 	}, nil
 }
 
 // QuoteRoute is the API endpoint for handling quote related requests.
 const (
 	QuoteRoute    = "/quotes"
+	AckRoute      = "/ack"
 	cacheInterval = time.Minute
 )
 
@@ -120,6 +140,7 @@
 	// GET routes without the AuthMiddleware
 	// engine.PUT("/quotes", h.ModifyQuote)
 	engine.GET(QuoteRoute, h.GetQuotes)
+	engine.GET(AckRoute, r.GetRelayAck)
 
 	r.engine = engine
 
@@ -188,3 +209,28 @@
 		c.Next()
 	}
 }
+
+// GetRelayAck checks if a relay is pending or not.
+func (r *QuoterAPIServer) GetRelayAck(c *gin.Context) {
+	transactionID := c.Query("id")
+	if transactionID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Must specify 'id'"})
+		return
+	}
+
+	// If the tx id is already in the cache, it should not be relayed.
+	// Otherwise, insert into the cache.
+	r.ackMux.Lock()
+	ack := r.relayAckCache.Get(transactionID)
+	shouldRelay := ack == nil
+	if shouldRelay {
+		r.relayAckCache.Set(transactionID, true, ttlcache.DefaultTTL)
+	}
+	r.ackMux.Unlock()
+
+	resp := relapi.GetRelayAckResponse{
+		TxID:        transactionID,
+		ShouldRelay: shouldRelay,
+	}
+	c.JSON(http.StatusOK, resp)
+}

services/rfq/api/rest/server_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/synapsecns/sanguine/ethergo/signer/wallet"
 	"github.com/synapsecns/sanguine/services/rfq/api/model"
+	"github.com/synapsecns/sanguine/services/rfq/relayer/relapi"
 )
 
 func (c *ServerSuite) TestNewQuoterAPIServer() {
@@ -203,6 +204,50 @@ func (c *ServerSuite) TestPutAndGetQuoteByRelayer() {
 	c.Assert().True(found, "Newly added quote not found")
 }
 
+func (c *ServerSuite) TestGetAck() {
+	c.startQuoterAPIServer()
+
+	// Send GET request
+	client := &http.Client{}
+	testTxID := "0x123"
+	req, err := http.NewRequestWithContext(c.GetTestContext(), http.MethodGet, fmt.Sprintf("http://localhost:%d/ack?id=%s", c.port, testTxID), nil)
+	c.Require().NoError(err)
+	resp, err := client.Do(req)
+	c.Require().NoError(err)
+	c.Equal(http.StatusOK, resp.StatusCode)
+
+	// Expect ack with shouldRelay=true
+	var result relapi.GetRelayAckResponse
+	err = json.NewDecoder(resp.Body).Decode(&result)
+	c.Require().NoError(err)
+	expectedResult := relapi.GetRelayAckResponse{
+		TxID:        testTxID,
+		ShouldRelay: true,
+	}
+	c.Equal(expectedResult, result)
+	err = resp.Body.Close()
+	c.Require().NoError(err)
+
+	// Send another request with same txID
+	req, err = http.NewRequestWithContext(c.GetTestContext(), http.MethodGet, fmt.Sprintf("http://localhost:%d/ack?id=%s", c.port, testTxID), nil)
+	c.Require().NoError(err)
+	resp, err = client.Do(req)
+	c.Require().NoError(err)
+	c.Equal(http.StatusOK, resp.StatusCode)
+
+	// Expect ack with shouldRelay=false
+	err = json.NewDecoder(resp.Body).Decode(&result)
+	c.Require().NoError(err)
+	expectedResult = relapi.GetRelayAckResponse{
+		TxID:        testTxID,
+		ShouldRelay: false,
+	}
+	c.Equal(expectedResult, result)
+	err = resp.Body.Close()
+	c.Require().NoError(err)
+	c.GetTestContext().Done()
+}
+
 // startQuoterAPIServer starts the API server and waits for it to initialize.
 func (c *ServerSuite) startQuoterAPIServer() {
 	go func() {

services/rfq/relayer/relapi/handler.go

@@ -60,7 +60,7 @@
 func (h *Handler) GetQuoteRequestStatusByTxID(c *gin.Context) {
 	txIDStr := c.Query("id")
 	if txIDStr == "" {
-		c.JSON(http.StatusBadRequest, gin.H{"error": "Must specify 'txID'"})
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Must specify 'id'"})
 		return
 	}
 

services/rfq/relayer/relapi/model.go

@@ -15,3 +15,9 @@ type GetTxRetryResponse struct {
 	Nonce     uint64 `json:"nonce"`
 	GasAmount string `json:"gas_amount"`
 }
+
+// GetRelayAckResponse contains the schema for a POST /relay/ack response.
+type GetRelayAckResponse struct {
+	TxID        string `json:"tx_id"`
+	ShouldRelay bool   `json:"should_relay"`
+}

services/rfq/relayer/service/handlers.go

@@ -138,14 +138,25 @@
 		return fmt.Errorf("could not get committable balance: %w", err)
 	}
 
-	// if committableBalance > destAmount
+	// check if we have enough inventory to handle the request
 	if committableBalance.Cmp(request.Transaction.DestAmount) < 0 {
 		err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.NotEnoughInventory)
 		if err != nil {
 			return fmt.Errorf("could not update request status: %w", err)
 		}
 		return nil
 	}
+
+	// get ack from API to synchronize calls with other relayers and avoid reverts
+	resp, err := q.apiClient.GetRelayAck(ctx, hexutil.Encode(request.TransactionID[:]))
+	if err != nil {
+		return fmt.Errorf("could not get relay ack: %w", err)
+	}
+	if !resp.ShouldRelay {
+		span.SetAttributes(attribute.Bool("should_relay", false))
+		return nil
+	}
+
 	err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.CommittedPending)
 	if err != nil {
 		return fmt.Errorf("could not update request status: %w", err)

services/rfq/relayer/service/relayer.go

@@ -21,6 +21,7 @@
 	cctpSql "github.com/synapsecns/sanguine/services/cctp-relayer/db/sql"
 	"github.com/synapsecns/sanguine/services/cctp-relayer/relayer"
 	omniClient "github.com/synapsecns/sanguine/services/omnirpc/client"
+	rfqAPIClient "github.com/synapsecns/sanguine/services/rfq/api/client"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/inventory"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/pricer"
@@ -40,6 +41,7 @@
 	client         omniClient.RPCClient
 	chainListeners map[int]listener.ContractListener
 	apiServer      *relapi.RelayerAPIServer
+	apiClient      rfqAPIClient.UnauthenticatedClient
 	inventory      inventory.Manager
 	quoter         quoter.Quoter
 	submitter      submitter.TransactionSubmitter
@@ -120,6 +122,11 @@
 		return nil, fmt.Errorf("could not get api server: %w", err)
 	}
 
+	apiClient, err := rfqAPIClient.NewUnauthenticatedClient(metricHandler, cfg.GetRfqAPIURL())
+	if err != nil {
+		return nil, fmt.Errorf("error creating RFQ API client: %w", err)
+	}
+
 	cache := ttlcache.New[common.Hash, bool](ttlcache.WithTTL[common.Hash, bool](time.Second * 30))
 	rel := Relayer{
 		db:             store,
@@ -134,6 +141,7 @@
 		signer:         sg,
 		chainListeners: chainListeners,
 		apiServer:      apiServer,
+		apiClient:      apiClient,
 	}
 	return &rel, nil
 }

services/rfq/relayer/service/statushandler.go

@@ -10,6 +10,7 @@
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/jellydator/ttlcache/v3"
 	"github.com/synapsecns/sanguine/core/metrics"
+	"github.com/synapsecns/sanguine/services/rfq/api/client"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/chain"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/inventory"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/quoter"
@@ -42,6 +43,8 @@
 	RelayerAddress common.Address
 	// metrics is the metrics handler.
 	metrics metrics.Handler
+	// apiClient is used to get acks before submitting a relay transaction.
+	apiClient client.UnauthenticatedClient
 }
 
 // Handler is the handler for a quote request.
@@ -68,6 +71,7 @@
 		metrics:        r.metrics,
 		RelayerAddress: r.signer.Address(),
 		claimCache:     r.claimCache,
+		apiClient:      r.apiClient,
 	}
 
 	qr.handlers[reldb.Seen] = r.deadlineMiddleware(r.gasMiddleware(qr.handleSeen))