RFQ Relayer: concurrent request processing

services/rfq/api/rest/server.go

@@ -290,7 +290,7 @@ func (r *QuoterAPIServer) PutRelayAck(c *gin.Context) {
 	// Otherwise, insert the current relayer's address into the cache.
 	r.ackMux.Lock()
 	ack := r.relayAckCache.Get(ackReq.TxID)
-	shouldRelay := ack == nil
+	shouldRelay := ack == nil || common.HexToAddress(relayerAddr).Hex() == common.HexToAddress(ack.Value()).Hex()
 	if shouldRelay {
 		r.relayAckCache.Set(ackReq.TxID, relayerAddr, ttlcache.DefaultTTL)
 	} else {

services/rfq/api/rest/server_test.go

@@ -235,12 +235,12 @@ func (c *ServerSuite) TestPutAck() {
 	c.Require().NoError(err)
 	c.Equal(http.StatusOK, resp.StatusCode)
 
-	// Expect ack with shouldRelay=false
+	// Expect ack with shouldRelay=true
 	err = json.NewDecoder(resp.Body).Decode(&result)
 	c.Require().NoError(err)
 	expectedResult = relapi.PutRelayAckResponse{
 		TxID:           testTxID,
-		ShouldRelay:    false,
+		ShouldRelay:    true,
 		RelayerAddress: c.testWallet.Address().Hex(),
 	}
 	c.Equal(expectedResult, result)

services/rfq/relayer/quoter/quoter.go

@@ -25,6 +25,7 @@
 	"github.com/synapsecns/sanguine/services/rfq/relayer/relconfig"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb"
 	"golang.org/x/exp/slices"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/synapsecns/sanguine/ethergo/signer/signer"
@@ -149,23 +150,35 @@
 	}
 
 	if m.screener != nil {
-		blocked, err := m.screener.ScreenAddress(ctx, screenerRuleset, quote.Transaction.OriginSender.String())
-		if err != nil {
-			span.RecordError(fmt.Errorf("error screening address: %w", err))
-			return false, fmt.Errorf("error screening address: %w", err)
-		}
-		if blocked {
-			span.AddEvent(fmt.Sprintf("address %s blocked", quote.Transaction.OriginSender))
-			return false, nil
-		}
+		// screen sender and recipient in parallel
+		g, gctx := errgroup.WithContext(ctx)
+		var senderBlocked, recipientBlocked bool
+		g.Go(func() error {
+			senderBlocked, err = m.screener.ScreenAddress(gctx, screenerRuleset, quote.Transaction.OriginSender.String())
+			if err != nil {
+				span.RecordError(fmt.Errorf("error screening address: %w", err))
+				return fmt.Errorf("error screening address: %w", err)
+			}
+			return nil
+		})
+		g.Go(func() error {
+			recipientBlocked, err = m.screener.ScreenAddress(gctx, screenerRuleset, quote.Transaction.DestRecipient.String())
+			if err != nil {
+				span.RecordError(fmt.Errorf("error screening address: %w", err))
+				return fmt.Errorf("error screening address: %w", err)
+			}
+			return nil
+		})
 
-		blocked, err = m.screener.ScreenAddress(ctx, screenerRuleset, quote.Transaction.DestRecipient.String())
+		err = g.Wait()
 		if err != nil {
-			span.RecordError(fmt.Errorf("error screening address: %w", err))
-			return false, fmt.Errorf("error screening address: %w", err)
+			return false, fmt.Errorf("error screening addresses: %w", err)
 		}
-		if blocked {
-			span.AddEvent(fmt.Sprintf("address %s blocked", quote.Transaction.DestRecipient))
+		if senderBlocked || recipientBlocked {
+			span.SetAttributes(
+				attribute.Bool("sender_blocked", senderBlocked),
+				attribute.Bool("recipient_blocked", recipientBlocked),
+			)
 			return false, nil
 		}
 	}

services/rfq/relayer/reldb/base/rebalance.go

@@ -5,6 +5,8 @@
 	"errors"
 	"fmt"
 
+	"database/sql"
+
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb"
 	"gorm.io/gorm"
 )
@@ -125,3 +127,13 @@
 
 	return rebalance, nil
 }
+
+// GetDBStats gets the database stats.
+func (s Store) GetDBStats(ctx context.Context) (*sql.DBStats, error) {
+	sqlDB, err := s.DB().WithContext(ctx).DB()
+	if err != nil {
+		return nil, fmt.Errorf("could not get db: %w", err)
+	}
+	stats := sqlDB.Stats()
+	return &stats, nil
+}

services/rfq/relayer/reldb/db.go

@@ -2,6 +2,7 @@ package reldb
 
 import (
 	"context"
+	"database/sql"
 	"database/sql/driver"
 	"errors"
 	"fmt"
@@ -43,6 +44,8 @@ type Reader interface {
 	GetPendingRebalances(ctx context.Context, chainIDs ...uint64) ([]*Rebalance, error)
 	// GetRebalance gets a rebalance by ID. Should return ErrNoRebalanceForID if not found.
 	GetRebalanceByID(ctx context.Context, rebalanceID string) (*Rebalance, error)
+	// GetDBStats gets the database stats.
+	GetDBStats(ctx context.Context) (*sql.DBStats, error)
 }
 
 // Service is the interface for the database service.

services/rfq/relayer/reldb/mysql/mysql.go

@@ -4,14 +4,15 @@ package mysql
 import (
 	"context"
 	"fmt"
+	"time"
+
 	"github.com/ipfs/go-log"
 	"github.com/synapsecns/sanguine/core/dbcommon"
 	"github.com/synapsecns/sanguine/core/metrics"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb/base"
 	"gorm.io/driver/mysql"
 	"gorm.io/gorm"
 	"gorm.io/gorm/schema"
-	"time"
 )
 
 var logger = log.Logger("mysql-logger")

services/rfq/relayer/service/handlers.go

@@ -5,11 +5,13 @@
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/synapsecns/sanguine/core/metrics"
+	"github.com/synapsecns/sanguine/core/retry"
 	"github.com/synapsecns/sanguine/services/rfq/api/model"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/inventory"
@@ -18,6 +20,8 @@
 	"go.opentelemetry.io/otel/trace"
 )
 
+var maxRPCRetryTime = 15 * time.Second
+
 // handleBridgeRequestedLog handles the BridgeRequestedLog event.
 // Step 1: Seen
 //
@@ -33,14 +37,26 @@
 		metrics.EndSpanWithErr(span, err)
 	}()
 
-	// TODO: consider a mapmutex
+	unlocker, ok := r.relayMtx.TryLock(hexutil.Encode(req.TransactionId[:]))
+	if !ok {
+		span.SetAttributes(attribute.Bool("locked", true))
+		// already processing this request
+		return nil
+	}
+
+	defer unlocker.Unlock()
+
 	_, err = r.db.GetQuoteRequestByID(ctx, req.TransactionId)
 	// expect no results
 	if !errors.Is(err, reldb.ErrNoQuoteForID) {
 		// maybe a db err? if so error out & try again later
 		if err != nil {
 			return fmt.Errorf("could not call db: %w", err)
 		}
+
+		span.AddEvent("already known")
+		// already seen this request
+		return nil
 	}
 
 	// TODO: these should be premade
@@ -54,9 +70,17 @@
 		return fmt.Errorf("could not get correct fast bridge: %w", err)
 	}
 
-	bridgeTx, err := fastBridge.GetBridgeTransaction(&bind.CallOpts{Context: ctx}, req.Request)
+	var bridgeTx fastbridge.IFastBridgeBridgeTransaction
+	call := func(ctx context.Context) error {
+		bridgeTx, err = fastBridge.GetBridgeTransaction(&bind.CallOpts{Context: ctx}, req.Request)
+		if err != nil {
+			return fmt.Errorf("could not get bridge transaction: %w", err)
+		}
+		return nil
+	}
+	err = retry.WithBackoff(ctx, call, retry.WithMaxTotalTime(maxRPCRetryTime))
 	if err != nil {
-		return fmt.Errorf("could not get bridge transaction: %w", err)
+		return fmt.Errorf("could not make call: %w", err)
 	}
 
 	// TODO: you can just pull these out of inventory. If they don't exist mark as invalid.
@@ -71,7 +95,7 @@
 		return fmt.Errorf("could not get decimals: %w", err)
 	}
 
-	err = r.db.StoreQuoteRequest(ctx, reldb.QuoteRequest{
+	dbReq := reldb.QuoteRequest{
 		BlockNumber:         req.Raw.BlockNumber,
 		RawRequest:          req.Request,
 		OriginTokenDecimals: *originDecimals,
@@ -81,11 +105,25 @@
 		Transaction:         bridgeTx,
 		Status:              reldb.Seen,
 		OriginTxHash:        req.Raw.TxHash,
-	})
+	}
+	err = r.db.StoreQuoteRequest(ctx, dbReq)
 	if err != nil {
 		return fmt.Errorf("could not get db: %w", err)
 	}
 
+	// immediately forward the request to handleSeen
+	span.AddEvent("sending to handleSeen")
+	qr, err := r.requestToHandler(ctx, dbReq)
+	if err != nil {
+		return fmt.Errorf("could not get quote request handler: %w", err)
+	}
+	// Forward instead of lock since we called lock above.
+	fwdErr := qr.Forward(ctx, dbReq)
+	if fwdErr != nil {
+		logger.Errorf("could not forward to handle seen: %w", fwdErr)
+		span.AddEvent("could not forward to handle seen")
+	}
+
 	return nil
 }
 
@@ -167,10 +205,20 @@
 		return nil
 	}
 
+	request.Status = reldb.CommittedPending
 	err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.CommittedPending)
 	if err != nil {
 		return fmt.Errorf("could not update request status: %w", err)
 	}
+
+	// immediately forward the request to handleCommitPending
+	span.AddEvent("forwarding to handleCommitPending")
+	fwdErr := q.Forward(ctx, request)
+	if fwdErr != nil {
+		logger.Errorf("could not forward to handle commit pending: %w", fwdErr)
+		span.AddEvent("could not forward to handle commit pending")
+	}
+
 	return nil
 }
 
@@ -201,20 +249,40 @@
 		return nil
 	}
 
-	bs, err := q.Origin.Bridge.BridgeStatuses(&bind.CallOpts{Context: ctx}, request.TransactionID)
+	var bs uint8
+	call := func(ctx context.Context) error {
+		bs, err = q.Origin.Bridge.BridgeStatuses(&bind.CallOpts{Context: ctx}, request.TransactionID)
+		if err != nil {
+			return fmt.Errorf("could not get bridge status: %w", err)
+		}
+		return nil
+	}
+	err = retry.WithBackoff(ctx, call, retry.WithMaxTotalTime(maxRPCRetryTime))
 	if err != nil {
-		return fmt.Errorf("could not get bridge status: %w", err)
+		return fmt.Errorf("could not make contract call: %w", err)
 	}
 
 	span.AddEvent("status_check", trace.WithAttributes(attribute.String("chain_bridge_status", fastbridge.BridgeStatus(bs).String())))
 
 	// sanity check to make sure it's still requested.
-	if bs == fastbridge.REQUESTED.Int() {
-		err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.CommittedConfirmed)
-		if err != nil {
-			return fmt.Errorf("could not update request status: %w", err)
-		}
+	if bs != fastbridge.REQUESTED.Int() {
+		return nil
 	}
+
+	request.Status = reldb.CommittedConfirmed
+	err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.CommittedConfirmed)
+	if err != nil {
+		return fmt.Errorf("could not update request status: %w", err)
+	}
+
+	// immediately forward to handleCommitConfirmed
+	span.AddEvent("forwarding to handleCommitConfirmed")
+	fwdErr := q.Forward(ctx, request)
+	if fwdErr != nil {
+		logger.Errorf("could not forward to handle commit confirmed: %w", fwdErr)
+		span.AddEvent("could not forward to handle commit confirmed")
+	}
+
 	return nil
 }
 
@@ -253,7 +321,9 @@
 		return fmt.Errorf("could not get quote request: %w", err)
 	}
 	// we might've accidentally gotten this later, if so we'll just ignore it
-	if reqID.Status != reldb.RelayStarted {
+	// note that in the edge case where we pessimistically marked as DeadlineExceeded
+	// and the relay was actually succesful, we should continue the proving process
+	if reqID.Status != reldb.RelayStarted && reqID.Status != reldb.DeadlineExceeded {
 		logger.Warnf("got relay log for request that was not relay started (transaction id: %s, txhash: %s)", hexutil.Encode(reqID.TransactionID[:]), req.Raw.TxHash)
 		return nil
 	}
@@ -321,9 +391,17 @@
 
 	// make sure relayer hasn't already proved. This is neeeded in case of an abrupt halt in event sourcing
 	// note:  this assumes caller has already checked the sender is the relayer.
-	bs, err := q.Origin.Bridge.BridgeStatuses(&bind.CallOpts{Context: ctx}, request.TransactionID)
+	var bs uint8
+	call := func(ctx context.Context) error {
+		bs, err = q.Origin.Bridge.BridgeStatuses(&bind.CallOpts{Context: ctx}, request.TransactionID)
+		if err != nil {
+			return fmt.Errorf("could not get bridge status: %w", err)
+		}
+		return nil
+	}
+	err = retry.WithBackoff(ctx, call, retry.WithMaxTotalTime(maxRPCRetryTime))
 	if err != nil {
-		return fmt.Errorf("could not get bridge status: %w", err)
+		return fmt.Errorf("could not make contract call: %w", err)
 	}
 
 	if bs == fastbridge.RelayerClaimed.Int() {
@@ -334,9 +412,17 @@
 		return nil
 	}
 
-	canClaim, err := q.Origin.Bridge.CanClaim(&bind.CallOpts{Context: ctx}, request.TransactionID, q.RelayerAddress)
+	var canClaim bool
+	claimCall := func(ctx context.Context) error {
+		canClaim, err = q.Origin.Bridge.CanClaim(&bind.CallOpts{Context: ctx}, request.TransactionID, q.RelayerAddress)
+		if err != nil {
+			return fmt.Errorf("could not check if can claim: %w", err)
+		}
+		return nil
+	}
+	err = retry.WithBackoff(ctx, claimCall, retry.WithMaxTotalTime(maxRPCRetryTime))
 	if err != nil {
-		return fmt.Errorf("could not check if can claim: %w", err)
+		return fmt.Errorf("could not make call: %w", err)
 	}
 
 	// can't claim yet. we'll check again later