Add CCTP rebalancing to RFQ relayer #2073
Conversation
WalkthroughThe recent changes focus on enhancing the inventory management system for a Request for Quote (RFQ) service. Key updates include the initiation of the inventory manager, improved token approval processes, and the introduction of a rebalancing method for token management. Additionally, configurations for RFQ and CCTP bridges have been refined, with specific adjustments made to rebalancing settings. These updates aim to streamline operations and ensure efficient token management across different blockchain networks. Changes
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (invoked as PR comments)
Additionally, you can add CodeRabbit Configration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review Status
Actionable comments generated: 8
Configuration used: .coderabbit.yaml
Files selected for processing (14)
- services/rfq/e2e/setup_test.go (2 hunks)
- services/rfq/relayer/inventory/manager.go (11 hunks)
- services/rfq/relayer/quoter/quoter.go (3 hunks)
- services/rfq/relayer/relapi/server.go (1 hunks)
- services/rfq/relayer/relapi/suite_test.go (1 hunks)
- services/rfq/relayer/relconfig/config.go (2 hunks)
- services/rfq/relayer/relconfig/config_test.go (4 hunks)
- services/rfq/relayer/relconfig/enum.go (1 hunks)
- services/rfq/relayer/relconfig/getters.go (2 hunks)
- services/rfq/relayer/relconfig/rebalancemethod_string.go (1 hunks)
- services/rfq/relayer/service/handlers.go (2 hunks)
- services/rfq/relayer/service/relayer.go (4 hunks)
- services/rfq/relayer/service/statushandler.go (4 hunks)
- services/rfq/relayer/service/suite_test.go (1 hunks)
Files skipped from review due to trivial changes (2)
- services/rfq/relayer/relconfig/enum.go
- services/rfq/relayer/relconfig/rebalancemethod_string.go
Additional comments: 26
services/rfq/relayer/relapi/server.go (2)
- 62-62: The change from
BridgetoRFQAddressin thechainCfgstruct is correctly reflected here when initializing thechainListener. This update aligns with the PR objectives to enhance the RFQ relayer's functionality.- 66-66: Similarly, the update from
BridgetoRFQAddressis correctly applied here for the creation of a newchaininstance. This change is consistent with the overall goal of the PR and correctly implements the new configuration structure.services/rfq/relayer/service/suite_test.go (2)
- 72-72: The renaming of the
Bridgefield toRFQAddressfor theoriginBackendchain configuration is correctly implemented. This change is in line with the PR's objectives to update configuration structures to support new functionality.- 75-75: The renaming of the
Bridgefield toRFQAddressfor thedestBackendchain configuration is also correctly implemented. This ensures consistency across the codebase and aligns with the PR's objectives.services/rfq/relayer/service/statushandler.go (2)
- 41-42: The renaming of
RelayerAdresstoRelayerAddresscorrects a typo and improves code readability. This change enhances consistency across the codebase.- 133-137: The modification in the chain initialization logic to retrieve the RFQ address from the configuration is correctly implemented. This change aligns with the PR's objectives to utilize the updated
RFQAddressfield in theChainConfigstruct.services/rfq/relayer/relconfig/config.go (1)
- 52-55: The renaming of the
Bridgefield toRFQAddressand the addition of theCCTPAddressfield in theChainConfigstruct are correctly implemented. These changes support the new functionality introduced in the PR and align with the objectives to enhance the RFQ relayer's functionality.services/rfq/relayer/relapi/suite_test.go (2)
- 82-82: The renaming of the
Bridgefield toRFQAddressfor theoriginChainIDconfiguration is correctly implemented. This change aligns with the PR's objectives and ensures consistency in the test suite setup.- 85-85: The renaming of the
Bridgefield toRFQAddressfor thedestChainIDconfiguration is also correctly implemented. This ensures that the test suite accurately reflects the changes made in the main codebase.services/rfq/relayer/service/relayer.go (3)
- 65-75: The error handling for getting the RFQ address and chain client has been improved. This ensures that the relayer fails early if it cannot obtain necessary configuration details, enhancing the robustness of the initialization process.
- 87-89: Refactoring the initialization of
im(Inventory Manager) by directly passing theTransactionSubmittertoNewInventoryManagersimplifies the setup process and reduces redundancy. This change improves code readability and maintainability.- 136-136: The modification to call
ApproveAllTokenswithout passing thesubmittersimplifies the token approval process. This change likely reflects an internal update to howApproveAllTokensmanages submissions, potentially leading to cleaner and more encapsulated logic within theinventory.Managerimplementation.services/rfq/relayer/relconfig/config_test.go (2)
- 17-18: The renaming of the
Bridgefield toRFQAddressand the addition ofCCTPAddressin theChainConfigstruct are correctly reflected in the test setup. This change aligns with the PR's objective to enhance RFQ relayer functionality and support inventory rebalancing through the CCTP bridge.Also applies to: 34-35, 53-54
- 71-96: The test cases for
GetRFQAddressandGetCCTPAddresshave been correctly updated to use the new fields. These tests ensure that the configuration getters return the expected values for RFQ and CCTP addresses, which is crucial for the correct operation of the relayer in different chain environments.services/rfq/e2e/setup_test.go (1)
- 200-200: The update to use
RFQAddressinstead of theBridgefield in thesetupRelayerfunction of theIntegrationSuitecorrectly reflects the changes made in the PR. This ensures that the end-to-end tests are aligned with the new configuration structure and can accurately test the relayer's functionality with the correct RFQ addresses.Also applies to: 212-212
services/rfq/relayer/relconfig/getters.go (4)
- 90-102: The
GetRFQAddressfunction has been correctly updated to retrieve the RFQ address for a given chain ID. The error handling and type assertion are properly implemented.- 104-114: The
GetCCTPAddressfunction is correctly implemented to retrieve the CCTP address for a given chain ID, with appropriate error handling and type assertion.- 382-396: The
GetMaintenanceBalancePctfunction correctly retrieves the maintenance balance percentage for a given chain and token. The error handling and validation for a positive percentage value are properly implemented.- 398-412: The
GetInitialBalancePctfunction correctly retrieves the initial balance percentage for a given chain and token. The error handling and validation for a positive percentage value are properly implemented.services/rfq/relayer/quoter/quoter.go (3)
- 241-243: The refactoring to use
GetRFQAddressfor obtaining the destination RFQ address improves encapsulation and error handling. This change ensures that the method for retrieving RFQ addresses is consistent and centralized, which is beneficial for maintainability and potential future changes to how addresses are managed.- 279-281: Similar to the destination RFQ address, using
GetRFQAddressfor obtaining the origin RFQ address maintains consistency and improves error handling. This approach centralizes the retrieval of RFQ addresses, making the codebase more maintainable and adaptable to changes.- 297-298: The assignment of
OriginFastBridgeAddressandDestFastBridgeAddressusing the obtained RFQ addresses in the quote object is a logical step following the retrieval of these addresses. This ensures that the quotes are correctly associated with their respective bridge addresses, which is crucial for the RFQ process. It's good to see that the changes are consistently applied throughout the quote generation logic.services/rfq/relayer/inventory/manager.go (4)
- 21-21: The addition of the
cctpcontract import is necessary for the new rebalancing functionality. Ensure that thecctppackage is correctly implemented and tested, as it plays a crucial role in the rebalancing process.- 66-71: The addition of
txSubmitter,cctpContracts, andchainIDfields toinventoryManagerImplsupports the new rebalancing functionality. Ensure that these fields are properly initialized and used securely, especially in multi-threaded contexts given the presence of mutex locks in the implementation.- 134-135: The addition of
chainIDandaddrfields totokenMetadatastruct is necessary for the rebalancing logic. It's important to ensure that these fields are correctly populated during token initialization to avoid issues in rebalancing operations.- 268-299: The
Rebalancemethod's implementation is central to the new functionality. It's important to ensure that the rebalancing logic is correct and efficient. Specifically, the switch-case handling different rebalancing methods should be extensible for future methods. Also, validate the correctness of the rebalance decision logic and the error handling throughout the method.
| // handleClaimCompleted handles the claim completed status and marks the claim as completed. | ||
| // Step 9: ClaimCompleted | ||
| // | ||
| // Since this marks the completion of a RFQ bridge sequence, we check if a rebalance for the given token | ||
| // is needed, and trigger it on the inventory manager if so. | ||
| func (q *QuoteRequestHandler) handleClaimCompleted(ctx context.Context, _ trace.Span, request reldb.QuoteRequest) (err error) { | ||
| err = q.Inventory.Rebalance(ctx, int(request.Transaction.DestChainId), request.Transaction.DestToken) | ||
| if err != nil { | ||
| return fmt.Errorf("could not rebalance: %w", err) | ||
| } | ||
|
|
||
| err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.ClaimPending) | ||
| if err != nil { | ||
| return fmt.Errorf("could not update request status: %w", err) | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
The handleClaimCompleted function correctly implements the logic to trigger rebalancing upon a claim completion event. However, there's a potential issue with the final status update to reldb.ClaimPending after rebalancing. This status might not accurately reflect the completion of the claim process. Consider updating the status to reldb.ClaimCompleted or a similar status that accurately represents the completion of the claim process.
- err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.ClaimPending)
+ err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.ClaimCompleted)Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| // handleClaimCompleted handles the claim completed status and marks the claim as completed. | |
| // Step 9: ClaimCompleted | |
| // | |
| // Since this marks the completion of a RFQ bridge sequence, we check if a rebalance for the given token | |
| // is needed, and trigger it on the inventory manager if so. | |
| func (q *QuoteRequestHandler) handleClaimCompleted(ctx context.Context, _ trace.Span, request reldb.QuoteRequest) (err error) { | |
| err = q.Inventory.Rebalance(ctx, int(request.Transaction.DestChainId), request.Transaction.DestToken) | |
| if err != nil { | |
| return fmt.Errorf("could not rebalance: %w", err) | |
| } | |
| err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.ClaimPending) | |
| if err != nil { | |
| return fmt.Errorf("could not update request status: %w", err) | |
| } | |
| return nil | |
| } | |
| // handleClaimCompleted handles the claim completed status and marks the claim as completed. | |
| // Step 9: ClaimCompleted | |
| // | |
| // Since this marks the completion of a RFQ bridge sequence, we check if a rebalance for the given token | |
| // is needed, and trigger it on the inventory manager if so. | |
| func (q *QuoteRequestHandler) handleClaimCompleted(ctx context.Context, _ trace.Span, request reldb.QuoteRequest) (err error) { | |
| err = q.Inventory.Rebalance(ctx, int(request.Transaction.DestChainId), request.Transaction.DestToken) | |
| if err != nil { | |
| return fmt.Errorf("could not rebalance: %w", err) | |
| } | |
| err = q.db.UpdateQuoteRequestStatus(ctx, request.TransactionID, reldb.ClaimCompleted) | |
| if err != nil { | |
| return fmt.Errorf("could not update request status: %w", err) | |
| } | |
| return nil | |
| } |
| // GetRebalanceMethod returns the rebalance method for the given chain and token. | ||
| func (c Config) GetRebalanceMethod(chainID int, token string) (method RebalanceMethod, err error) { | ||
| chainConfig, ok := c.Chains[chainID] | ||
| if !ok { | ||
| return method, fmt.Errorf("no chain config for chain %d", chainID) | ||
| } | ||
| tokenConfig, ok := chainConfig.Tokens[token] | ||
| if !ok { | ||
| return method, fmt.Errorf("no token config for chain %d and token %s", chainID, token) | ||
| } | ||
| switch tokenConfig.RebalanceMethod { | ||
| case "cctp": | ||
| return RebalanceMethodCCTP, nil | ||
| case "native": | ||
| return RebalanceMethodNative, nil | ||
| } | ||
| return RebalanceMethodNone, nil | ||
| } |
There was a problem hiding this comment.
The GetRebalanceMethod function correctly retrieves the rebalance method for a given chain and token. However, consider handling the case where the RebalanceMethod is not recognized (other than "cctp" or "native") more explicitly by returning an error indicating an unsupported rebalance method.
switch tokenConfig.RebalanceMethod {
case "cctp":
return RebalanceMethodCCTP, nil
case "native":
return RebalanceMethodNative, nil
+ default:
+ return RebalanceMethodNone, fmt.Errorf("unsupported rebalance method: %s", tokenConfig.RebalanceMethod)
}Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| // GetRebalanceMethod returns the rebalance method for the given chain and token. | |
| func (c Config) GetRebalanceMethod(chainID int, token string) (method RebalanceMethod, err error) { | |
| chainConfig, ok := c.Chains[chainID] | |
| if !ok { | |
| return method, fmt.Errorf("no chain config for chain %d", chainID) | |
| } | |
| tokenConfig, ok := chainConfig.Tokens[token] | |
| if !ok { | |
| return method, fmt.Errorf("no token config for chain %d and token %s", chainID, token) | |
| } | |
| switch tokenConfig.RebalanceMethod { | |
| case "cctp": | |
| return RebalanceMethodCCTP, nil | |
| case "native": | |
| return RebalanceMethodNative, nil | |
| } | |
| return RebalanceMethodNone, nil | |
| } | |
| // GetRebalanceMethod returns the rebalance method for the given chain and token. | |
| func (c Config) GetRebalanceMethod(chainID int, token string) (method RebalanceMethod, err error) { | |
| chainConfig, ok := c.Chains[chainID] | |
| if !ok { | |
| return method, fmt.Errorf("no chain config for chain %d", chainID) | |
| } | |
| tokenConfig, ok := chainConfig.Tokens[token] | |
| if !ok { | |
| return method, fmt.Errorf("no token config for chain %d and token %s", chainID, token) | |
| } | |
| switch tokenConfig.RebalanceMethod { | |
| case "cctp": | |
| return RebalanceMethodCCTP, nil | |
| case "native": | |
| return RebalanceMethodNative, nil | |
| default: | |
| return RebalanceMethodNone, fmt.Errorf("unsupported rebalance method: %s", tokenConfig.RebalanceMethod) | |
| } | |
| return RebalanceMethodNone, nil | |
| } |
| @@ -76,7 +76,7 @@ func (r *Relayer) requestToHandler(ctx context.Context, req reldb.QuoteRequest) | |||
| // no more need for deadline middleware now, we already relayed. | |||
| qr.handlers[reldb.RelayCompleted] = r.gasMiddleware(qr.handleRelayCompleted) | |||
| qr.handlers[reldb.ProvePosted] = qr.handleProofPosted | |||
| // TODO: we probably want a claim complete state once we've seen that event on chain | |||
| qr.handlers[reldb.ClaimCompleted] = qr.handleClaimCompleted | |||
There was a problem hiding this comment.
The addition of a handler for the ClaimCompleted state is a significant enhancement. It's important to ensure that this new handler is thoroughly tested, especially since it's part of the quote request handling logic which is critical to the application's functionality.
Would you like me to help with writing tests for the new handleClaimCompleted method?
| // RebalanceMethod is the method to use for rebalancing. | ||
| RebalanceMethod string `yaml:"rebalance_method"` | ||
| // MaintenanceBalancePct is the percentage of the total balance under which a rebalance will be triggered. | ||
| MaintenanceBalancePct float64 `yaml:"maintenance_balance_pct"` | ||
| // InitialBalancePct is the percentage of the total balance to retain when triggering a rebalance. | ||
| InitialBalancePct float64 `yaml:"initial_balance_pct"` |
There was a problem hiding this comment.
The addition of RebalanceMethod, MaintenanceBalancePct, and InitialBalancePct fields in the TokenConfig struct for rebalancing settings is a significant enhancement. It's important to ensure that these new settings are validated to prevent configuration errors that could impact the rebalancing functionality.
Would you like me to help with implementing validation for these new configuration settings?
| 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 |
There was a problem hiding this comment.
The rebalanceCCTP method performs the actual rebalancing operation using the CCTP bridge. Ensure that the contract interaction logic is secure and correctly handles potential errors. The use of placeholders (0 and []byte{}) in the SendCircleToken call should be reviewed and replaced with actual values if necessary.
Consider replacing the placeholders in the SendCircleToken call with actual values or configurations, as these might impact the functionality of the rebalancing operation.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #2073 +/- ##
===================================================
- Coverage 48.80073% 48.07063% -0.73011%
===================================================
Files 370 372 +2
Lines 27225 27807 +582
Branches 83 83
===================================================
+ Hits 13286 13367 +81
- Misses 12554 13039 +485
- Partials 1385 1401 +16
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Configuration used: .coderabbit.yaml
Files selected for processing (1)
- services/rfq/relayer/relconfig/rebalancemethod_string.go (1 hunks)
Files skipped from review as they are similar to previous changes (1)
- services/rfq/relayer/relconfig/rebalancemethod_string.go
There was a problem hiding this comment.
Review Status
Actionable comments generated: 4
Configuration used: .coderabbit.yaml
Files selected for processing (3)
- services/rfq/relayer/inventory/manager.go (11 hunks)
- services/rfq/relayer/relconfig/config_test.go (4 hunks)
- services/rfq/relayer/service/statushandler.go (4 hunks)
Files skipped from review as they are similar to previous changes (2)
- services/rfq/relayer/relconfig/config_test.go
- services/rfq/relayer/service/statushandler.go
Additional comments: 4
services/rfq/relayer/inventory/manager.go (4)
- 21-21: The addition of the
cctpimport is necessary for the new rebalancing functionality. This change aligns with the PR objectives to enhance inventory management using the CCTP bridge.- 66-71: The addition of
txSubmitter,cctpContracts, andchainIDfields toinventoryManagerImplsupports the new rebalancing functionality. Ensure these fields are properly initialized and used securely throughout the class.- 149-157: The modification of the
NewInventoryManagerfunction signature to include thetxSubmitterparameter is necessary for the rebalancing functionality. Ensure that all instantiations ofNewInventoryManagerare updated accordingly.- 134-135: The update to
initializeTokensto include chain ID and token address intokenMetadatais a positive change for clarity and functionality. Ensure that these new fields are utilized correctly throughout the codebase, especially in the context of rebalancing.Also applies to: 461-471
| 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) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
📝 NOTE
This review was outside the diff hunks and was mapped to the diff hunk with the greatest overlap. Original lines [190-242]
The updated logic in ApproveAllTokens to handle approvals for both RFQ and CCTP bridges is crucial for the new functionality. Ensure that error handling is robust, especially in the context of transaction submissions. Consider adding more detailed logging for each step to aid in debugging and monitoring.
Consider enhancing error logging with more context, such as the chain ID and token address, to improve traceability in case of failures.
| // 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(chainID, token) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get rebalance: %w", err) | ||
| } | ||
| if rebalance == nil { | ||
| return nil | ||
| } | ||
|
|
||
| //nolint:exhaustive | ||
| switch method { | ||
| case relconfig.RebalanceMethodCCTP: | ||
| return i.rebalanceCCTP(ctx, rebalance) | ||
| case relconfig.RebalanceMethodNative: | ||
| return fmt.Errorf("native rebalance method not implemented") | ||
| default: | ||
| return fmt.Errorf("unknown rebalance method: %s", method) | ||
| } | ||
| } |
There was a problem hiding this comment.
The Rebalance method's implementation is key to the new inventory rebalancing functionality. Ensure that the contract interaction logic within rebalanceCCTP is secure and correctly handles potential errors. The placeholders (0 and []byte{}) in the SendCircleToken call should be reviewed and replaced with actual values if necessary.
Consider replacing the placeholders in the SendCircleToken call with actual values or configurations, as these might impact the functionality of the rebalancing operation.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Configuration used: .coderabbit.yaml
Files selected for processing (2)
- services/rfq/go.mod (1 hunks)
- services/rfq/relayer/relconfig/config.go (4 hunks)
Files skipped from review as they are similar to previous changes (2)
- services/rfq/go.mod
- services/rfq/relayer/relconfig/config.go
…ic-listener Use Generic DB for listener
There was a problem hiding this comment.
Review Status
Actionable comments generated: 2
Configuration used: .coderabbit.yaml
Files selected for processing (24)
- ethergo/README.md (1 hunks)
- ethergo/example/README.md (1 hunks)
- ethergo/example/counter/counter.abigen.go (3 hunks)
- ethergo/example/counter/counter.contractinfo.json (1 hunks)
- ethergo/example/counter/counter.sol (1 hunks)
- ethergo/example/deploymanager.go (1 hunks)
- ethergo/listener/db/doc.go (1 hunks)
- ethergo/listener/db/service.go (1 hunks)
- ethergo/listener/db/store.go (1 hunks)
- ethergo/listener/export_test.go (2 hunks)
- ethergo/listener/listener.go (9 hunks)
- ethergo/listener/listener_test.go (1 hunks)
- ethergo/listener/suite_test.go (1 hunks)
- services/rfq/e2e/rfq_test.go (10 hunks)
- services/rfq/go.mod (4 hunks)
- services/rfq/relayer/chain/chain.go (1 hunks)
- services/rfq/relayer/inventory/rebalance.go (1 hunks)
- services/rfq/relayer/relapi/server.go (3 hunks)
- services/rfq/relayer/reldb/base/model.go (3 hunks)
- services/rfq/relayer/reldb/base/store.go (3 hunks)
- services/rfq/relayer/reldb/db.go (4 hunks)
- services/rfq/relayer/reldb/db_test.go (1 hunks)
- services/rfq/relayer/service/relayer.go (6 hunks)
- services/rfq/testutil/deployers.go (3 hunks)
Files skipped from review due to trivial changes (1)
- ethergo/listener/db/doc.go
Files skipped from review as they are similar to previous changes (8)
- services/rfq/go.mod
- services/rfq/relayer/chain/chain.go
- services/rfq/relayer/inventory/rebalance.go
- services/rfq/relayer/relapi/server.go
- services/rfq/relayer/reldb/base/model.go
- services/rfq/relayer/reldb/base/store.go
- services/rfq/relayer/service/relayer.go
- services/rfq/testutil/deployers.go
Additional comments: 36
services/rfq/relayer/reldb/db_test.go (1)
- 13-13: The change from
reldb.ErrNoLatestBlockForChainIDtolistener.ErrNoLatestBlockForChainIDaligns with the refactoring effort to centralize or reorganize error definitions. Ensure thatlistener.ErrNoLatestBlockForChainIDis consistently used across the project for similar error scenarios.ethergo/example/deploymanager.go (1)
- 12-30: The introduction of the
DeployManagerstruct, its constructor functionNewDeployManager, and the methodGetCountereffectively utilize modern Go features, such as generics, and adhere to best practices. The use oft.Helper()in test-related functions is appropriate and enhances test output readability.ethergo/listener/export_test.go (1)
- 23-36: The refactoring of the
TestChainListenerArgsstruct and theNewTestChainListenerfunction, including the reordering of fields and the use of thedbpackage for theStorefield, enhances clarity and maintainability. Ensure that all tests relying on these constructs are updated accordingly to accommodate these changes.ethergo/example/counter/counter.sol (1)
- 8-35: The updates to the
Countercontract, including the addition of events (Incremented,Decremented,IncrementedByUser), thedeployBlockvariable, and event emissions within functions, adhere to Solidity best practices. These changes enhance contract observability and auditability while efficiently using storage for thedeployBlockvariable.ethergo/listener/listener_test.go (1)
- 11-57: The
TestListenForEventsfunction effectively simulates contract interactions and tests the event listening functionality using a wait group for concurrency management. Consider addressing the TODO comment about checking for a timeout to enhance the robustness of the test.ethergo/listener/db/store.go (1)
- 1-71: The implementation of the
Storestruct and its methodsPutLatestBlockandLatestBlockForChaineffectively utilizegormfor ORM operations, with appropriate error handling and adherence to best practices in database schema management. The custom errorErrNoLatestBlockForChainIDand the use of a custom namer for field names enhance clarity and consistency.ethergo/listener/suite_test.go (1)
- 1-157: The introduction of a test suite for the listener component and the
NewSqliteStorefunction for creating a SQLite store for testing purposes demonstrate a structured approach to testing. This enhances maintainability and readability, and the focused test cases with appropriate assertions align with Go testing practices.ethergo/README.md (1)
- 32-41: The updates to the README.md, including the removal of references to the
chaindirectory and the addition of thelistenerdirectory, accurately reflect significant architectural changes in the project. Keeping the documentation aligned with the codebase is crucial for clarity and user guidance.ethergo/listener/listener.go (6)
- 7-8: The replacement of the
fastbridgepackage with thedb2package for database operations indicates a significant change in how database interactions are handled. Ensure that all instances wherefastbridgewas used are properly updated to usedb2, and thatdb2provides all necessary functionalities that were previously covered byfastbridge.- 39-44: The
chainListenerstruct now includes adb2.ChainListenerDBtype for thestorefield and aninitialBlockparameter. This change is crucial for integrating the new database interactions and handling the initial block from which the listener starts. Ensure that theinitialBlockparameter is correctly utilized throughout the listener's logic, especially in methods that deal with block ranges and event fetching.- 59-66: The
NewChainListenerfunction signature has been updated to include theinitialBlockparameter and use the newdb2.ChainListenerDBtype. This change is necessary for initializing the listener with the correct starting block and database interface. Verify that all calls toNewChainListenerthroughout the codebase have been updated to pass the correctinitialBlockvalue and that thedb2.ChainListenerDBis properly initialized before being passed to this function.Verification successful
Given the lack of output from the initial scripts, it's necessary to take a different approach to verify the implementation of the changes mentioned in the review comment. Let's broaden the search to capture any usage of
NewChainListenerand the initialization ofdb2.ChainListenerDBin the codebase.
To accurately verify the implementation of the changes mentioned in the review comment, let's correct the syntax error in the previous
rgcommands and re-run the search without specifying an incorrect flag.
The verification process has confirmed that the
NewChainListenerfunction signature has been correctly updated to include theinitialBlockparameter and use thedb2.ChainListenerDBtype across the codebase. Calls toNewChainListenercorrectly pass theinitialBlockvalue, anddb2.ChainListenerDBis properly initialized before being passed to this function. Based on the provided evidence, the changes mentioned in the review comment have been successfully implemented.* 99-99: The `Address` method now returns the `address` directly instead of calling `contract.Address()`. This simplification is beneficial for performance and readability. However, ensure that the `address` field is always correctly set during the listener's initialization to prevent potential nil pointer dereferences or incorrect address usage. * 128-128: The modification in the `doPoll` method to return `nil` when the `latestBlock` is the same as the `startBlock` is a logical optimization to avoid unnecessary processing. This change should help in reducing the load on the system for chains with slow block times. Ensure that this logic does not interfere with the listener's ability to catch up on missed events in case of temporary downtimes or delays. * 185-191: > 📝 **NOTE** > This review was outside the diff hunks and was mapped to the diff hunk with the greatest overlap. Original lines [165-207]Scripts Executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Search for calls to NewChainListener and verify the correct initialBlock value is passed. ast-grep --lang go --pattern $'NewChainListener($_, $_, $_, $_, $_)' # Ensure db2.ChainListenerDB is properly initialized before being passed to NewChainListener. ast-grep --lang go --pattern $'db2.ChainListenerDB($_)'Length of output: 126
Script:
#!/bin/bash # Broaden the search for occurrences of NewChainListener to ensure no usage is missed. rg "NewChainListener" --go # Search for any initialization or reference to db2.ChainListenerDB to verify proper usage. rg "db2.ChainListenerDB" --goLength of output: 914
Script:
#!/bin/bash # Correct the search for occurrences of NewChainListener. rg "NewChainListener" # Correct the search for any initialization or reference to db2.ChainListenerDB. rg "db2.ChainListenerDB"Length of output: 1752
The
getMetadatamethod has been refactored to removedeployBlockretrieval and to handle block comparisons based onstartBlockandinitialBlock. This change aligns with the introduction of theinitialBlockparameter and simplifies the metadata fetching process. Verify that the removal ofdeployBlockretrieval does not affect any functionalities that relied on this information and that the block comparisons are correctly implemented to ensure the listener starts from the appropriate block.services/rfq/relayer/reldb/db.go (4)
- 8-9: The addition of the
math/bigpackage import is necessary for handling big integers in theRebalancestruct, specifically for theOriginAmountfield. This is a good practice for dealing with large numeric values in blockchain-related operations. Ensure that all operations involvingOriginAmountproperly utilize themath/bigpackage's functionalities to handle big integers correctly.- 19-28: The introduction of
StoreRebalanceandUpdateRebalanceStatusmethods in theWriterinterface is crucial for managing rebalance actions in the database. Ensure that the implementations of these methods correctly handle database transactions, including error handling and rollback mechanisms in case of failures, to maintain data integrity.- 41-42: The
HasPendingRebalancemethod in theReaderinterface is a valuable addition for checking the existence of pending rebalances for given chain IDs. This method can help in decision-making processes related to initiating new rebalances. Ensure that this method efficiently queries the database to minimize performance impacts, especially when dealing with a large number of rebalance records.- 157-207: The definition of the
Rebalancestruct and theRebalanceStatusenum introduces a structured way to represent rebalance actions and their statuses within the database. This structured approach enhances code readability and maintainability. Ensure that theRebalanceIDfield is correctly used as a unique identifier for rebalance actions and that theRebalanceStatusenum covers all possible states of a rebalance action.ethergo/example/counter/counter.contractinfo.json (1)
- 1-1: The addition of event declarations (
Incremented,Decremented,IncrementedByUser) and a new public immutable variabledeployBlockin theCountercontract enhances the contract's functionality by providing event logging capabilities and access to the deployment block number. These changes are well-aligned with Solidity best practices for contract development. Ensure that the events are emitted correctly in the contract's functions and that thedeployBlockvariable is properly initialized in the constructor.services/rfq/e2e/rfq_test.go (9)
- 31-31: The introduction of the
cctpDeployManagerfield in theIntegrationSuitestruct is a key part of setting up the CCTP rebalancing feature. This change aligns with the PR objectives.- 38-38: Adding a
storefield of typereldb.Serviceto theIntegrationSuitestruct is crucial for testing the rebalancing feature, as it likely involves database interactions to track pending rebalances.- 76-76: The initialization of
cctpDeployManagerwithin theSetupTestmethod is correctly done, ensuring that the CCTP rebalancing feature can be tested effectively.- 107-110: The logic for adjusting token amounts to account for decimals is a good practice, ensuring that tests accurately reflect real-world scenarios.
- 114-116: The process of minting initial USDC to the relayer wallet on the destination chain is correctly implemented. However, ensure that error handling is robust and that the transaction is properly awaited for confirmation.
- 147-147: The logic to verify that the destination amount is greater than the requested RFQ amount is crucial for ensuring the quote meets the user's needs. This check is well-placed within the flow of the test.
- 161-168: The use of
Bridgefunction to simulate a bridging transaction is correctly implemented. Pay attention to the subtraction of a small amount (10_000_000) fromrealRFQAmountforDestAmount, ensuring a realistic test scenario.- 213-213: The balance check to verify the rebalance has occurred by comparing the balance against a threshold is a smart way to validate the rebalancing logic. However, ensure that the threshold calculation (
1.5 * realStartAmount) aligns with the expected rebalancing logic.Verification successful
The verification process confirms that the balance threshold calculation indeed aligns with the rebalancing logic, using a multiplier of 1.5 times the
realStartAmount, as mentioned in the review comment. This aligns with the expected behavior described in the comment.* 234-241: Checking for pending rebalances on both the destination and origin chains is a critical part of validating the rebalancing feature. Ensure that the `HasPendingRebalance` method is correctly implemented and returns accurate results.Scripts Executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Verify that the balance threshold calculation aligns with rebalancing logic grep "balanceThresh" services/rfq/e2e/rfq_test.goLength of output: 196
Verification successful
The
rgcommand successfully identified theHasPendingRebalancemethod within the codebase, specifically inservices/rfq/relayer/reldb/db.goand its implementation inservices/rfq/relayer/reldb/base/rebalance.go. This method is designed to check if there is a pending rebalance for the given chain IDs, aligning with the usage observed in theservices/rfq/e2e/rfq_test.gofile. The method accepts a context and a variadic number ofuint64chain IDs, returning a boolean indicating the presence of a pending rebalance and an error if applicable.Given this information, the review comment regarding the criticality of checking for pending rebalances on both the destination and origin chains and ensuring the
HasPendingRebalancemethod's correct implementation and accurate results is validated by the method's presence and its usage in the codebase.Scripts Executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Verify implementation of HasPendingRebalance method ast-grep --lang go --pattern $'HasPendingRebalance($_, $_)'Length of output: 60
Script:
#!/bin/bash # Broaden the search for HasPendingRebalance method ast-grep --lang go --pattern $'func HasPendingRebalance'Length of output: 402
Script:
#!/bin/bash # Fallback: Use rg to search for HasPendingRebalance in the codebase rg 'HasPendingRebalance'Length of output: 968
ethergo/example/README.md (3)
- 198-224: The documentation for creating a typecast getter
DeployManageris clear and provides a good example of how to avoid naked casts of contract handles. This enhances code readability and maintainability.- 241-246: The example provided for the
GetCountermethod in theDeployManagerdemonstrates a practical use of the typecast getter. This method simplifies the process of retrieving contract handles and improves code clarity.- 249-269: The addition of a test to ensure that dependencies are correctly listed in each deployer is a good practice. This test helps maintain the integrity of the deployment process by verifying that all dependencies are accounted for.
ethergo/example/counter/counter.abigen.go (5)
- 34-34: The ABI string has been updated to include a constructor, new events, and functions. Ensure that this ABI string accurately reflects the latest version of the smart contract it represents. Changes in the ABI string can significantly affect how the Go code interacts with the contract on the blockchain.
- 217-232: The
DeployBlockfunction has been added as a new method to retrieve the block number at which the contract was deployed. This is a useful addition for tracking contract deployment without external tools. Ensure that the Solidity contract has a correspondingdeployBlockmethod that returns the expecteduint256type.- 373-438: The addition of
CounterDecrementedIteratorand related logic for handling theDecrementedevent is correctly implemented following Go and Ethereum binding standards. This allows for efficient iteration over logs of theDecrementedevent. Ensure that the event names and types match those defined in the smart contract.- 507-572: Similar to the
CounterDecrementedIterator, theCounterIncrementedIteratoris well-implemented for theIncrementedevent. This consistency in handling different events is crucial for maintainability and ease of understanding the code. Again, verify that the event details align with the smart contract.- 641-706: The
CounterIncrementedByUserIteratorfor theIncrementedByUserevent follows the established pattern for event iterators. This addition is important for applications that need to track increments made by specific users. As with other events, ensure the contract's event definitions are consistent with these bindings.
| @@ -14,23 +15,27 @@ import ( | |||
| "github.com/synapsecns/sanguine/ethergo/backends/anvil" | |||
| "github.com/synapsecns/sanguine/ethergo/signer/signer/localsigner" | |||
| "github.com/synapsecns/sanguine/ethergo/signer/wallet" | |||
| cctpTest "github.com/synapsecns/sanguine/services/cctp-relayer/testutil" | |||
There was a problem hiding this comment.
Tip
Codebase Verification
The verification process did not find the cctp-relayer/testutil dependency in the services/cctp-relayer/go.mod file. This suggests that the dependency necessary for the new rebalancing feature might not be properly declared, potentially leading to issues with dependency resolution.
Analysis chain
The addition of the cctpTest package import is necessary for the new rebalancing feature. Ensure that the go.mod file is updated accordingly to include this new dependency.
---
<!--
[observations] The script successfully located multiple `go.mod` files across the repository, indicating a multi-module setup. The relevant `go.mod` file for our verification is `services/cctp-relayer/go.mod`, given the context of the import path `github.com/synapsecns/sanguine/services/cctp-relayer/testutil`.
[decomposition] The next step is to verify the presence of the `cctp-relayer/testutil` dependency in the `services/cctp-relayer/go.mod` file, as it's the most relevant to the import statement in question.
[analysis] Since the `go.mod` file within the `services/cctp-relayer` directory is directly related to the import path mentioned, checking this file for the dependency is the correct approach. This will confirm whether the dependency has been properly declared.
[actions] Plan to generate a new script to check the `services/cctp-relayer/go.mod` file for the `cctp-relayer/testutil` dependency.
-->
</details>
<details>
<summary>Scripts Executed</summary>
The following scripts were executed for the analysis:
---
Script:
```shell
#!/bin/bash
# Verify if go.mod contains cctp-relayer/testutil as a dependency
grep "cctp-relayer/testutil" go.mod
Length of output: 75
Script:
#!/bin/bash
# Find the go.mod file in the repository
fd go.mod
# Assuming go.mod is found, re-run the grep command with the correct path
Length of output: 489
Script:
#!/bin/bash
# Verify if services/cctp-relayer/go.mod contains cctp-relayer/testutil as a dependency
grep "cctp-relayer/testutil" services/cctp-relayer/go.mod
Length of output: 57
| for _, rpcAddr := range []string{i.originBackend.RPCAddress(), i.destBackend.RPCAddress()} { | ||
| anvilClient, err := anvil.Dial(i.GetTestContext(), rpcAddr) | ||
| i.NoError(err) | ||
|
|
||
| // because can claim works on last block timestamp, we need to do something | ||
| err = anvilClient.Mine(i.GetTestContext(), 1) | ||
| i.NoError(err) | ||
| go func() { | ||
| for { | ||
| select { | ||
| case <-i.GetTestContext().Done(): | ||
| return | ||
| case <-time.After(time.Second * 4): | ||
| // increase time by 30 mintutes every second, should be enough to get us a fastish e2e test | ||
| // we don't need to worry about deadline since we're only doing this on origin | ||
| err = anvilClient.IncreaseTime(i.GetTestContext(), 60*30) | ||
| i.NoError(err) | ||
|
|
||
| // because can claim works on last block timestamp, we need to do something | ||
| err = anvilClient.Mine(i.GetTestContext(), 1) | ||
| i.NoError(err) | ||
| } | ||
| } | ||
| } | ||
| }() | ||
| }() | ||
| } |
There was a problem hiding this comment.
The refactoring to handle multiple RPC addresses in a loop is a significant improvement in terms of code maintainability and readability. Ensure that error handling within the loop is robust and that the loop correctly terminates under all expected conditions. Additionally, consider abstracting this logic into a helper function for better readability and reuse, as previously suggested.
- // Current loop implementation
+ // Suggestion: Abstract this logic into a helper function for improved maintainabilityCommittable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| for _, rpcAddr := range []string{i.originBackend.RPCAddress(), i.destBackend.RPCAddress()} { | |
| anvilClient, err := anvil.Dial(i.GetTestContext(), rpcAddr) | |
| i.NoError(err) | |
| // because can claim works on last block timestamp, we need to do something | |
| err = anvilClient.Mine(i.GetTestContext(), 1) | |
| i.NoError(err) | |
| go func() { | |
| for { | |
| select { | |
| case <-i.GetTestContext().Done(): | |
| return | |
| case <-time.After(time.Second * 4): | |
| // increase time by 30 mintutes every second, should be enough to get us a fastish e2e test | |
| // we don't need to worry about deadline since we're only doing this on origin | |
| err = anvilClient.IncreaseTime(i.GetTestContext(), 60*30) | |
| i.NoError(err) | |
| // because can claim works on last block timestamp, we need to do something | |
| err = anvilClient.Mine(i.GetTestContext(), 1) | |
| i.NoError(err) | |
| } | |
| } | |
| } | |
| }() | |
| }() | |
| } | |
| for _, rpcAddr := range []string{i.originBackend.RPCAddress(), i.destBackend.RPCAddress()} { | |
| anvilClient, err := anvil.Dial(i.GetTestContext(), rpcAddr) | |
| i.NoError(err) | |
| go func() { | |
| for { | |
| select { | |
| case <-i.GetTestContext().Done(): | |
| return | |
| case <-time.After(time.Second * 4): | |
| // Suggestion: Abstract this logic into a helper function for improved maintainability | |
| // increase time by 30 mintutes every second, should be enough to get us a fastish e2e test | |
| // we don't need to worry about deadline since we're only doing this on origin | |
| err = anvilClient.IncreaseTime(i.GetTestContext(), 60*30) | |
| i.NoError(err) | |
| // because can claim works on last block timestamp, we need to do something | |
| err = anvilClient.Mine(i.GetTestContext(), 1) | |
| i.NoError(err) | |
| } | |
| } | |
| }() | |
| } |
…ne into feat/relayer-rebalance
There was a problem hiding this comment.
Review Status
Actionable comments generated: 4
Configuration used: .coderabbit.yaml
Files selected for processing (3)
- services/rfq/relayer/inventory/export_test.go (1 hunks)
- services/rfq/relayer/inventory/manager.go (14 hunks)
- services/rfq/relayer/inventory/rebalance.go (1 hunks)
Files skipped from review as they are similar to previous changes (2)
- services/rfq/relayer/inventory/export_test.go
- services/rfq/relayer/inventory/rebalance.go
Additional comments: 2
services/rfq/relayer/inventory/manager.go (2)
- 69-74: The addition of
txSubmitter,rebalanceManagers, andchainIDfields to theinventoryManagerImplstruct is a crucial part of supporting the new rebalancing functionality. However, it's important to ensure that these fields are properly initialized and used throughout the class to avoid null pointer exceptions or logic errors.Please verify that these new fields are correctly initialized in all relevant constructors or initialization methods and that their usage throughout the class is consistent and error-free.
- 269-330: > 📝 NOTE
This review was outside the diff hunks and was mapped to the diff hunk with the greatest overlap. Original lines [260-296]
The
ApproveAllTokensmethod has been enhanced to manage approvals for both RFQ and CCTP bridges. This is a critical update for ensuring that token approvals are correctly managed across different bridges. However, it's important to ensure that error handling is robust and that the method does not inadvertently approve tokens more than once, especially in cases where the submitter hasn't finished from the last boot (as noted in the comments). Consider adding checks to prevent double submissions of approvals.Please verify that the logic to prevent double submissions of approvals is robust and test this method thoroughly to ensure it behaves as expected under various conditions.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 4
Configuration used: .coderabbit.yaml
Files selected for processing (1)
- services/rfq/relayer/inventory/manager.go (14 hunks)
Additional comments: 1
services/rfq/relayer/inventory/manager.go (1)
- 69-74: The introduction of
txSubmitter,rebalanceManagers, andchainIDfields in theinventoryManagerImplstruct is a necessary update to support the new rebalancing functionality. However, ensure that these fields are properly initialized and used throughout the implementation to avoid any null pointer exceptions or logic errors.Verification successful
Based on the provided context and the output from the shell scripts, it's clear that the introduction of
txSubmitter,rebalanceManagers, andchainIDfields in theinventoryManagerImplstruct is part of a larger update to support new rebalancing functionality. The output shows that these fields are properly initialized in theNewInventoryManagerfunction and are used throughout the implementation, which aligns with the requirements mentioned in the review comment.The initialization of these fields is confirmed by their presence in the
NewInventoryManagerfunction signature and the assignment within this function. The usage of these fields throughout the implementation is evident from the numerous references totxSubmitter,rebalanceManagers, andchainIDacross different files and contexts, indicating their integration into the broader system logic.Given this information, it can be concluded that the introduction and utilization of these fields have been correctly implemented as per the requirements outlined in the review comment.
Scripts Executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Verify that all new fields are properly initialized in the NewInventoryManager function. rg --type go 'NewInventoryManager\(' # Verify that these fields are used correctly throughout the implementation. rg --type go 'txSubmitter|rebalanceManagers|chainID'Length of output: 363239
There was a problem hiding this comment.
Review Status
Actionable comments generated: 6
Configuration used: .coderabbit.yaml
Files selected for processing (5)
- services/rfq/relayer/inventory/manager.go (14 hunks)
- services/rfq/relayer/inventory/manager_test.go (1 hunks)
- services/rfq/relayer/relconfig/config.go (4 hunks)
- services/rfq/relayer/relconfig/config_test.go (5 hunks)
- services/rfq/relayer/relconfig/getters.go (5 hunks)
Files skipped from review as they are similar to previous changes (3)
- services/rfq/relayer/inventory/manager_test.go
- services/rfq/relayer/relconfig/config_test.go
- services/rfq/relayer/relconfig/getters.go
Additional comments: 4
services/rfq/relayer/relconfig/config.go (3)
- 47-50: The conversion of
DBSelectorIntervalSecondstoDBSelectorIntervaland the introduction ofRebalanceIntervalastime.Durationtypes are positive changes for clarity and type safety. However, ensure that all places in the codebase that use these fields are updated to work withtime.Durationinstead of integers representing seconds.Verification successful
The conversion of
DBSelectorIntervalSecondstoDBSelectorIntervaland the introduction ofRebalanceIntervalastime.Durationtypes have been correctly implemented across the codebase. The presence of getter functions for both configurations, along with default values and their usage in logical parts of the application, confirms that these fields are handled astime.Durationappropriately. This ensures clarity and type safety, as intended by the changes. All places in the codebase that use these fields appear to have been updated to work withtime.Durationinstead of integers representing seconds, as per the verification results.* 55-58: Renaming the `Bridge` field to `RFQAddress` and adding the `CCTPAddress` field in the `ChainConfig` struct are significant for clarity and functionality. Ensure that all references to the old `Bridge` field are updated to `RFQAddress` across the codebase. * 85-86: The addition of `CCTPStartBlock` in the `ChainConfig` struct is crucial for specifying the start block for CCTP event listening. Ensure that this new field is properly utilized where event listening is set up.Scripts Executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Search for usages of DBSelectorInterval and RebalanceInterval to ensure they are correctly handled as time.Duration rg --type go 'DBSelectorInterval|RebalanceInterval' services/rfq/Length of output: 1664
Verification successful
The examination of
GetCCTPStartBlockusage inrebalance.goconfirms that theCCTPStartBlockfield is indeed utilized within the system, specifically in obtaining aninitialBlockfor a givenchainID. This usage strongly suggests its role in setting up event listening, aligning with the intent of the original comment. While the direct evidence of event listening setup was not explicitly shown, the context of its usage supports the intended utilization of specifying the start block for CCTP event listening.Scripts Executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Verify CCTPStartBlock is utilized in event listening setup rg --type go 'CCTPStartBlock' services/rfq/Length of output: 681
Script:
#!/bin/bash # Examine the usage of GetCCTPStartBlock in rebalance.go rg --type go 'GetCCTPStartBlock' services/rfq/relayer/inventory/rebalance.goLength of output: 132
services/rfq/relayer/inventory/manager.go (1)
- 69-74: The introduction of
txSubmitter,rebalanceManagers, andchainIDfields in theinventoryManagerImplstruct is a crucial update for supporting the new rebalancing functionality. Ensure that these fields are properly initialized and used throughout the class to avoid any null pointer exceptions or logic errors.
| // RebalanceMethod is the method to use for rebalancing. | ||
| RebalanceMethod string `yaml:"rebalance_method"` | ||
| // MaintenanceBalancePct is the percentage of the total balance under which a rebalance will be triggered. | ||
| MaintenanceBalancePct float64 `yaml:"maintenance_balance_pct"` | ||
| // InitialBalancePct is the percentage of the total balance to retain when triggering a rebalance. | ||
| InitialBalancePct float64 `yaml:"initial_balance_pct"` | ||
| // MaxRebalanceAmount is the maximum amount to rebalance in human-readable units. | ||
| MaxRebalanceAmount string `yaml:"max_rebalance_amount"` |
There was a problem hiding this comment.
The introduction of rebalancing settings (RebalanceMethod, MaintenanceBalancePct, InitialBalancePct, and MaxRebalanceAmount) in the TokenConfig struct is a key enhancement. It's crucial to implement validation for these new settings to prevent configuration errors that could impact rebalancing functionality. Consider adding a validation function in the Config struct or as part of the LoadConfig function to ensure these settings are within acceptable ranges.
Would you like me to help with implementing validation for these new configuration settings?
| 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) { | ||
| rebalanceMethods, err := cfg.GetRebalanceMethods() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not get rebalance methods: %w", err) | ||
| } | ||
| rebalanceManagers := make(map[relconfig.RebalanceMethod]RebalanceManager) | ||
| for method := range rebalanceMethods { | ||
| //nolint:exhaustive | ||
| switch method { | ||
| case relconfig.RebalanceMethodCCTP: | ||
| rebalanceManagers[method] = newRebalanceManagerCCTP(cfg, handler, clientFetcher, txSubmitter, relayer, db) | ||
| default: | ||
| return nil, fmt.Errorf("unsupported rebalance method: %s", method) | ||
| } | ||
| } | ||
|
|
||
| i := inventoryManagerImpl{ | ||
| relayerAddress: relayer, | ||
| handler: handler, | ||
| cfg: cfg, | ||
| chainClient: clientFetcher, | ||
| db: db, | ||
| relayerAddress: relayer, | ||
| handler: handler, | ||
| cfg: cfg, | ||
| chainClient: clientFetcher, | ||
| txSubmitter: txSubmitter, | ||
| rebalanceManagers: rebalanceManagers, | ||
| db: db, | ||
| } | ||
|
|
||
| err := i.initializeTokens(ctx, cfg) | ||
| err = i.initializeTokens(ctx, cfg) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not initialize tokens: %w", err) | ||
| } | ||
|
|
||
| // TODO: move | ||
| go func() { | ||
| return &i, nil |
There was a problem hiding this comment.
The NewInventoryManager function has been significantly modified to accommodate the new rebalancing functionality. While the changes are necessary, the function now has a high number of parameters, which could impact readability and maintainability. Consider refactoring this function to reduce its complexity, possibly by introducing a struct to encapsulate some of the parameters or by using a builder pattern.
- 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) {
+ // Consider introducing a struct to encapsulate some parameters or using a builder pattern.Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| 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) { | |
| rebalanceMethods, err := cfg.GetRebalanceMethods() | |
| if err != nil { | |
| return nil, fmt.Errorf("could not get rebalance methods: %w", err) | |
| } | |
| rebalanceManagers := make(map[relconfig.RebalanceMethod]RebalanceManager) | |
| for method := range rebalanceMethods { | |
| //nolint:exhaustive | |
| switch method { | |
| case relconfig.RebalanceMethodCCTP: | |
| rebalanceManagers[method] = newRebalanceManagerCCTP(cfg, handler, clientFetcher, txSubmitter, relayer, db) | |
| default: | |
| return nil, fmt.Errorf("unsupported rebalance method: %s", method) | |
| } | |
| } | |
| i := inventoryManagerImpl{ | |
| relayerAddress: relayer, | |
| handler: handler, | |
| cfg: cfg, | |
| chainClient: clientFetcher, | |
| db: db, | |
| relayerAddress: relayer, | |
| handler: handler, | |
| cfg: cfg, | |
| chainClient: clientFetcher, | |
| txSubmitter: txSubmitter, | |
| rebalanceManagers: rebalanceManagers, | |
| db: db, | |
| } | |
| err := i.initializeTokens(ctx, cfg) | |
| err = i.initializeTokens(ctx, cfg) | |
| if err != nil { | |
| return nil, fmt.Errorf("could not initialize tokens: %w", err) | |
| } | |
| // TODO: move | |
| go func() { | |
| return &i, nil | |
| // Consider introducing a struct to encapsulate some parameters or using a builder pattern. | |
| 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) { | |
| rebalanceMethods, err := cfg.GetRebalanceMethods() | |
| if err != nil { | |
| return nil, fmt.Errorf("could not get rebalance methods: %w", err) | |
| } | |
| rebalanceManagers := make(map[relconfig.RebalanceMethod]RebalanceManager) | |
| for method := range rebalanceMethods { | |
| //nolint:exhaustive | |
| switch method { | |
| case relconfig.RebalanceMethodCCTP: | |
| rebalanceManagers[method] = newRebalanceManagerCCTP(cfg, handler, clientFetcher, txSubmitter, relayer, db) | |
| default: | |
| return nil, fmt.Errorf("unsupported rebalance method: %s", method) | |
| } | |
| } | |
| i := inventoryManagerImpl{ | |
| relayerAddress: relayer, | |
| handler: handler, | |
| cfg: cfg, | |
| chainClient: clientFetcher, | |
| txSubmitter: txSubmitter, | |
| rebalanceManagers: rebalanceManagers, | |
| db: db, | |
| } | |
| err = i.initializeTokens(ctx, cfg) | |
| if err != nil { | |
| return nil, fmt.Errorf("could not initialize tokens: %w", err) | |
| } | |
| return &i, nil |
There was a problem hiding this comment.
Review Status
Actionable comments generated: 6
Configuration used: .coderabbit.yaml
Files selected for processing (1)
- services/rfq/relayer/inventory/rebalance.go (1 hunks)
Additional comments: 3
services/rfq/relayer/inventory/rebalance.go (3)
- 23-28: The
RebalanceDatastruct is well-defined and encapsulates all necessary information for a rebalance action. It's clear and straightforward, making it easy to understand the data involved in a rebalance operation.- 30-36: The
RebalanceManagerinterface is concise and captures the essential operations needed for rebalancing: starting the manager and executing a rebalance action. This design adheres to the interface segregation principle, ensuring that the interface is specific to the rebalance functionality.- 57-68: The constructor function
newRebalanceManagerCCTPis correctly implemented, initializing maps forcctpContractsandchainListenersto avoid nil map assignment errors. This is a good practice for ensuring the struct is ready for use immediately after construction.
| type rebalanceManagerCCTP struct { | ||
| // cfg is the config | ||
| cfg relconfig.Config | ||
| // handler is the metrics handler | ||
| handler metrics.Handler | ||
| // chainClient is an omnirpc client | ||
| chainClient submitter.ClientFetcher | ||
| // txSubmitter is the transaction submitter | ||
| txSubmitter submitter.TransactionSubmitter | ||
| // cctpContracts is the map of cctp contracts (used for rebalancing) | ||
| cctpContracts map[int]*cctp.SynapseCCTP | ||
| // relayerAddress contains the relayer address | ||
| relayerAddress common.Address | ||
| // chainListeners is the map of chain listeners for CCTP events | ||
| chainListeners map[int]listener.ContractListener | ||
| // db is the database | ||
| db reldb.Service |
There was a problem hiding this comment.
The rebalanceManagerCCTP struct is well-organized, with clear field names indicating their purposes. However, consider documenting each field to improve code readability and maintainability, especially for complex fields like cctpContracts and chainListeners.
+ // cfg holds the configuration settings for the rebalance manager.
cfg relconfig.Config
+ // handler is responsible for managing metrics.
handler metrics.Handler
+ // chainClient fetches clients for different chains.
chainClient submitter.ClientFetcher
+ // txSubmitter submits transactions to the blockchain.
txSubmitter submitter.TransactionSubmitter
+ // cctpContracts maps chain IDs to their respective CCTP contracts.
cctpContracts map[int]*cctp.SynapseCCTP
+ // relayerAddress is the Ethereum address of the relayer.
relayerAddress common.Address
+ // chainListeners maps chain IDs to their respective contract listeners.
chainListeners map[int]listener.ContractListener
+ // db provides access to the relational database.
db reldb.ServiceCommittable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| type rebalanceManagerCCTP struct { | |
| // cfg is the config | |
| cfg relconfig.Config | |
| // handler is the metrics handler | |
| handler metrics.Handler | |
| // chainClient is an omnirpc client | |
| chainClient submitter.ClientFetcher | |
| // txSubmitter is the transaction submitter | |
| txSubmitter submitter.TransactionSubmitter | |
| // cctpContracts is the map of cctp contracts (used for rebalancing) | |
| cctpContracts map[int]*cctp.SynapseCCTP | |
| // relayerAddress contains the relayer address | |
| relayerAddress common.Address | |
| // chainListeners is the map of chain listeners for CCTP events | |
| chainListeners map[int]listener.ContractListener | |
| // db is the database | |
| db reldb.Service | |
| type rebalanceManagerCCTP struct { | |
| // cfg holds the configuration settings for the rebalance manager. | |
| cfg relconfig.Config | |
| // handler is responsible for managing metrics. | |
| handler metrics.Handler | |
| // chainClient fetches clients for different chains. | |
| chainClient submitter.ClientFetcher | |
| // txSubmitter submits transactions to the blockchain. | |
| txSubmitter submitter.TransactionSubmitter | |
| // cctpContracts maps chain IDs to their respective CCTP contracts. | |
| cctpContracts map[int]*cctp.SynapseCCTP | |
| // relayerAddress is the Ethereum address of the relayer. | |
| relayerAddress common.Address | |
| // chainListeners maps chain IDs to their respective contract listeners. | |
| chainListeners map[int]listener.ContractListener | |
| // db provides access to the relational database. | |
| db reldb.Service |
| func (c *rebalanceManagerCCTP) Start(ctx context.Context) (err error) { | ||
| err = c.initContracts(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("could not initialize contracts: %w", err) | ||
| } | ||
|
|
||
| err = c.initListeners(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("could not initialize listeners: %w", err) | ||
| } | ||
|
|
||
| g, _ := errgroup.WithContext(ctx) | ||
| for cid := range c.cfg.Chains { | ||
| // capture func literal | ||
| chainID := cid | ||
| g.Go(func() error { | ||
| return c.listen(ctx, chainID) | ||
| }) | ||
| } | ||
|
|
||
| err = g.Wait() | ||
| if err != nil { | ||
| return fmt.Errorf("error listening to contract: %w", err) | ||
| } | ||
| return nil |
There was a problem hiding this comment.
The Start method correctly initializes contracts and listeners before starting to listen for CCTP events. The use of errgroup for concurrent listening across multiple chains is a good practice for performance. However, ensure proper error handling and logging within the goroutines to avoid silent failures.
+ // Add logging for goroutine errors
g.Go(func() error {
err := c.listen(ctx, chainID)
+ if err != nil {
+ logger.Errorf("Error listening on chain %d: %v", chainID, err)
+ }
return err
})Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| func (c *rebalanceManagerCCTP) Start(ctx context.Context) (err error) { | |
| err = c.initContracts(ctx) | |
| if err != nil { | |
| return fmt.Errorf("could not initialize contracts: %w", err) | |
| } | |
| err = c.initListeners(ctx) | |
| if err != nil { | |
| return fmt.Errorf("could not initialize listeners: %w", err) | |
| } | |
| g, _ := errgroup.WithContext(ctx) | |
| for cid := range c.cfg.Chains { | |
| // capture func literal | |
| chainID := cid | |
| g.Go(func() error { | |
| return c.listen(ctx, chainID) | |
| }) | |
| } | |
| err = g.Wait() | |
| if err != nil { | |
| return fmt.Errorf("error listening to contract: %w", err) | |
| } | |
| return nil | |
| func (c *rebalanceManagerCCTP) Start(ctx context.Context) (err error) { | |
| err = c.initContracts(ctx) | |
| if err != nil { | |
| return fmt.Errorf("could not initialize contracts: %w", err) | |
| } | |
| err = c.initListeners(ctx) | |
| if err != nil { | |
| return fmt.Errorf("could not initialize listeners: %w", err) | |
| } | |
| g, _ := errgroup.WithContext(ctx) | |
| for cid := range c.cfg.Chains { | |
| // capture func literal | |
| chainID := cid | |
| g.Go(func() error { | |
| // Add logging for goroutine errors | |
| err := c.listen(ctx, chainID) | |
| if err != nil { | |
| logger.Errorf("Error listening on chain %d: %v", chainID, err) | |
| } | |
| return err | |
| }) | |
| } | |
| err = g.Wait() | |
| if err != nil { | |
| return fmt.Errorf("error listening to contract: %w", err) | |
| } | |
| return nil |
| func (c *rebalanceManagerCCTP) initContracts(ctx context.Context) (err error) { | ||
| for chainID := range c.cfg.Chains { | ||
| contractAddr, err := c.cfg.GetCCTPAddress(chainID) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get cctp address: %w", err) | ||
| } | ||
| chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) | ||
| 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) | ||
| } | ||
| c.cctpContracts[chainID] = contract | ||
| } | ||
| return nil |
There was a problem hiding this comment.
The initContracts method demonstrates good error handling with detailed error messages. However, consider parallelizing the initialization of contracts for different chains to improve startup performance, similar to how listening is handled concurrently.
+ // Consider using errgroup for parallel initializationCommittable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| func (c *rebalanceManagerCCTP) initContracts(ctx context.Context) (err error) { | |
| for chainID := range c.cfg.Chains { | |
| contractAddr, err := c.cfg.GetCCTPAddress(chainID) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp address: %w", err) | |
| } | |
| chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) | |
| 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) | |
| } | |
| c.cctpContracts[chainID] = contract | |
| } | |
| return nil | |
| func (c *rebalanceManagerCCTP) initContracts(ctx context.Context) (err error) { | |
| // Consider using errgroup for parallel initialization | |
| for chainID := range c.cfg.Chains { | |
| contractAddr, err := c.cfg.GetCCTPAddress(chainID) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp address: %w", err) | |
| } | |
| chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) | |
| 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) | |
| } | |
| c.cctpContracts[chainID] = contract | |
| } | |
| return nil | |
| } |
| func (c *rebalanceManagerCCTP) initListeners(ctx context.Context) (err error) { | ||
| for chainID := range c.cfg.GetChains() { | ||
| cctpAddr, err := c.cfg.GetCCTPAddress(chainID) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get cctp address: %w", err) | ||
| } | ||
| chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get chain client: %w", err) | ||
| } | ||
| initialBlock, err := c.cfg.GetCCTPStartBlock(chainID) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get cctp start block: %w", err) | ||
| } | ||
| chainListener, err := listener.NewChainListener(chainClient, c.db, common.HexToAddress(cctpAddr), initialBlock, c.handler) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get chain listener: %w", err) | ||
| } | ||
| c.chainListeners[chainID] = chainListener | ||
| } | ||
| return nil |
There was a problem hiding this comment.
Similar to the initContracts method, initListeners also has good error handling. Again, parallelizing the initialization of listeners for different chains could enhance performance.
+ // Consider using errgroup for parallel initialization of listenersCommittable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| func (c *rebalanceManagerCCTP) initListeners(ctx context.Context) (err error) { | |
| for chainID := range c.cfg.GetChains() { | |
| cctpAddr, err := c.cfg.GetCCTPAddress(chainID) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp address: %w", err) | |
| } | |
| chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) | |
| if err != nil { | |
| return fmt.Errorf("could not get chain client: %w", err) | |
| } | |
| initialBlock, err := c.cfg.GetCCTPStartBlock(chainID) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp start block: %w", err) | |
| } | |
| chainListener, err := listener.NewChainListener(chainClient, c.db, common.HexToAddress(cctpAddr), initialBlock, c.handler) | |
| if err != nil { | |
| return fmt.Errorf("could not get chain listener: %w", err) | |
| } | |
| c.chainListeners[chainID] = chainListener | |
| } | |
| return nil | |
| func (c *rebalanceManagerCCTP) initListeners(ctx context.Context) (err error) { | |
| // Consider using errgroup for parallel initialization of listeners | |
| for chainID := range c.cfg.GetChains() { | |
| cctpAddr, err := c.cfg.GetCCTPAddress(chainID) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp address: %w", err) | |
| } | |
| chainClient, err := c.chainClient.GetClient(ctx, big.NewInt(int64(chainID))) | |
| if err != nil { | |
| return fmt.Errorf("could not get chain client: %w", err) | |
| } | |
| initialBlock, err := c.cfg.GetCCTPStartBlock(chainID) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp start block: %w", err) | |
| } | |
| chainListener, err := listener.NewChainListener(chainClient, c.db, common.HexToAddress(cctpAddr), initialBlock, c.handler) | |
| if err != nil { | |
| return fmt.Errorf("could not get chain listener: %w", err) | |
| } | |
| c.chainListeners[chainID] = chainListener | |
| } | |
| return nil | |
| } |
| func (c *rebalanceManagerCCTP) Execute(parentCtx context.Context, rebalance *RebalanceData) (err error) { | ||
| contract, ok := c.cctpContracts[rebalance.OriginMetadata.ChainID] | ||
| if !ok { | ||
| return fmt.Errorf("could not find cctp contract for chain %d", rebalance.OriginMetadata.ChainID) | ||
| } | ||
| ctx, span := c.handler.Tracer().Start(parentCtx, "rebalance.Execute", trace.WithAttributes( | ||
| attribute.Int("rebalance_origin", int(rebalance.OriginMetadata.ChainID)), | ||
| attribute.Int("rebalance_dest", int(rebalance.DestMetadata.ChainID)), | ||
| attribute.String("rebalance_amount", rebalance.Amount.String()), | ||
| )) | ||
| defer func(err error) { | ||
| metrics.EndSpanWithErr(span, err) | ||
| }(err) | ||
|
|
||
| // perform rebalance by calling sendCircleToken() | ||
| _, err = c.txSubmitter.SubmitTransaction(ctx, big.NewInt(int64(rebalance.OriginMetadata.ChainID)), func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) { | ||
| tx, err = contract.SendCircleToken( | ||
| transactor, | ||
| c.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) | ||
| } | ||
|
|
||
| // store the rebalance in the db | ||
| model := reldb.Rebalance{ | ||
| Origin: uint64(rebalance.OriginMetadata.ChainID), | ||
| Destination: uint64(rebalance.DestMetadata.ChainID), | ||
| OriginAmount: rebalance.Amount, | ||
| Status: reldb.RebalanceInitiated, | ||
| } | ||
| err = c.db.StoreRebalance(ctx, model) | ||
| if err != nil { | ||
| return fmt.Errorf("could not store rebalance: %w", err) | ||
| } | ||
| return nil |
There was a problem hiding this comment.
The Execute method is well-implemented, with clear error handling and transaction submission logic. However, the TODO comments on lines 161 and 162 indicate unfinished work related to transaction parameters. Address these TODOs to ensure the rebalance operation is fully functional and secure.
Ensure the transaction parameters are correctly set and remove the TODO comments after addressing them.
| // nolint:cyclop | ||
| func (c *rebalanceManagerCCTP) listen(parentCtx context.Context, chainID int) (err error) { | ||
| listener, ok := c.chainListeners[chainID] | ||
| if !ok { | ||
| return fmt.Errorf("could not find listener for chain %d", chainID) | ||
| } | ||
| ethClient, err := c.chainClient.GetClient(parentCtx, big.NewInt(int64(chainID))) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get chain client: %w", err) | ||
| } | ||
| cctpAddr := common.HexToAddress(c.cfg.Chains[chainID].CCTPAddress) | ||
| parser, err := cctp.NewSynapseCCTPEvents(cctpAddr, ethClient) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get cctp events: %w", err) | ||
| } | ||
|
|
||
| err = listener.Listen(parentCtx, func(parentCtx context.Context, log types.Log) (err error) { | ||
| ctx, span := c.handler.Tracer().Start(parentCtx, "rebalance.Listen", trace.WithAttributes( | ||
| attribute.Int(metrics.ChainID, chainID), | ||
| )) | ||
| defer func(err error) { | ||
| metrics.EndSpanWithErr(span, err) | ||
| }(err) | ||
|
|
||
| switch log.Topics[0] { | ||
| case cctp.CircleRequestSentTopic: | ||
| parsedEvent, err := parser.ParseCircleRequestSent(log) | ||
| if err != nil { | ||
| logger.Warnf("could not parse circle request sent: %w", err) | ||
| return nil | ||
| } | ||
| if parsedEvent.Sender != c.relayerAddress { | ||
| return nil | ||
| } | ||
| span.SetAttributes( | ||
| attribute.String("log_type", "CircleRequestSent"), | ||
| attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), | ||
| ) | ||
| origin := uint64(chainID) | ||
| err = c.db.UpdateRebalanceStatus(ctx, parsedEvent.RequestID, &origin, reldb.RebalancePending) | ||
| if err != nil { | ||
| logger.Warnf("could not update rebalance status: %w", err) | ||
| return nil | ||
| } | ||
| case cctp.CircleRequestFulfilledTopic: | ||
| parsedEvent, err := parser.ParseCircleRequestFulfilled(log) | ||
| if err != nil { | ||
| logger.Warnf("could not parse circle request fulfilled: %w", err) | ||
| return nil | ||
| } | ||
| if parsedEvent.Recipient != c.relayerAddress { | ||
| return nil | ||
| } | ||
| span.SetAttributes( | ||
| attribute.String("log_type", "CircleRequestFulfilled"), | ||
| attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), | ||
| ) | ||
| err = c.db.UpdateRebalanceStatus(parentCtx, parsedEvent.RequestID, nil, reldb.RebalanceCompleted) | ||
| if err != nil { | ||
| logger.Warnf("could not update rebalance status: %w", err) | ||
| return nil | ||
| } | ||
| default: | ||
| logger.Warnf("unknown event %s", log.Topics[0]) | ||
| } | ||
| return nil | ||
| }) | ||
| if err != nil { | ||
| return fmt.Errorf("could not listen to contract: %w", err) | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
The listen method correctly handles CCTP events and updates rebalance statuses based on those events. However, the logging for unknown events (line 250) could be improved by including more context about the event for easier debugging.
- logger.Warnf("unknown event %s", log.Topics[0])
+ logger.Warnf("unknown event %s for chain %d", log.Topics[0], chainID)Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| // nolint:cyclop | |
| func (c *rebalanceManagerCCTP) listen(parentCtx context.Context, chainID int) (err error) { | |
| listener, ok := c.chainListeners[chainID] | |
| if !ok { | |
| return fmt.Errorf("could not find listener for chain %d", chainID) | |
| } | |
| ethClient, err := c.chainClient.GetClient(parentCtx, big.NewInt(int64(chainID))) | |
| if err != nil { | |
| return fmt.Errorf("could not get chain client: %w", err) | |
| } | |
| cctpAddr := common.HexToAddress(c.cfg.Chains[chainID].CCTPAddress) | |
| parser, err := cctp.NewSynapseCCTPEvents(cctpAddr, ethClient) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp events: %w", err) | |
| } | |
| err = listener.Listen(parentCtx, func(parentCtx context.Context, log types.Log) (err error) { | |
| ctx, span := c.handler.Tracer().Start(parentCtx, "rebalance.Listen", trace.WithAttributes( | |
| attribute.Int(metrics.ChainID, chainID), | |
| )) | |
| defer func(err error) { | |
| metrics.EndSpanWithErr(span, err) | |
| }(err) | |
| switch log.Topics[0] { | |
| case cctp.CircleRequestSentTopic: | |
| parsedEvent, err := parser.ParseCircleRequestSent(log) | |
| if err != nil { | |
| logger.Warnf("could not parse circle request sent: %w", err) | |
| return nil | |
| } | |
| if parsedEvent.Sender != c.relayerAddress { | |
| return nil | |
| } | |
| span.SetAttributes( | |
| attribute.String("log_type", "CircleRequestSent"), | |
| attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), | |
| ) | |
| origin := uint64(chainID) | |
| err = c.db.UpdateRebalanceStatus(ctx, parsedEvent.RequestID, &origin, reldb.RebalancePending) | |
| if err != nil { | |
| logger.Warnf("could not update rebalance status: %w", err) | |
| return nil | |
| } | |
| case cctp.CircleRequestFulfilledTopic: | |
| parsedEvent, err := parser.ParseCircleRequestFulfilled(log) | |
| if err != nil { | |
| logger.Warnf("could not parse circle request fulfilled: %w", err) | |
| return nil | |
| } | |
| if parsedEvent.Recipient != c.relayerAddress { | |
| return nil | |
| } | |
| span.SetAttributes( | |
| attribute.String("log_type", "CircleRequestFulfilled"), | |
| attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), | |
| ) | |
| err = c.db.UpdateRebalanceStatus(parentCtx, parsedEvent.RequestID, nil, reldb.RebalanceCompleted) | |
| if err != nil { | |
| logger.Warnf("could not update rebalance status: %w", err) | |
| return nil | |
| } | |
| default: | |
| logger.Warnf("unknown event %s", log.Topics[0]) | |
| } | |
| return nil | |
| }) | |
| if err != nil { | |
| return fmt.Errorf("could not listen to contract: %w", err) | |
| } | |
| return nil | |
| } | |
| // nolint:cyclop | |
| func (c *rebalanceManagerCCTP) listen(parentCtx context.Context, chainID int) (err error) { | |
| listener, ok := c.chainListeners[chainID] | |
| if !ok { | |
| return fmt.Errorf("could not find listener for chain %d", chainID) | |
| } | |
| ethClient, err := c.chainClient.GetClient(parentCtx, big.NewInt(int64(chainID))) | |
| if err != nil { | |
| return fmt.Errorf("could not get chain client: %w", err) | |
| } | |
| cctpAddr := common.HexToAddress(c.cfg.Chains[chainID].CCTPAddress) | |
| parser, err := cctp.NewSynapseCCTPEvents(cctpAddr, ethClient) | |
| if err != nil { | |
| return fmt.Errorf("could not get cctp events: %w", err) | |
| } | |
| err = listener.Listen(parentCtx, func(parentCtx context.Context, log types.Log) (err error) { | |
| ctx, span := c.handler.Tracer().Start(parentCtx, "rebalance.Listen", trace.WithAttributes( | |
| attribute.Int(metrics.ChainID, chainID), | |
| )) | |
| defer func(err error) { | |
| metrics.EndSpanWithErr(span, err) | |
| }(err) | |
| switch log.Topics[0] { | |
| case cctp.CircleRequestSentTopic: | |
| parsedEvent, err := parser.ParseCircleRequestSent(log) | |
| if err != nil { | |
| logger.Warnf("could not parse circle request sent: %w", err) | |
| return nil | |
| } | |
| if parsedEvent.Sender != c.relayerAddress { | |
| return nil | |
| } | |
| span.SetAttributes( | |
| attribute.String("log_type", "CircleRequestSent"), | |
| attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), | |
| ) | |
| origin := uint64(chainID) | |
| err = c.db.UpdateRebalanceStatus(ctx, parsedEvent.RequestID, &origin, reldb.RebalancePending) | |
| if err != nil { | |
| logger.Warnf("could not update rebalance status: %w", err) | |
| return nil | |
| } | |
| case cctp.CircleRequestFulfilledTopic: | |
| parsedEvent, err := parser.ParseCircleRequestFulfilled(log) | |
| if err != nil { | |
| logger.Warnf("could not parse circle request fulfilled: %w", err) | |
| return nil | |
| } | |
| if parsedEvent.Recipient != c.relayerAddress { | |
| return nil | |
| } | |
| span.SetAttributes( | |
| attribute.String("log_type", "CircleRequestFulfilled"), | |
| attribute.String("request_id", hexutil.Encode(parsedEvent.RequestID[:])), | |
| ) | |
| err = c.db.UpdateRebalanceStatus(parentCtx, parsedEvent.RequestID, nil, reldb.RebalanceCompleted) | |
| if err != nil { | |
| logger.Warnf("could not update rebalance status: %w", err) | |
| return nil | |
| } | |
| default: | |
| logger.Warnf("unknown event %s for chain %d", log.Topics[0], chainID) | |
| } | |
| return nil | |
| }) | |
| if err != nil { | |
| return fmt.Errorf("could not listen to contract: %w", err) | |
| } | |
| return nil | |
| } |
There was a problem hiding this comment.
Review Status
Actionable comments generated: 5
Configuration used: .coderabbit.yaml
Files selected for processing (2)
- services/rfq/relayer/inventory/manager.go (14 hunks)
- services/rfq/relayer/inventory/rebalance.go (1 hunks)
Files skipped from review as they are similar to previous changes (1)
- services/rfq/relayer/inventory/rebalance.go
Additional comments: 1
services/rfq/relayer/inventory/manager.go (1)
- 66-77: > 📝 NOTE
This review was outside the diff hunks and was mapped to the diff hunk with the greatest overlap. Original lines [55-74]
The
inventoryManagerImplstruct has been updated with new fields:txSubmitter,rebalanceManagers, andchainID. These changes are essential for supporting the new rebalancing functionality. Ensure that these fields are properly initialized and used throughout the implementation to avoid any potential issues related to uninitialized fields or incorrect usage.
| // Start starts the inventory manager. | ||
| Start(ctx context.Context) (err error) | ||
| // GetCommittableBalance gets the total balance available for quotes | ||
| // this does not include on-chain balances committed in previous quotes that may be | ||
| // refunded in the event of a revert. | ||
| GetCommittableBalance(ctx context.Context, chainID int, token common.Address, options ...BalanceFetchArgOption) (*big.Int, error) | ||
| // 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 |
There was a problem hiding this comment.
The addition of the Start and Rebalance methods, alongside the modification to the ApproveAllTokens method, significantly enhances the inventory manager's functionality. It's crucial to ensure these methods are thoroughly tested, especially the Rebalance method, due to its critical role in maintaining optimal inventory levels. Consider adding unit tests to cover various scenarios and edge cases for these methods.
Would you like assistance in creating unit tests for these methods?
| // NewInventoryManager creates a new inventory manager. | ||
| // TODO: too many args here. | ||
| // | ||
| //nolint:gocognit | ||
| 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) { | ||
| rebalanceMethods, err := cfg.GetRebalanceMethods() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not get rebalance methods: %w", err) | ||
| } | ||
| rebalanceManagers := make(map[relconfig.RebalanceMethod]RebalanceManager) | ||
| for method := range rebalanceMethods { | ||
| //nolint:exhaustive | ||
| switch method { | ||
| case relconfig.RebalanceMethodCCTP: | ||
| rebalanceManagers[method] = newRebalanceManagerCCTP(cfg, handler, clientFetcher, txSubmitter, relayer, db) | ||
| default: | ||
| return nil, fmt.Errorf("unsupported rebalance method: %s", method) | ||
| } | ||
| } | ||
|
|
||
| i := inventoryManagerImpl{ | ||
| relayerAddress: relayer, | ||
| handler: handler, | ||
| cfg: cfg, | ||
| chainClient: clientFetcher, | ||
| db: db, | ||
| relayerAddress: relayer, | ||
| handler: handler, | ||
| cfg: cfg, | ||
| chainClient: clientFetcher, | ||
| txSubmitter: txSubmitter, | ||
| rebalanceManagers: rebalanceManagers, | ||
| db: db, | ||
| } | ||
|
|
||
| err := i.initializeTokens(ctx, cfg) | ||
| err = i.initializeTokens(ctx, cfg) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not initialize tokens: %w", err) | ||
| } | ||
|
|
||
| // TODO: move | ||
| go func() { | ||
| return &i, nil |
There was a problem hiding this comment.
The NewInventoryManager function has been significantly modified to accommodate the new rebalancing functionality. While the changes are necessary, the function now has a high number of parameters, which could impact readability and maintainability. Consider refactoring this function to reduce its complexity, possibly by introducing a struct to encapsulate some of the parameters or by using a builder pattern.
- 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) {
+ // Consider introducing a struct to encapsulate some parameters or using a builder pattern.Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| // NewInventoryManager creates a new inventory manager. | |
| // TODO: too many args here. | |
| // | |
| //nolint:gocognit | |
| 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) { | |
| rebalanceMethods, err := cfg.GetRebalanceMethods() | |
| if err != nil { | |
| return nil, fmt.Errorf("could not get rebalance methods: %w", err) | |
| } | |
| rebalanceManagers := make(map[relconfig.RebalanceMethod]RebalanceManager) | |
| for method := range rebalanceMethods { | |
| //nolint:exhaustive | |
| switch method { | |
| case relconfig.RebalanceMethodCCTP: | |
| rebalanceManagers[method] = newRebalanceManagerCCTP(cfg, handler, clientFetcher, txSubmitter, relayer, db) | |
| default: | |
| return nil, fmt.Errorf("unsupported rebalance method: %s", method) | |
| } | |
| } | |
| i := inventoryManagerImpl{ | |
| relayerAddress: relayer, | |
| handler: handler, | |
| cfg: cfg, | |
| chainClient: clientFetcher, | |
| db: db, | |
| relayerAddress: relayer, | |
| handler: handler, | |
| cfg: cfg, | |
| chainClient: clientFetcher, | |
| txSubmitter: txSubmitter, | |
| rebalanceManagers: rebalanceManagers, | |
| db: db, | |
| } | |
| err := i.initializeTokens(ctx, cfg) | |
| err = i.initializeTokens(ctx, cfg) | |
| if err != nil { | |
| return nil, fmt.Errorf("could not initialize tokens: %w", err) | |
| } | |
| // TODO: move | |
| go func() { | |
| return &i, nil | |
| // NewInventoryManager creates a new inventory manager. | |
| // TODO: too many args here. | |
| // | |
| //nolint:gocognit | |
| // Consider introducing a struct to encapsulate some parameters or using a builder pattern. | |
| 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) { | |
| rebalanceMethods, err := cfg.GetRebalanceMethods() | |
| if err != nil { | |
| return nil, fmt.Errorf("could not get rebalance methods: %w", err) | |
| } | |
| rebalanceManagers := make(map[relconfig.RebalanceMethod]RebalanceManager) | |
| for method := range rebalanceMethods { | |
| //nolint:exhaustive | |
| switch method { | |
| case relconfig.RebalanceMethodCCTP: | |
| rebalanceManagers[method] = newRebalanceManagerCCTP(cfg, handler, clientFetcher, txSubmitter, relayer, db) | |
| default: | |
| return nil, fmt.Errorf("unsupported rebalance method: %s", method) | |
| } | |
| } | |
| i := inventoryManagerImpl{ | |
| relayerAddress: relayer, | |
| handler: handler, | |
| cfg: cfg, | |
| chainClient: clientFetcher, | |
| txSubmitter: txSubmitter, | |
| rebalanceManagers: rebalanceManagers, | |
| db: db, | |
| } | |
| err = i.initializeTokens(ctx, cfg) | |
| if err != nil { | |
| return nil, fmt.Errorf("could not initialize tokens: %w", err) | |
| } | |
| return &i, nil |
| func (i *inventoryManagerImpl) Start(ctx context.Context) error { | ||
| g, _ := errgroup.WithContext(ctx) | ||
| for _, rebalanceManager := range i.rebalanceManagers { | ||
| rebalanceManager := rebalanceManager | ||
| g.Go(func() error { | ||
| err := rebalanceManager.Start(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("could not start rebalance manager: %w", err) | ||
| } | ||
| return nil | ||
| }) | ||
| } | ||
|
|
||
| // continuously refresh balances | ||
| g.Go(func() error { | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-time.After(defaultPollPeriod * time.Second): | ||
| return fmt.Errorf("context canceled: %w", ctx.Err()) | ||
| case <-time.After(250 * time.Millisecond): | ||
| // this returning an error isn't really possible unless a config error happens | ||
| // TODO: need better error handling. | ||
| err = i.refreshBalances(ctx) | ||
| err := i.refreshBalances(ctx) | ||
| if err != nil { | ||
| logger.Errorf("could not refresh balances") | ||
| return | ||
| //nolint:nilerr | ||
| return nil | ||
| } | ||
| } | ||
| } | ||
| }() | ||
| }) | ||
|
|
||
| return &i, nil | ||
| // continuously check for rebalances | ||
| rebalanceInterval := i.cfg.GetRebalanceInterval() | ||
| if rebalanceInterval > 0 { | ||
| g.Go(func() error { | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return fmt.Errorf("context canceled: %w", ctx.Err()) | ||
| case <-time.After(rebalanceInterval): | ||
| err := i.refreshBalances(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("could not refresh balances: %w", err) | ||
| } | ||
| for chainID, chainConfig := range i.cfg.Chains { | ||
| for tokenName, tokenConfig := range chainConfig.Tokens { | ||
| err = i.Rebalance(ctx, chainID, common.HexToAddress(tokenConfig.Address)) | ||
| if err != nil { | ||
| logger.Errorf("could not rebalance %s on chain %d: %v", tokenName, chainID, err) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| err := g.Wait() | ||
| if err != nil { | ||
| return fmt.Errorf("error starting inventory manager: %w", err) | ||
| } | ||
| return nil |
There was a problem hiding this comment.
The implementation of the Start method is comprehensive, handling the start-up of rebalance managers and continuously refreshing balances and checking for rebalances. However, the error handling within the goroutines could be improved by logging the errors or handling them in a way that doesn't potentially swallow them (e.g., the nil return in line 218). Consider implementing a more robust error handling strategy.
- //nolint:nilerr
- return nil
+ // Implement error handling strategy here.Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.
| func (i *inventoryManagerImpl) Start(ctx context.Context) error { | |
| g, _ := errgroup.WithContext(ctx) | |
| for _, rebalanceManager := range i.rebalanceManagers { | |
| rebalanceManager := rebalanceManager | |
| g.Go(func() error { | |
| err := rebalanceManager.Start(ctx) | |
| if err != nil { | |
| return fmt.Errorf("could not start rebalance manager: %w", err) | |
| } | |
| return nil | |
| }) | |
| } | |
| // continuously refresh balances | |
| g.Go(func() error { | |
| for { | |
| select { | |
| case <-ctx.Done(): | |
| return | |
| case <-time.After(defaultPollPeriod * time.Second): | |
| return fmt.Errorf("context canceled: %w", ctx.Err()) | |
| case <-time.After(250 * time.Millisecond): | |
| // this returning an error isn't really possible unless a config error happens | |
| // TODO: need better error handling. | |
| err = i.refreshBalances(ctx) | |
| err := i.refreshBalances(ctx) | |
| if err != nil { | |
| logger.Errorf("could not refresh balances") | |
| return | |
| //nolint:nilerr | |
| return nil | |
| } | |
| } | |
| } | |
| }() | |
| }) | |
| return &i, nil | |
| // continuously check for rebalances | |
| rebalanceInterval := i.cfg.GetRebalanceInterval() | |
| if rebalanceInterval > 0 { | |
| g.Go(func() error { | |
| for { | |
| select { | |
| case <-ctx.Done(): | |
| return fmt.Errorf("context canceled: %w", ctx.Err()) | |
| case <-time.After(rebalanceInterval): | |
| err := i.refreshBalances(ctx) | |
| if err != nil { | |
| return fmt.Errorf("could not refresh balances: %w", err) | |
| } | |
| for chainID, chainConfig := range i.cfg.Chains { | |
| for tokenName, tokenConfig := range chainConfig.Tokens { | |
| err = i.Rebalance(ctx, chainID, common.HexToAddress(tokenConfig.Address)) | |
| if err != nil { | |
| logger.Errorf("could not rebalance %s on chain %d: %v", tokenName, chainID, err) | |
| } | |
| } | |
| } | |
| } | |
| } | |
| }) | |
| } | |
| err := g.Wait() | |
| if err != nil { | |
| return fmt.Errorf("error starting inventory manager: %w", err) | |
| } | |
| return nil | |
| // Implement error handling strategy here. |
| } | ||
|
|
||
| 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) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not get erc20: %w", err) | ||
| } | ||
|
|
||
| approveAmount, err := erc20.Approve(transactor, common.HexToAddress(i.cfg.Chains[chainID].Bridge), abi.MaxInt256) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not approve: %w", err) | ||
| } | ||
| // approve RFQ contract. | ||
| // Note: in the case where submitter hasn't finished from last boot, | ||
| // this will double submit approvals unfortunately. | ||
| if address != chain.EthAddress && token.StartAllowanceRFQ.Cmp(big.NewInt(0)) == 0 { | ||
| tokenAddr := address // capture func literal | ||
| contractAddr, err := i.cfg.GetRFQAddress(chainID) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get RFQ address: %w", err) | ||
| } | ||
| err = i.approve(ctx, tokenAddr, common.HexToAddress(contractAddr), backendClient) | ||
| if err != nil { | ||
| return fmt.Errorf("could not approve RFQ contract: %w", err) | ||
| } | ||
| } | ||
|
|
||
| return approveAmount, nil | ||
| }) | ||
| // approve CCTP contract | ||
| if address != chain.EthAddress && token.StartAllowanceCCTP.Cmp(big.NewInt(0)) == 0 { | ||
| tokenAddr := address // capture func literal | ||
| contractAddr, err := i.cfg.GetCCTPAddress(chainID) | ||
| if err != nil { | ||
| return fmt.Errorf("could not submit approval: %w", err) | ||
| return fmt.Errorf("could not get CCTP address: %w", err) | ||
| } | ||
| err = i.approve(ctx, tokenAddr, common.HexToAddress(contractAddr), backendClient) | ||
| if err != nil { | ||
| return fmt.Errorf("could not approve CCTP contract: %w", err) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // approve submits an ERC20 approval for a given token and contract address. | ||
| func (i *inventoryManagerImpl) approve(ctx context.Context, tokenAddr, contractAddr common.Address, backendClient client.EVM) (err error) { | ||
| erc20, err := ierc20.NewIERC20(tokenAddr, backendClient) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get erc20: %w", err) | ||
| } | ||
| chainID, err := backendClient.ChainID(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get chain id: %w", err) | ||
| } | ||
|
|
||
| _, err = i.txSubmitter.SubmitTransaction(ctx, chainID, func(transactor *bind.TransactOpts) (tx *types.Transaction, err error) { | ||
| tx, err = erc20.Approve(transactor, contractAddr, abi.MaxInt256) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not approve: %w", err) | ||
| } | ||
| return tx, nil | ||
| }) | ||
| if err != nil { | ||
| return fmt.Errorf("could not submit approval: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // HasSufficientGas checks if there is sufficient gas for a given route. | ||
| func (i *inventoryManagerImpl) HasSufficientGas(ctx context.Context, origin, dest int) (sufficient bool, err error) { | ||
| gasThresh, err := i.cfg.GetMinGasToken(dest) |
There was a problem hiding this comment.
📝 NOTE
This review was outside the diff hunks and was mapped to the diff hunk with the greatest overlap. Original lines [260-296]
The ApproveAllTokens method has been enhanced to manage approvals for both RFQ and CCTP bridges. This is a crucial update for supporting the new rebalancing functionality. Ensure that the error handling is robust, especially considering the potential for double submissions of approvals as noted in the comments. Additionally, consider optimizing the approval process to avoid unnecessary transactions if possible.
Consider adding logic to check existing allowances before attempting to set new approvals, to minimize unnecessary transactions.
| // 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(parentCtx context.Context, chainID int, token common.Address) error { | ||
| // evaluate the rebalance method | ||
| 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 | ||
| } | ||
| ctx, span := i.handler.Tracer().Start(parentCtx, "Rebalance", trace.WithAttributes( | ||
| attribute.Int(metrics.ChainID, chainID), | ||
| attribute.String("token", token.Hex()), | ||
| attribute.String("rebalance_method", method.String()), | ||
| )) | ||
| defer func(err error) { | ||
| metrics.EndSpanWithErr(span, err) | ||
| }(err) | ||
|
|
||
| // build the rebalance action | ||
| rebalance, err := getRebalance(span, i.cfg, i.tokens, chainID, token) | ||
| if err != nil { | ||
| return fmt.Errorf("could not get rebalance: %w", err) | ||
| } | ||
| if rebalance == nil { | ||
| return nil | ||
| } | ||
| span.SetAttributes( | ||
| attribute.String("rebalance_origin", strconv.Itoa(rebalance.OriginMetadata.ChainID)), | ||
| attribute.String("rebalance_dest", strconv.Itoa(rebalance.DestMetadata.ChainID)), | ||
| attribute.String("rebalance_amount", rebalance.Amount.String()), | ||
| ) | ||
|
|
||
| // make sure there are no pending rebalances that touch the given path | ||
| pending, err := i.db.HasPendingRebalance(ctx, uint64(rebalance.OriginMetadata.ChainID), uint64(rebalance.DestMetadata.ChainID)) | ||
| if err != nil { | ||
| return fmt.Errorf("could not check pending rebalance: %w", err) | ||
| } | ||
| span.SetAttributes(attribute.Bool("rebalance_pending", pending)) | ||
| if pending { | ||
| return nil | ||
| } | ||
|
|
||
| // execute the rebalance | ||
| manager, ok := i.rebalanceManagers[method] | ||
| if !ok { | ||
| return fmt.Errorf("no rebalance manager for method: %s", method) | ||
| } | ||
| err = manager.Execute(ctx, rebalance) | ||
| if err != nil { | ||
| return fmt.Errorf("could not execute rebalance: %w", err) | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
The Rebalance method implements the core functionality of token rebalancing. Given its impact on the inventory management system's reliability and accuracy, it's crucial to ensure this method is thoroughly tested. The method involves several steps, including evaluating the rebalance method, building the rebalance action, checking for pending rebalances, and executing the rebalance. Each of these steps has potential failure points that need to be carefully handled.
Would you like assistance in creating unit tests for the Rebalance method?
Description
Adds a
Rebalance()func to theinventory.Managerinterface, which is called upon witnessing aClaimCompletedevent. If configured, this triggers a CCTP bridge to rebalance the inventory.Rebalances are triggered using two new parameters:
MaintenanceBalancePct: if the balance for a given chain goes below this value, a rebalance is triggered with this chain as the destinationInitialBalancePct: the balance threshold that will be targeted for the origin of a rebalance transactionA
RebalanceMethodenum is also added- currently onlyRebalanceMethodCCTPis implemented but this lays the groundwork for other future methods.Summary by CodeRabbit