Add RFQ Guard (#2840)

Co-authored-by: Trajan0x <trajan0x@users.noreply.github.com>

docs/bridge/docs/rfq/Relayer/Relayer.md

@@ -193,6 +193,8 @@ The relayer is configured with a yaml file. The following is an example configur
  - `rebalance_interval` - How often to rebalance, formatted as (s = seconds, m = minutes, h = hours)
  - `relayer_api_port` - the relayer api is used to control the relayer. <!--TODO: more info here--> This api should be secured/not public.
  - `base_chain_config`: Base chain config is the default config applied for each chain if the other chains do not override it. This is covered in the chains section.
+ - `enable_guard`  - Run a guard on the same instance.
+ - `submit_single_quotes` - Wether to use the batch endpoint for posting quotes to the api. This can be useful for debugging.
  - `chains` - each chain has a different config that overrides base_chain_config. Here are the parameters for each chain
   - `rfq_address` - the address of the rfq contract on this chain. These addresses are available [here](../Contracts.md).
 

ethergo/backends/anvil/anvil.go

@@ -80,8 +80,10 @@ func NewAnvilBackend(ctx context.Context, t *testing.T, args *OptionBuilder) *Ba
 
 	runOptions := &dockertest.RunOptions{
 		Repository: "ghcr.io/foundry-rs/foundry",
-		Tag:        "nightly-deb3116955eea4333f9e4e4516104be4182e9ee2",
-		Cmd:        []string{strings.Join(append([]string{"anvil"}, commandArgs...), " ")},
+		Tag:        "nightly-1bac1b3d79243cea755800bf396c30a3d74741bf",
+		Platform:   "linux/amd64",
+
+		Cmd: []string{strings.Join(append([]string{"anvil"}, commandArgs...), " ")},
 		Labels: map[string]string{
 			"test-id": uuid.New().String(),
 		},

ethergo/submitter/submitter.go

@@ -48,6 +48,8 @@ type TransactionSubmitter interface {
 	GetSubmissionStatus(ctx context.Context, chainID *big.Int, nonce uint64) (status SubmissionStatus, err error)
 	// Address returns the address of the signer.
 	Address() common.Address
+	// Started returns whether the submitter is running.
+	Started() bool
 }
 
 // txSubmitterImpl is the implementation of the transaction submitter.
@@ -83,6 +85,10 @@ type txSubmitterImpl struct {
 	// distinctChainIDs is the distinct chain ids for the transaction submitter.
 	// note: this map should not be appended to!
 	distinctChainIDs []*big.Int
+	// started indicates whether the submitter has started.
+	started bool
+	// startMux is the mutex for started.
+	startMux sync.RWMutex
 }
 
 // ClientFetcher is the interface for fetching a chain client.
@@ -107,6 +113,13 @@ func NewTransactionSubmitter(metrics metrics.Handler, signer signer.Signer, fetc
 	}
 }
 
+// Started returns whether the submitter is running.
+func (t *txSubmitterImpl) Started() bool {
+	t.startMux.RLock()
+	defer t.startMux.RUnlock()
+	return t.started
+}
+
 // GetRetryInterval returns the retry interval for the transaction submitter.
 func (t *txSubmitterImpl) GetRetryInterval() time.Duration {
 	retryInterval := time.Second * 2
@@ -126,9 +139,29 @@ func (t *txSubmitterImpl) GetDistinctInterval() time.Duration {
 	return retryInterval
 }
 
+// attemptMarkStarted attempts to mark the submitter as started.
+// if the submitter is already started, an error is returned.
+func (t *txSubmitterImpl) attemptMarkStarted() error {
+	t.startMux.Lock()
+	defer t.startMux.Unlock()
+	if t.started {
+		return ErrSubmitterAlreadyStarted
+	}
+	t.started = true
+	return nil
+}
+
+// ErrSubmitterAlreadyStarted is the error for when the submitter is already started.
+var ErrSubmitterAlreadyStarted = errors.New("submitter already started")
+
 // Start starts the transaction submitter.
 // nolint: cyclop
 func (t *txSubmitterImpl) Start(parentCtx context.Context) (err error) {
+	err = t.attemptMarkStarted()
+	if err != nil {
+		return err
+	}
+
 	t.otelRecorder, err = newOtelRecorder(t.metrics, t.signer)
 	if err != nil {
 		return fmt.Errorf("could not create otel recorder: %w", err)
@@ -313,6 +346,9 @@ func (t *txSubmitterImpl) triggerProcessQueue(ctx context.Context) {
 	}
 }
 
+// ErrNotStarted is the error for when the submitter is not started.
+var ErrNotStarted = errors.New("submitter is not started")
+
 // nolint: cyclop
 func (t *txSubmitterImpl) SubmitTransaction(parentCtx context.Context, chainID *big.Int, call ContractCallType) (nonce uint64, err error) {
 	ctx, span := t.metrics.Tracer().Start(parentCtx, "submitter.SubmitTransaction", trace.WithAttributes(
@@ -324,6 +360,10 @@ func (t *txSubmitterImpl) SubmitTransaction(parentCtx context.Context, chainID *
 		metrics.EndSpanWithErr(span, err)
 	}()
 
+	if !t.Started() {
+		logger.Errorf("%v in a future version, this will hard error", ErrNotStarted.Error())
+	}
+
 	// make sure we have a client for this chain.
 	chainClient, err := t.fetcher.GetClient(ctx, chainID)
 	if err != nil {

services/rfq/contracts/fastbridge/events.go

@@ -17,6 +17,8 @@ var (
 	BridgeProofProvidedTopic common.Hash
 	// BridgeDepositClaimedTopic is the topic emitted by a bridge relay.
 	BridgeDepositClaimedTopic common.Hash
+	// BridgeProofDisputedTopic is the topic emitted by a bridge dispute.
+	BridgeProofDisputedTopic common.Hash
 )
 
 // static checks to make sure topics actually exist.
@@ -32,6 +34,7 @@ func init() {
 	BridgeRelayedTopic = parsedABI.Events["BridgeRelayed"].ID
 	BridgeProofProvidedTopic = parsedABI.Events["BridgeProofProvided"].ID
 	BridgeDepositClaimedTopic = parsedABI.Events["BridgeDepositClaimed"].ID
+	BridgeProofDisputedTopic = parsedABI.Events["BridgeProofDisputed"].ID
 
 	_, err = parsedABI.EventByID(BridgeRequestedTopic)
 	if err != nil {
@@ -47,6 +50,11 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
+
+	_, err = parsedABI.EventByID(BridgeProofDisputedTopic)
+	if err != nil {
+		panic(err)
+	}
 }
 
 // topicMap maps events to topics.
@@ -57,6 +65,7 @@ func topicMap() map[EventType]common.Hash {
 		BridgeRelayedEvent:        BridgeRelayedTopic,
 		BridgeProofProvidedEvent:  BridgeProofProvidedTopic,
 		BridgeDepositClaimedEvent: BridgeDepositClaimedTopic,
+		BridgeDisputeEvent:        BridgeProofDisputedTopic,
 	}
 }
 

services/rfq/contracts/fastbridge/parser.go

@@ -20,6 +20,8 @@ const (
 	BridgeProofProvidedEvent
 	// BridgeDepositClaimedEvent is the event type for the BridgeDepositClaimed event.
 	BridgeDepositClaimedEvent
+	// BridgeDisputeEvent is the event type for the BridgeDispute event.
+	BridgeDisputeEvent
 )
 
 // Parser parses events from the fastbridge contracat.
@@ -82,6 +84,13 @@ func (p parserImpl) ParseEvent(log ethTypes.Log) (_ EventType, event interface{}
 			return noOpEvent, nil, false
 		}
 		return eventType, claimed, true
+	case BridgeDisputeEvent:
+		disputed, err := p.filterer.ParseBridgeProofDisputed(log)
+		if err != nil {
+			return noOpEvent, nil, false
+		}
+		return eventType, disputed, true
+
 	}
 
 	return eventType, nil, true

services/rfq/e2e/rfq_test.go

@@ -1,6 +1,7 @@
 package e2e_test
 
 import (
+	"fmt"
 	"math/big"
 	"testing"
 	"time"
@@ -19,6 +20,8 @@ import (
 	omnirpcClient "github.com/synapsecns/sanguine/services/omnirpc/client"
 	"github.com/synapsecns/sanguine/services/rfq/api/client"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/fastbridge"
+	"github.com/synapsecns/sanguine/services/rfq/guard/guarddb"
+	guardService "github.com/synapsecns/sanguine/services/rfq/guard/service"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/chain"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/reldb"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/service"
@@ -36,9 +39,12 @@ type IntegrationSuite struct {
 	omniClient    omnirpcClient.RPCClient
 	metrics       metrics.Handler
 	store         reldb.Service
+	guardStore    guarddb.Service
 	apiServer     string
 	relayer       *service.Relayer
+	guard         *guardService.Guard
 	relayerWallet wallet.Wallet
+	guardWallet   wallet.Wallet
 	userWallet    wallet.Wallet
 }
 
@@ -67,11 +73,6 @@ const (
 func (i *IntegrationSuite) SetupTest() {
 	i.TestSuite.SetupTest()
 
-	// TODO: no need for this when anvil CI issues are fixed
-	if core.GetEnvBool("CI", false) {
-		return
-	}
-
 	i.manager = testutil.NewDeployManager(i.T())
 	i.cctpDeployManager = cctpTest.NewDeployManager(i.T())
 	// TODO: consider jaeger
@@ -82,6 +83,7 @@ func (i *IntegrationSuite) SetupTest() {
 	// setup the api server
 	i.setupQuoterAPI()
 	i.setupRelayer()
+	i.setupGuard()
 }
 
 // getOtherBackend gets the backend that is not the current one. This is a helper
@@ -96,9 +98,13 @@ func (i *IntegrationSuite) getOtherBackend(backend backends.SimulatedTestBackend
 }
 
 func (i *IntegrationSuite) TestUSDCtoUSDC() {
-	if core.GetEnvBool("CI", false) {
-		i.T().Skip("skipping until anvil issues are fixed in CI")
-	}
+	// start the relayer and guard
+	go func() {
+		_ = i.relayer.Start(i.GetTestContext())
+	}()
+	go func() {
+		_ = i.guard.Start(i.GetTestContext())
+	}()
 
 	// load token contracts
 	const startAmount = 1000
@@ -240,13 +246,26 @@ func (i *IntegrationSuite) TestUSDCtoUSDC() {
 		i.NoError(err)
 		return len(originPendingRebals) > 0
 	})
+
+	i.Eventually(func() bool {
+		// verify that the guard has marked the tx as validated
+		results, err := i.guardStore.GetPendingProvensByStatus(i.GetTestContext(), guarddb.Validated)
+		i.NoError(err)
+		return len(results) == 1
+	})
 }
 
 // nolint: cyclop
 func (i *IntegrationSuite) TestETHtoETH() {
-	if core.GetEnvBool("CI", false) {
-		i.T().Skip("skipping until anvil issues are fixed in CI")
-	}
+
+	// start the relayer and guard
+	go func() {
+		_ = i.relayer.Start(i.GetTestContext())
+	}()
+	go func() {
+		_ = i.guard.Start(i.GetTestContext())
+	}()
+
 	// Send ETH to the relayer on destination
 	const initialBalance = 10
 	i.destBackend.FundAccount(i.GetTestContext(), i.relayerWallet.Address(), *big.NewInt(initialBalance))
@@ -347,4 +366,108 @@ func (i *IntegrationSuite) TestETHtoETH() {
 		}
 		return false
 	})
+
+	i.Eventually(func() bool {
+		// verify that the guard has marked the tx as validated
+		results, err := i.guardStore.GetPendingProvensByStatus(i.GetTestContext(), guarddb.Validated)
+		i.NoError(err)
+		return len(results) == 1
+	})
+}
+
+func (i *IntegrationSuite) TestDispute() {
+	// start the guard
+	go func() {
+		_ = i.guard.Start(i.GetTestContext())
+	}()
+
+	// load token contracts
+	const startAmount = 1000
+	const rfqAmount = 900
+	opts := i.destBackend.GetTxContext(i.GetTestContext(), nil)
+	destUSDC, destUSDCHandle := i.cctpDeployManager.GetMockMintBurnTokenType(i.GetTestContext(), i.destBackend)
+	realStartAmount, err := testutil.AdjustAmount(i.GetTestContext(), big.NewInt(startAmount), destUSDC.ContractHandle())
+	i.NoError(err)
+	realRFQAmount, err := testutil.AdjustAmount(i.GetTestContext(), big.NewInt(rfqAmount), destUSDC.ContractHandle())
+	i.NoError(err)
+
+	// add initial usdc to relayer on destination
+	tx, err := destUSDCHandle.MintPublic(opts.TransactOpts, i.relayerWallet.Address(), realStartAmount)
+	i.Nil(err)
+	i.destBackend.WaitForConfirmation(i.GetTestContext(), tx)
+	i.Approve(i.destBackend, destUSDC, i.relayerWallet)
+
+	// add initial USDC to relayer on origin
+	optsOrigin := i.originBackend.GetTxContext(i.GetTestContext(), nil)
+	originUSDC, originUSDCHandle := i.cctpDeployManager.GetMockMintBurnTokenType(i.GetTestContext(), i.originBackend)
+	tx, err = originUSDCHandle.MintPublic(optsOrigin.TransactOpts, i.relayerWallet.Address(), realStartAmount)
+	i.Nil(err)
+	i.originBackend.WaitForConfirmation(i.GetTestContext(), tx)
+	i.Approve(i.originBackend, originUSDC, i.relayerWallet)
+
+	// add initial USDC to user on origin
+	tx, err = originUSDCHandle.MintPublic(optsOrigin.TransactOpts, i.userWallet.Address(), realRFQAmount)
+	i.Nil(err)
+	i.originBackend.WaitForConfirmation(i.GetTestContext(), tx)
+	i.Approve(i.originBackend, originUSDC, i.userWallet)
+
+	// now we can send the money
+	_, originFastBridge := i.manager.GetFastBridge(i.GetTestContext(), i.originBackend)
+	auth := i.originBackend.GetTxContext(i.GetTestContext(), i.userWallet.AddressPtr())
+	// we want 499 usdc for 500 requested within a day
+	tx, err = originFastBridge.Bridge(auth.TransactOpts, fastbridge.IFastBridgeBridgeParams{
+		DstChainId:   uint32(i.destBackend.GetChainID()),
+		To:           i.userWallet.Address(),
+		OriginToken:  originUSDC.Address(),
+		SendChainGas: true,
+		DestToken:    destUSDC.Address(),
+		OriginAmount: realRFQAmount,
+		DestAmount:   new(big.Int).Sub(realRFQAmount, big.NewInt(10_000_000)),
+		Deadline:     new(big.Int).SetInt64(time.Now().Add(time.Hour * 24).Unix()),
+	})
+	i.NoError(err)
+	i.originBackend.WaitForConfirmation(i.GetTestContext(), tx)
+
+	// fetch the txid and raw request
+	var txID [32]byte
+	var rawRequest []byte
+	parser, err := fastbridge.NewParser(originFastBridge.Address())
+	i.NoError(err)
+	i.Eventually(func() bool {
+		receipt, err := i.originBackend.TransactionReceipt(i.GetTestContext(), tx.Hash())
+		i.NoError(err)
+		for _, log := range receipt.Logs {
+			_, parsedEvent, ok := parser.ParseEvent(*log)
+			if !ok {
+				continue
+			}
+			event, ok := parsedEvent.(*fastbridge.FastBridgeBridgeRequested)
+			if ok {
+				rawRequest = event.Request
+				txID = event.TransactionId
+				return true
+			}
+		}
+		return false
+	})
+
+	// call prove() from the relayer wallet before relay actually occurred on dest
+	relayerAuth := i.originBackend.GetTxContext(i.GetTestContext(), i.relayerWallet.AddressPtr())
+	fakeHash := common.HexToHash("0xdeadbeef")
+	tx, err = originFastBridge.Prove(relayerAuth.TransactOpts, rawRequest, fakeHash)
+	i.NoError(err)
+	i.originBackend.WaitForConfirmation(i.GetTestContext(), tx)
+
+	// verify that the guard calls Dispute()
+	i.Eventually(func() bool {
+		results, err := i.guardStore.GetPendingProvensByStatus(i.GetTestContext(), guarddb.Disputed)
+		i.NoError(err)
+		if len(results) != 1 {
+			return false
+		}
+		fmt.Printf("GOT RESULTS: %v\n", results)
+		result, err := i.guardStore.GetPendingProvenByID(i.GetTestContext(), txID)
+		i.NoError(err)
+		return result.TxHash == fakeHash && result.Status == guarddb.Disputed && result.TransactionID == txID
+	})
 }