Add CCTP rebalancing to RFQ relayer

services/rfq/e2e/setup_test.go

@@ -197,7 +197,7 @@ func (i *IntegrationSuite) setupRelayer() {
 		// generated ex-post facto
 		Chains: map[int]relconfig.ChainConfig{
 			originBackendChainID: {
-				Bridge:        i.manager.Get(i.GetTestContext(), i.originBackend, testutil.FastBridgeType).Address().String(),
+				RFQAddress:    i.manager.Get(i.GetTestContext(), i.originBackend, testutil.FastBridgeType).Address().String(),
 				Confirmations: 0,
 				Tokens: map[string]relconfig.TokenConfig{
 					"ETH": {
@@ -209,7 +209,7 @@ func (i *IntegrationSuite) setupRelayer() {
 				NativeToken: "ETH",
 			},
 			destBackendChainID: {
-				Bridge:        i.manager.Get(i.GetTestContext(), i.destBackend, testutil.FastBridgeType).Address().String(),
+				RFQAddress:    i.manager.Get(i.GetTestContext(), i.destBackend, testutil.FastBridgeType).Address().String(),
 				Confirmations: 0,
 				Tokens: map[string]relconfig.TokenConfig{
 					"ETH": {

services/rfq/relayer/inventory/manager.go

@@ -18,6 +18,7 @@
 	"github.com/synapsecns/sanguine/core"
 	"github.com/synapsecns/sanguine/core/metrics"
 	"github.com/synapsecns/sanguine/ethergo/submitter"
+	"github.com/synapsecns/sanguine/services/cctp-relayer/contracts/cctp"
 	"github.com/synapsecns/sanguine/services/rfq/contracts/ierc20"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/chain"
 	"github.com/synapsecns/sanguine/services/rfq/relayer/relconfig"
@@ -39,9 +40,12 @@
 	// GetCommittableBalances gets the total balances committable for all tracked tokens.
 	GetCommittableBalances(ctx context.Context, options ...BalanceFetchArgOption) (map[int]map[common.Address]*big.Int, error)
 	// ApproveAllTokens approves all tokens for the relayer address.
-	ApproveAllTokens(ctx context.Context, submitter submitter.TransactionSubmitter) error
+	ApproveAllTokens(ctx context.Context) error
 	// HasSufficientGas checks if there is sufficient gas for a given route.
 	HasSufficientGas(ctx context.Context, origin, dest int) (bool, error)
+	// Rebalance checks whether a given token should be rebalanced, and
+	// executes the rebalance if necessary.
+	Rebalance(ctx context.Context, chainID int, token common.Address) error
 }
 
 type inventoryManagerImpl struct {
@@ -59,7 +63,12 @@
 	relayerAddress common.Address
 	// chainClient is an omnirpc client
 	chainClient submitter.ClientFetcher
-	db          reldb.Service
+	// txSubmitter is the transaction submitter
+	txSubmitter submitter.TransactionSubmitter
+	// cctpContracts is the map of cctp contracts (used for rebalancing)
+	cctpContracts map[int]*cctp.SynapseCCTP
+	// db is the database
+	db reldb.Service
 }
 
 // GetCommittableBalance gets the committable balances.
@@ -122,6 +131,8 @@
 	decimals       uint8
 	startAllowance *big.Int
 	isGasToken     bool
+	chainID        int
+	addr           common.Address
 }
 
 var (
@@ -135,12 +146,15 @@
 const defaultPollPeriod = 5
 
 // NewInventoryManager creates a list of tokens we should use.
-func NewInventoryManager(ctx context.Context, clientFetcher submitter.ClientFetcher, handler metrics.Handler, cfg relconfig.Config, relayer common.Address, db reldb.Service) (Manager, error) {
+// TODO: too many args here
+func NewInventoryManager(ctx context.Context, clientFetcher submitter.ClientFetcher, handler metrics.Handler, cfg relconfig.Config, relayer common.Address, txSubmitter submitter.TransactionSubmitter, db reldb.Service) (Manager, error) {
 	i := inventoryManagerImpl{
 		relayerAddress: relayer,
 		handler:        handler,
 		cfg:            cfg,
 		chainClient:    clientFetcher,
+		txSubmitter:    txSubmitter,
+		cctpContracts:  make(map[int]*cctp.SynapseCCTP),
 		db:             db,
 	}
 
@@ -173,7 +187,7 @@
 const maxBatchSize = 10
 
 // ApproveAllTokens approves all checks if allowance is set and if not approves.
-func (i *inventoryManagerImpl) ApproveAllTokens(ctx context.Context, submitter submitter.TransactionSubmitter) error {
+func (i *inventoryManagerImpl) ApproveAllTokens(ctx context.Context) error {
 	i.mux.RLock()
 	defer i.mux.RUnlock()
 
@@ -185,25 +199,46 @@
 
 		for address, token := range tokenMap {
 			// if startAllowance is 0
 			if address != chain.EthAddress && token.startAllowance.Cmp(big.NewInt(0)) == 0 {
 				chainID := chainID // capture func literal
 				address := address // capture func literal
-				// init an approval in submitter. Note: in the case where submitter hasn't finished from last boot, this will double submit approvals unfortanutely
-				_, err = submitter.SubmitTransaction(ctx, big.NewInt(int64(chainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) {
-					erc20, err := ierc20.NewIERC20(address, backendClient)
+
+				erc20, err := ierc20.NewIERC20(address, backendClient)
+				if err != nil {
+					return fmt.Errorf("could not get erc20: %w", err)
+				}
+
+				// init an approval for RFQ bridge in submitter. Note: in the case where submitter hasn't finished from last boot, this will double submit approvals unfortanutely
+				_, err = i.txSubmitter.SubmitTransaction(ctx, big.NewInt(int64(chainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) {
+					rfqAddr, err := i.cfg.GetRFQAddress(chainID)
 					if err != nil {
-						return nil, fmt.Errorf("could not get erc20: %w", err)
+						return nil, fmt.Errorf("could not get rfq address: %w", err)
 					}
+					tx, err = erc20.Approve(transactor, common.HexToAddress(rfqAddr), abi.MaxInt256)
+					if err != nil {
+						return nil, fmt.Errorf("could not approve rfq: %w", err)
+					}
+					return tx, nil
+				})
+				if err != nil {
+					return fmt.Errorf("could not submit RFQ approval: %w", err)
+				}
 
-					approveAmount, err := erc20.Approve(transactor, common.HexToAddress(i.cfg.Chains[chainID].Bridge), abi.MaxInt256)
+				// approve CCTP bridge, if configured
+				_, err = i.txSubmitter.SubmitTransaction(ctx, big.NewInt(int64(chainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) {
+					cctpAddr, err := i.cfg.GetCCTPAddress(chainID)
 					if err != nil {
-						return nil, fmt.Errorf("could not approve: %w", err)
+						return nil, fmt.Errorf("could not get cctp address: %w", err)
+					}
+					tx, err = erc20.Approve(transactor, common.HexToAddress(cctpAddr), abi.MaxInt256)
+					if err != nil {
+						return nil, fmt.Errorf("could not approve cctp: %w", err)
 					}
 
-					return approveAmount, nil
+					return tx, nil
 				})
 				if err != nil {
-					return fmt.Errorf("could not submit approval: %w", err)
+					return fmt.Errorf("could not submit CCTP approval: %w", err)
 				}
 			}
 		}
@@ -230,6 +265,151 @@
 	return sufficient, nil
 }
 
+// Rebalance checks whether a given token should be rebalanced, and executes the rebalance if necessary.
+// Note that if there are multiple tokens whose balance is below the maintenance balance, only the lowest balance
+// will be rebalanced.
+func (i *inventoryManagerImpl) Rebalance(ctx context.Context, chainID int, token common.Address) error {
+	method, err := i.cfg.GetRebalanceMethod(chainID, token.Hex())
+	if err != nil {
+		return fmt.Errorf("could not get rebalance method: %w", err)
+	}
+	if method == relconfig.RebalanceMethodNone {
+		return nil
+	}
+
+	err = i.refreshBalances(ctx)
+	if err != nil {
+		return fmt.Errorf("could not refresh balances: %w", err)
+	}
+
+	rebalance, err := i.getRebalance(ctx, chainID, token)
+	if err != nil {
+		return fmt.Errorf("could not get rebalance: %w", err)
+	}
+	if rebalance == nil {
+		return nil
+	}
+
+	switch method {
+	case relconfig.RebalanceMethodCCTP:
+		return i.rebalanceCCTP(ctx, rebalance)
+	default:
+		return fmt.Errorf("unknown rebalance method: %s", method)
+	}
+}
+
+// rebalanceData contains metadata for a rebalance action.
+type rebalanceData struct {
+	origin         int
+	dest           int
+	originMetadata *tokenMetadata
+	destMetadata   *tokenMetadata
+	amount         *big.Int
+}
+
+func (i *inventoryManagerImpl) getRebalance(ctx context.Context, chainID int, token common.Address) (rebalance *rebalanceData, err error) {
+	maintenancePct, err := i.cfg.GetMaintenanceBalancePct(chainID, token.Hex())
+	if err != nil {
+		return nil, fmt.Errorf("could not get maintenance pct: %w", err)
+	}
+
+	// get token metadata
+	var rebalanceTokenData *tokenMetadata
+	for address, tokenData := range i.tokens[chainID] {
+		if address == token {
+			rebalanceTokenData = tokenData
+			break
+		}
+	}
+
+	// get total balance for given token across all chains
+	totalBalance := big.NewInt(0)
+	for _, tokenMap := range i.tokens {
+		for _, tokenData := range tokenMap {
+			if tokenData.name == rebalanceTokenData.name {
+				totalBalance.Add(totalBalance, tokenData.balance)
+			}
+		}
+	}
+
+	// check if any balances are below maintenance threshold
+	var minTokenData, maxTokenData *tokenMetadata
+	for _, tokenMap := range i.tokens {
+		for _, tokenData := range tokenMap {
+			if tokenData.name == rebalanceTokenData.name {
+				if minTokenData == nil || minTokenData.balance.Cmp(minTokenData.balance) < 0 {
+					minTokenData = tokenData
+				}
+				if maxTokenData == nil || maxTokenData.balance.Cmp(maxTokenData.balance) > 0 {
+					maxTokenData = tokenData
+				}
+			}
+		}
+	}
+
+	// get the initialPct for the origin chain
+	initialPct, err := i.cfg.GetInitialBalancePct(maxTokenData.chainID, maxTokenData.addr.Hex())
+	if err != nil {
+		return nil, fmt.Errorf("could not get initial pct: %w", err)
+	}
+
+	// check if the minimum balance is below the threshold and trigger rebalance
+	maintenanceThresh, _ := new(big.Float).Mul(new(big.Float).SetInt(totalBalance), big.NewFloat(maintenancePct)).Int(nil)
+	if minTokenData.balance.Cmp(maintenanceThresh) < 0 {
+		initialThresh, _ := new(big.Float).Mul(new(big.Float).SetInt(totalBalance), big.NewFloat(initialPct)).Int(nil)
+		amount := new(big.Int).Sub(maxTokenData.balance, initialThresh)
+		rebalance = &rebalanceData{
+			origin:         maxTokenData.chainID,
+			dest:           minTokenData.chainID,
+			originMetadata: maxTokenData,
+			destMetadata:   minTokenData,
+			amount:         amount,
+		}
+	}
+	return rebalance, nil
+}
+
+func (i *inventoryManagerImpl) rebalanceCCTP(ctx context.Context, rebalance *rebalanceData) (err error) {
+	// fetch the corresponding CCTP contract
+	contract, ok := i.cctpContracts[rebalance.dest]
+	if !ok {
+		contractAddr, err := i.cfg.GetCCTPAddress(rebalance.origin)
+		if err != nil {
+			return fmt.Errorf("could not get cctp address: %w", err)
+		}
+		chainClient, err := i.chainClient.GetClient(ctx, big.NewInt(int64(rebalance.origin)))
+		if err != nil {
+			return fmt.Errorf("could not get chain client: %w", err)
+		}
+		contract, err = cctp.NewSynapseCCTP(common.HexToAddress(contractAddr), chainClient)
+		if err != nil {
+			return fmt.Errorf("could not get cctp: %w", err)
+		}
+		i.cctpContracts[rebalance.dest] = contract
+	}
+
+	// perform rebalance by calling sendCircleToken()
+	_, err = i.txSubmitter.SubmitTransaction(ctx, big.NewInt(int64(rebalance.originMetadata.chainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) {
+		tx, err = contract.SendCircleToken(
+			transactor,
+			i.relayerAddress,
+			big.NewInt(int64(rebalance.destMetadata.chainID)),
+			rebalance.originMetadata.addr,
+			rebalance.amount,
+			0,        // TODO: inspect
+			[]byte{}, // TODO: inspect
+		)
+		if err != nil {
+			return nil, fmt.Errorf("could not send circle token: %w", err)
+		}
+		return tx, nil
+	})
+	if err != nil {
+		return fmt.Errorf("could not submit CCTP rebalance: %w", err)
+	}
+	return nil
+}
+
 // initializes tokens converts the configuration into a data structure we can use to determine inventory
 // it gets metadata like name, decimals, etc once and exports these to prometheus for ease of debugging.
 func (i *inventoryManagerImpl) initializeTokens(parentCtx context.Context, cfg relconfig.Config) (err error) {
@@ -274,6 +454,7 @@
 			}
 			rtoken := &tokenMetadata{
 				isGasToken: tokenName == nativeToken,
+				chainID:    chainID,
 			}
 
 			var token common.Address
@@ -283,6 +464,7 @@
 				token = common.HexToAddress(tokenCfg.Address)
 			}
 			i.tokens[chainID][token] = rtoken
+			rtoken.addr = token
 
 			// requires non-nil pointer
 			rtoken.balance = new(big.Int)
@@ -294,11 +476,15 @@
 				rtoken.balance = i.gasBalances[chainID]
 				// TODO: start allowance?
 			} else {
+				rfqAddr, err := cfg.GetRFQAddress(chainID)
+				if err != nil {
+					return fmt.Errorf("could not get rfq address: %w", err)
+				}
 				deferredCalls[chainID] = append(deferredCalls[chainID],
 					eth.CallFunc(funcBalanceOf, token, i.relayerAddress).Returns(rtoken.balance),
 					eth.CallFunc(funcDecimals, token).Returns(&rtoken.decimals),
 					eth.CallFunc(funcName, token).Returns(&rtoken.name),
-					eth.CallFunc(funcAllowance, token, i.relayerAddress, common.HexToAddress(i.cfg.Chains[chainID].Bridge)).Returns(rtoken.startAllowance),
+					eth.CallFunc(funcAllowance, token, i.relayerAddress, common.HexToAddress(rfqAddr)).Returns(rtoken.startAllowance),
 				)
 			}
 

services/rfq/relayer/quoter/quoter.go

@@ -238,10 +238,9 @@ func (m *Manager) prepareAndSubmitQuotes(ctx context.Context, inv map[int]map[co
 // We can do this by looking at the quotableTokens map, and finding the key that matches the destination chain token.
 // Generates quotes for a given chain ID, address, and balance.
 func (m *Manager) generateQuotes(ctx context.Context, chainID int, address common.Address, balance *big.Int) ([]model.PutQuoteRequest, error) {
-
-	destChainCfg, ok := m.config.Chains[chainID]
-	if !ok {
-		return nil, fmt.Errorf("error getting chain config for destination chain ID %d", chainID)
+	destRFQAddr, err := m.config.GetRFQAddress(chainID)
+	if err != nil {
+		return nil, fmt.Errorf("error getting destination RFQ address: %w", err)
 	}
 
 	destTokenID := fmt.Sprintf("%d-%s", chainID, address.Hex())
@@ -277,9 +276,9 @@ func (m *Manager) generateQuotes(ctx context.Context, chainID int, address commo
 				if err != nil {
 					return nil, fmt.Errorf("error getting total fee: %w", err)
 				}
-				originChainCfg, ok := m.config.Chains[origin]
-				if !ok {
-					return nil, fmt.Errorf("error getting chain config for origin chain ID %d", origin)
+				originRFQAddr, err := m.config.GetRFQAddress(origin)
+				if err != nil {
+					return nil, fmt.Errorf("error getting RFQ address: %w", err)
 				}
 
 				// Build the quote
@@ -295,8 +294,8 @@ func (m *Manager) generateQuotes(ctx context.Context, chainID int, address commo
 					DestAmount:              destAmount.String(),
 					MaxOriginAmount:         quoteAmount.String(),
 					FixedFee:                fee.String(),
-					OriginFastBridgeAddress: originChainCfg.Bridge,
-					DestFastBridgeAddress:   destChainCfg.Bridge,
+					OriginFastBridgeAddress: originRFQAddr,
+					DestFastBridgeAddress:   destRFQAddr,
 				}
 				quotes = append(quotes, quote)
 			}

services/rfq/relayer/relapi/server.go

@@ -59,11 +59,11 @@ func NewRelayerAPI(
 		if err != nil {
 			return nil, fmt.Errorf("could not create omnirpc client: %w", err)
 		}
-		chainListener, err := listener.NewChainListener(chainClient, store, common.HexToAddress(chainCfg.Bridge), handler)
+		chainListener, err := listener.NewChainListener(chainClient, store, common.HexToAddress(chainCfg.RFQAddress), handler)
 		if err != nil {
 			return nil, fmt.Errorf("could not get chain listener: %w", err)
 		}
-		chains[uint32(chainID)], err = chain.NewChain(ctx, chainClient, common.HexToAddress(chainCfg.Bridge), chainListener, submitter)
+		chains[uint32(chainID)], err = chain.NewChain(ctx, chainClient, common.HexToAddress(chainCfg.RFQAddress), chainListener, submitter)
 		if err != nil {
 			return nil, fmt.Errorf("could not create chain: %w", err)
 		}

services/rfq/relayer/relapi/suite_test.go

@@ -79,10 +79,10 @@ func (c *RelayerServerSuite) SetupTest() {
 	testConfig := relconfig.Config{
 		Chains: map[int]relconfig.ChainConfig{
 			int(c.originChainID): {
-				Bridge: ethFastBridgeAddress.Hex(),
+				RFQAddress: ethFastBridgeAddress.Hex(),
 			},
 			int(c.destChainID): {
-				Bridge: arbFastBridgeAddress.Hex(),
+				RFQAddress: arbFastBridgeAddress.Hex(),
 			},
 		},
 		RelayerAPIPort: strconv.Itoa(port),