feat(rfq-relayer): relayer supports active quoting (#3198)

* Feat: add active rfq subscription on quoter

* Feat: relayer subscribes to active quotes upon starting

* [goreleaser]

* Feat: specify ws url in relayer

* [goreleaser]

* [goreleaser]

* Fix: build

* [goreleaser]

* Feat: relayer tracing

* [goreleaser]

* Feat: use supports_active_quoting instead of ws url

* [goreleaser]

* WIP: add logs

* [goreleaser]

* WIP: more logs

* [goreleaser]

* More logs

* [goreleaser]

* More logs

* [goreleaser]

* More logs

* [goreleaser]

* Close conn when encountering write err

* [goreleaser]

* More logs

* [goreleaser]

* More logs

* [goreleaser]

* More logs

* [goreleaser]

* More logs

* [goreleaser]

* Logs with ts

* [goreleaser]

* More tracing

* [goreleaser]

* Fix: send to reqChan

* [goreleaser]

* Check for zero pong time

* Fix: make close_at and closed_quote_id optional

* [goreleaser]

* Feat: remove extra fields from responses

* [goreleaser]

* Fix: skip passive quote

* [goreleaser]

* Cleanup: remove logs

* Fix: use correct span

* Cleanup: remove logs

services/rfq/api/db/api_db.go

@@ -153,8 +153,8 @@ type ActiveQuoteRequest struct {
 	ExpirationWindow time.Duration            `gorm:"column:expiration_window"`
 	CreatedAt        time.Time                `gorm:"column:created_at"`
 	Status           ActiveQuoteRequestStatus `gorm:"column:status"`
-	ClosedAt         time.Time                `gorm:"column:fulfilled_at"`
-	ClosedQuoteID    string                   `gorm:"column:fulfilled_quote_id"`
+	ClosedAt         *time.Time               `gorm:"column:closed_at"`
+	ClosedQuoteID    *string                  `gorm:"column:closed_quote_id"`
 }
 
 // FromUserRequest converts a model.PutRFQRequest to an ActiveQuoteRequest.

services/rfq/api/db/sql/base/store.go

@@ -102,8 +102,8 @@ func (s *Store) UpdateActiveQuoteRequestStatus(ctx context.Context, requestID st
 		if quoteID == nil {
 			return fmt.Errorf("quote id is required for fulfilled status")
 		}
-		updates["fulfilled_quote_id"] = quoteID
-		updates["fulfilled_at"] = time.Now().UTC()
+		updates["closed_quote_id"] = quoteID
+		updates["closed_at"] = time.Now().UTC()
 	}
 	result := s.db.WithContext(ctx).
 		Model(&db.ActiveQuoteRequest{}).

services/rfq/api/model/response.go

@@ -56,17 +56,17 @@ type ActiveRFQMessage struct {
 
 // PutUserQuoteResponse represents a response to a user quote request.
 type PutUserQuoteResponse struct {
-	Success     bool      `json:"success"`
-	Reason      string    `json:"reason"`
-	UserAddress string    `json:"user_address"`
-	QuoteType   string    `json:"quote_type"`
-	Data        QuoteData `json:"data"`
+	Success        bool   `json:"success"`
+	Reason         string `json:"reason,omitempty"`
+	QuoteType      string `json:"quote_type,omitempty"`
+	DestAmount     string `json:"dest_amount,omitempty"`
+	RelayerAddress string `json:"relayer_address,omitempty"`
 }
 
 // WsRFQResponse represents a response to a quote request.
 type WsRFQResponse struct {
 	RequestID  string    `json:"request_id"`
-	QuoteID    string    `json:"quote_id"`
+	QuoteID    string    `json:"quote_id,omitempty"`
 	DestAmount string    `json:"dest_amount"`
 	UpdatedAt  time.Time `json:"updated_at"`
 }

services/rfq/api/rest/rfq.go

@@ -55,11 +55,12 @@ func (r *QuoterAPIServer) handleActiveRFQ(ctx context.Context, request *model.Pu
 	responses := r.collectRelayerResponses(ctx, request, requestID)
 	var quoteID string
 	var isUpdated bool
-	for _, resp := range responses {
+	for relayerAddr, resp := range responses {
 		quote, isUpdated = getBestQuote(quote, getRelayerQuoteData(request, resp))
 		if isUpdated {
 			quoteID = resp.QuoteID
 		}
+		quote.RelayerAddress = &relayerAddr
 	}
 	err = r.recordActiveQuote(ctx, quote, requestID, quoteID)
 	if err != nil {
@@ -90,13 +91,14 @@ func (r *QuoterAPIServer) collectRelayerResponses(ctx context.Context, request *
 		wg.Add(1)
 		go func(client WsClient) {
 			var respStatus db.ActiveQuoteResponseStatus
+			var err error
 			_, clientSpan := r.handler.Tracer().Start(collectionCtx, "collectRelayerResponses", trace.WithAttributes(
 				attribute.String("relayer_address", relayerAddr),
 				attribute.String("request_id", requestID),
 			))
 			defer func() {
 				clientSpan.SetAttributes(attribute.String("status", respStatus.String()))
-				metrics.EndSpan(clientSpan)
+				metrics.EndSpanWithErr(clientSpan, err)
 			}()
 
 			defer wg.Done()
@@ -105,6 +107,11 @@ func (r *QuoterAPIServer) collectRelayerResponses(ctx context.Context, request *
 				logger.Errorf("Error receiving quote response: %v", err)
 				return
 			}
+			clientSpan.AddEvent("received quote response", trace.WithAttributes(
+				attribute.String("relayer_address", relayerAddr),
+				attribute.String("request_id", requestID),
+				attribute.String("dest_amount", resp.DestAmount),
+			))
 
 			// validate the response
 			respStatus = getQuoteResponseStatus(expireCtx, resp)
@@ -247,6 +254,9 @@ func (r *QuoterAPIServer) handlePassiveRFQ(ctx context.Context, request *model.P
 		)
 
 		rawDestAmountInt, _ := rawDestAmount.Int(nil)
+		if rawDestAmountInt.Cmp(quote.FixedFee.BigInt()) < 0 {
+			continue
+		}
 		destAmount := new(big.Int).Sub(rawDestAmountInt, quote.FixedFee.BigInt()).String()
 		//nolint:gosec
 		quoteData := &model.QuoteData{

services/rfq/api/rest/rfq_test.go

@@ -103,7 +103,6 @@ func (c *ServerSuite) TestActiveRFQSingleRelayer() {
 	}
 
 	// Prepare the relayer quote response
-	originAmount := userRequestAmount.String()
 	destAmount := new(big.Int).Sub(userRequestAmount, big.NewInt(1000)).String()
 	quoteResp := &model.WsRFQResponse{
 		DestAmount: destAmount,
@@ -119,8 +118,7 @@ func (c *ServerSuite) TestActiveRFQSingleRelayer() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("active", userQuoteResp.QuoteType)
-	c.Assert().Equal(destAmount, *userQuoteResp.Data.DestAmount)
-	c.Assert().Equal(originAmount, userQuoteResp.Data.OriginAmount)
+	c.Assert().Equal(destAmount, userQuoteResp.DestAmount)
 
 	// Verify ActiveQuoteRequest insertion
 	activeQuoteRequests, err := c.database.GetActiveQuoteRequests(c.GetTestContext())
@@ -206,7 +204,6 @@ func (c *ServerSuite) TestActiveRFQMultipleRelayers() {
 	}
 
 	// Prepare the relayer quote responses
-	originAmount := userRequestAmount.String()
 	destAmount := "999000"
 	quoteResp := model.WsRFQResponse{
 		DestAmount: destAmount,
@@ -234,8 +231,7 @@ func (c *ServerSuite) TestActiveRFQMultipleRelayers() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("active", userQuoteResp.QuoteType)
-	c.Assert().Equal(destAmount, *userQuoteResp.Data.DestAmount)
-	c.Assert().Equal(originAmount, userQuoteResp.Data.OriginAmount)
+	c.Assert().Equal(destAmount, userQuoteResp.DestAmount)
 
 	// Verify ActiveQuoteRequest insertion
 	activeQuoteRequests, err := c.database.GetActiveQuoteRequests(c.GetTestContext())
@@ -309,9 +305,8 @@ func (c *ServerSuite) TestActiveRFQFallbackToPassive() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("passive", userQuoteResp.QuoteType)
-	c.Assert().Equal("998000", *userQuoteResp.Data.DestAmount) // destAmount is quote destAmount minus fixed fee
-	c.Assert().Equal(userRequestAmount.String(), userQuoteResp.Data.OriginAmount)
-	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), *userQuoteResp.Data.RelayerAddress)
+	c.Assert().Equal("998000", userQuoteResp.DestAmount) // destAmount is quote destAmount minus fixed fee
+	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), userQuoteResp.RelayerAddress)
 }
 
 func (c *ServerSuite) TestActiveRFQPassiveBestQuote() {
@@ -389,7 +384,6 @@ func (c *ServerSuite) TestActiveRFQPassiveBestQuote() {
 	// Assert the response
 	c.Assert().True(userQuoteResp.Success)
 	c.Assert().Equal("passive", userQuoteResp.QuoteType)
-	c.Assert().Equal("998900", *userQuoteResp.Data.DestAmount) // destAmount is quote destAmount minus fixed fee
-	c.Assert().Equal(userRequestAmount.String(), userQuoteResp.Data.OriginAmount)
-	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), *userQuoteResp.Data.RelayerAddress)
+	c.Assert().Equal("998900", userQuoteResp.DestAmount) // destAmount is quote destAmount minus fixed fee
+	c.Assert().Equal(c.relayerWallets[0].Address().Hex(), userQuoteResp.RelayerAddress)
 }

services/rfq/api/rest/server.go

@@ -533,16 +533,23 @@ func (r *QuoterAPIServer) PutRFQRequest(c *gin.Context) {
 	var activeQuote *model.QuoteData
 	if isActiveRFQ {
 		activeQuote = r.handleActiveRFQ(ctx, &req, requestID)
+		if activeQuote != nil && activeQuote.DestAmount != nil {
+			span.SetAttributes(attribute.String("active_quote_dest_amount", *activeQuote.DestAmount))
+		}
 	}
 	passiveQuote, err := r.handlePassiveRFQ(ctx, &req)
 	if err != nil {
 		logger.Error("Error handling passive RFQ", "error", err)
 	}
+	if passiveQuote != nil && passiveQuote.DestAmount != nil {
+		span.SetAttributes(attribute.String("passive_quote_dest_amount", *passiveQuote.DestAmount))
+	}
 	quote, _ := getBestQuote(activeQuote, passiveQuote)
 
 	// construct the response
 	var resp model.PutUserQuoteResponse
 	if quote == nil {
+		span.AddEvent("no quotes found")
 		resp = model.PutUserQuoteResponse{
 			Success: false,
 			Reason:  "no quotes found",
@@ -552,10 +559,15 @@ func (r *QuoterAPIServer) PutRFQRequest(c *gin.Context) {
 		if activeQuote == nil {
 			quoteType = quoteTypePassive
 		}
+		span.SetAttributes(
+			attribute.String("quote_type", quoteType),
+			attribute.String("quote_dest_amount", *quote.DestAmount),
+		)
 		resp = model.PutUserQuoteResponse{
-			Success:   true,
-			Data:      *quote,
-			QuoteType: quoteType,
+			Success:        true,
+			QuoteType:      quoteType,
+			DestAmount:     *quote.DestAmount,
+			RelayerAddress: *quote.RelayerAddress,
 		}
 	}
 	c.JSON(http.StatusOK, resp)

services/rfq/api/rest/suite_test.go

@@ -44,7 +44,6 @@ type ServerSuite struct {
 	handler              metrics.Handler
 	QuoterAPIServer      *rest.QuoterAPIServer
 	port                 uint16
-	wsPort               uint16
 	originChainID        int
 	destChainID          int
 }
@@ -75,8 +74,6 @@ func (c *ServerSuite) SetupTest() {
 	c.True(ok)
 	port, err := freeport.GetFreePort()
 	c.port = uint16(port)
-	wsPort, err := freeport.GetFreePort()
-	c.wsPort = uint16(wsPort)
 	c.Require().NoError(err)
 
 	testConfig := config.Config{

services/rfq/api/rest/ws.go

@@ -96,7 +96,8 @@ const (
 
 // Run runs the WebSocket client.
 func (c *wsClient) Run(ctx context.Context) (err error) {
-	c.lastPong = time.Now()
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	messageChan := make(chan []byte)
 	pingTicker := time.NewTicker(PingPeriod)
 	defer pingTicker.Stop()
@@ -122,11 +123,12 @@ func (c *wsClient) Run(ctx context.Context) (err error) {
 			err = c.handleRelayerMessage(ctx, msg)
 			if err != nil {
 				logger.Error("Error handling relayer message: %s", err)
+				return fmt.Errorf("error handling relayer message: %w", err)
 			}
 		case <-pingTicker.C:
 			err = c.trySendPing(c.lastPong)
 			if err != nil {
-				logger.Error("Error sending ping message: %s", err)
+				logger.Error("Error sending ping message: %w", err)
 			}
 		}
 	}
@@ -169,6 +171,8 @@ func (c *wsClient) sendRelayerRequest(ctx context.Context, req *model.WsRFQReque
 	return nil
 }
 
+// handleRelayerMessage handles messages from the relayer.
+// An error returned will result in the websocket connection being closed.
 func (c *wsClient) handleRelayerMessage(ctx context.Context, msg []byte) (err error) {
 	var rfqMsg model.ActiveRFQMessage
 	err = json.Unmarshal(msg, &rfqMsg)
@@ -191,13 +195,12 @@ func (c *wsClient) handleRelayerMessage(ctx context.Context, msg []byte) (err er
 		}
 	case SendQuoteOp:
 		err = c.handleSendQuote(ctx, rfqMsg.Content)
-		if err != nil {
-			return fmt.Errorf("error handling send quote: %w", err)
-		}
+		logger.Errorf("error handling send quote: %v", err)
 	case PongOp:
 		c.lastPong = time.Now()
 	default:
-		return fmt.Errorf("received unexpected operation from relayer: %s", rfqMsg.Op)
+		logger.Errorf("received unexpected operation from relayer: %s", rfqMsg.Op)
+		return nil
 	}
 
 	return nil
@@ -273,7 +276,7 @@ func (c *wsClient) handleSendQuote(ctx context.Context, content json.RawMessage)
 }
 
 func (c *wsClient) trySendPing(lastPong time.Time) (err error) {
-	if time.Since(lastPong) > PingPeriod {
+	if time.Since(lastPong) > PingPeriod && !lastPong.IsZero() {
 		err = c.conn.Close()
 		if err != nil {
 			return fmt.Errorf("error closing websocket connection: %w", err)

services/rfq/e2e/setup_test.go

@@ -314,7 +314,7 @@ func (i *IntegrationSuite) getRelayerConfig() relconfig.Config {
 		},
 		// generated ex-post facto
 		QuotableTokens: map[string][]string{},
-		RfqAPIURL:      i.apiServer,
+		RFQAPIURL:      i.apiServer,
 		Signer: signerConfig.SignerConfig{
 			Type: signerConfig.FileType.String(),
 			File: filet.TmpFile(i.T(), "", i.relayerWallet.PrivateKeyHex()).Name(),