feat: support Stork oracle #13
Conversation
WalkthroughThe recent changes enhance the Injective Price Oracle by integrating support for Stork feeds. New configuration options and methods have been introduced for managing WebSocket connections and processing Stork price data. The Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant CLI
participant OracleService
participant StorkWebSocket
User->>CLI: Provide Stork feed configurations
CLI->>OracleService: Initialize Oracle with Stork configurations
OracleService->>StorkWebSocket: Connect and retrieve price data
StorkWebSocket-->>OracleService: Send price data
OracleService->>CLI: Provide updated prices
CLI-->>User: Display updated prices
Poem
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 Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
Files selected for processing (9)
- cmd/injective-price-oracle/options.go (2 hunks)
- cmd/injective-price-oracle/oracle.go (3 hunks)
- examples/stork.toml (1 hunks)
- go.mod (2 hunks)
- oracle/feed_binance.go (1 hunks)
- oracle/feed_dynamic.go (1 hunks)
- oracle/feed_stork.go (1 hunks)
- oracle/models.go (1 hunks)
- oracle/service.go (14 hunks)
Files skipped from review due to trivial changes (2)
- examples/stork.toml
- oracle/models.go
Additional comments not posted (24)
cmd/injective-price-oracle/options.go (1)
Line range hint
139-156:
LGTM! Verify the usage of the new parameter.The addition of the
storkFeedsDirparameter looks good.However, ensure that the new parameter is correctly used in the codebase.
Verification successful
Verified: Proper usage of the
storkFeedsDirparameter.The
storkFeedsDirparameter has been correctly integrated and utilized in the codebase.
- Initialization: The parameter is properly initialized in
initExternalFeedsOptions.- Usage: The parameter is used in a conditional check and a directory walk in
oracle.go.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the usage of the new parameter `storkFeedsDir`. # Test: Search for the usage of the parameter. Expect: Proper usage in the codebase. rg --type go $'storkFeedsDir'Length of output: 491
Script:
#!/bin/bash # Extract relevant lines around the usage of `storkFeedsDir` in oracle.go for review. rg --type go --context 5 'storkFeedsDir' cmd/injective-price-oracle/oracle.goLength of output: 834
cmd/injective-price-oracle/oracle.go (1)
251-251: LGTM!The code correctly passes the Stork feed configurations to the
NewServicefunction.oracle/feed_stork.go (9)
26-34: LGTM! Struct definition is correct.The
StorkFeedConfigstruct is well-defined with appropriate field tags for TOML unmarshalling.
39-51: LGTM! Struct definition is correct.The
storkPriceFeedstruct is well-defined with appropriate fields for the Stork price feed.
53-61: LGTM! Function implementation is correct.The
ParseStorkFeedConfigfunction is well-implemented with proper error handling.
63-113: LGTM! Function implementation is correct.The
NewStorkPriceFeedfunction is well-implemented with proper error handling and initialization logic.
136-201: LGTM! Function implementation is correct.The
PullAssetPairfunction is well-implemented with proper error handling and WebSocket communication logic.
204-209: LGTM! Function implementation is correct.The
PullPricefunction is well-implemented as a placeholder that returns a zero price.
211-222: LGTM! Function implementation is correct.The
ConvertDataToAssetPairfunction is well-implemented with proper conversion logic.
224-236: LGTM! Function implementation is correct.The
ConvertSignedPricefunction is well-implemented with proper conversion logic.
238-245: LGTM! Function implementation is correct.The
CombineSignatureToStringfunction is well-implemented with proper string manipulation logic.go.mod (5)
3-5: LGTM! Go version update is correct.The Go version has been updated from 1.21 to 1.22 to leverage new language features and improvements.
8-8: LGTM! New dependency is correct.The new dependency
cosmossdk.io/mathappears to be necessary for mathematical operations.
10-10: LGTM! Dependency update is correct.The dependency
github.com/InjectiveLabs/sdk-gohas been updated to versionv1.50.2to supportMsgRelayStorkMessage.
17-17: LGTM! New dependency is correct.The new dependency
github.com/pelletier/go-toml/v2appears to be necessary for TOML parsing.
226-227: LGTM! Dependency replacement is correct.The dependency
cosmossdk.io/storehas been replaced with a specific version to ensure compatibility with the InjectiveLabs cosmos-sdk version.oracle/service.go (8)
37-38: LGTM! Interface update is correct.The
PricePullerinterface has been updated to include thePullAssetPairmethod to support the new Stork feed provider.
55-55: LGTM! Struct update is correct.The
oracleSvcstruct has been updated to include a map of supported Stork feeds to manage the new Stork feed configurations.
82-82: LGTM! New constant is correct.The new constant
FeedProviderStorkis necessary to identify the Stork feed provider.
Line range hint
146-194:
LGTM! Function update is correct.The
NewServicefunction has been updated to accept Stork feed configurations to initialize the service with Stork feed configurations.
240-293: LGTM! Function update is correct.The
processSetPriceFeedfunction has been updated to handle Stork oracle types to process Stork-specific logic for pulling asset pairs.
370-387: LGTM! New function is correct.The new function
composeStorkOracleMsgsis necessary to handle the composition of Stork-specific messages.
388-392: LGTM! Function update is correct.The
composeMsgsfunction has been updated to include Stork oracle messages to compose messages for Stork oracle types.
454-469: LGTM! Function update is correct.The
commitSetPricesfunction has been updated to handle Stork oracle types to process and commit prices for Stork oracle types.
oracle/feed_binance.go
Outdated
| func (f *binancePriceFeed) PullAssetPair(ctx context.Context) (assetPair oracletypes.AssetPair, err error){ | ||
| return oracletypes.AssetPair{}, nil | ||
| } |
There was a problem hiding this comment.
Incomplete Implementation: Add logic to PullAssetPair.
The method PullAssetPair currently returns an empty AssetPair object. Implement the necessary logic to return actual asset pair data.
- return oracletypes.AssetPair{}, nil
+ // TODO: Implement the logic to fetch and return the actual asset pair data.Committable suggestion was skipped due to low confidence.
oracle/feed_dynamic.go
Outdated
| func (f *dynamicPriceFeed) PullAssetPair(ctx context.Context) (assetPair oracletypes.AssetPair, err error){ | ||
| return oracletypes.AssetPair{}, nil | ||
| } |
There was a problem hiding this comment.
Incomplete Implementation: Add logic to PullAssetPair.
The method PullAssetPair currently returns an empty AssetPair object. Implement the necessary logic to return actual asset pair data.
- return oracletypes.AssetPair{}, nil
+ // TODO: Implement the logic to fetch and return the actual asset pair data.Committable suggestion was skipped due to low confidence.
There was a problem hiding this comment.
Actionable comments posted: 0
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
Files selected for processing (2)
- go.mod (2 hunks)
- oracle/service.go (14 hunks)
Files skipped from review as they are similar to previous changes (2)
- go.mod
- oracle/service.go
There was a problem hiding this comment.
Actionable comments posted: 0
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (2)
- examples/stork.toml (1 hunks)
- oracle/feed_stork.go (1 hunks)
Files skipped from review due to trivial changes (1)
- examples/stork.toml
Files skipped from review as they are similar to previous changes (1)
- oracle/feed_stork.go
ThanhNhann
force-pushed
the
nhan/stork-relayer
branch
2 times, most recently
from
f3b2eca to
ca84799
Compare
| return err | ||
| } | ||
|
|
||
| feedCfg, err := oracle.ParseStorkFeedConfig(cfgBody) |
There was a problem hiding this comment.
we should support price updates for multiple tickers
There was a problem hiding this comment.
Updated flow, can you plz recheck?
oracle/feed_binance.go
Outdated
| @@ -151,6 +151,10 @@ func (f *binancePriceFeed) PullPrice(ctx context.Context) ( | |||
| return priceResp.Price, nil | |||
| } | |||
|
|
|||
| func (f *binancePriceFeed) PullAssetPair(ctx context.Context) (assetPair oracletypes.AssetPair, err error){ | |||
There was a problem hiding this comment.
This is pretty hacky. Better to create a separate flow for Stork
There was a problem hiding this comment.
Could you check the new flow of Stork, currently I still use the base of the core logic
oracle/feed_dynamic.go
Outdated
| @@ -225,3 +225,7 @@ func (f *dynamicPriceFeed) PullPrice(ctx context.Context) ( | |||
|
|
|||
| return price, nil | |||
| } | |||
|
|
|||
| func (f *dynamicPriceFeed) PullAssetPair(ctx context.Context) (assetPair oracletypes.AssetPair, err error){ | |||
oracle/feed_stork.go
Outdated
| u, err := url.Parse(f.url) | ||
| if err != nil { | ||
| log.Fatal("Error parsing URL:", err) | ||
| return oracletypes.AssetPair{}, nil | ||
| } | ||
| header := http.Header{} | ||
| header.Add("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(f.header))) | ||
|
|
||
| dialer := websocket.DefaultDialer | ||
| dialer.EnableCompression = true | ||
|
|
||
| // Connect to the WebSocket server | ||
| conn, resp, err := dialer.Dial(u.String(), header) | ||
| if err != nil { | ||
| if resp != nil { | ||
| log.Printf("Handshake failed with status: %d\n", resp.StatusCode) | ||
| for k, v := range resp.Header { | ||
| log.Printf("%s: %v\n", k, v) | ||
| } | ||
| } | ||
| log.Fatal("Error connecting to WebSocket:", err) | ||
| return oracletypes.AssetPair{}, nil | ||
| } | ||
| defer conn.Close() |
There was a problem hiding this comment.
we shouldn't initiate and close a websocket connection every time we want to pull price.
There was a problem hiding this comment.
Moved it to here, so we don't need to make the connection every time we want to pull price
|
|
||
| signedPriceOfAssetPair.Signature = common.Hex2Bytes(signature) | ||
| signedPriceOfAssetPair.PublisherKey = signeds.PublisherKey | ||
| signedPriceOfAssetPair.Timestamp = ConvertTimestampToSecond(signeds.TimestampedSignature.Timestamp) |
There was a problem hiding this comment.
This code is quite fragile and assumes prices are ALWAYS signed by relayers in seconds, which may not be the case
There was a problem hiding this comment.
I think we have to have a step to convert the timestamp to second because our VerifyStorkMsgSignature needs that type of time to verify the signature, do you think where could we do that in our relayer or not?
There was a problem hiding this comment.
Actionable comments posted: 2
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (7)
- cmd/injective-price-oracle/options.go (3 hunks)
- cmd/injective-price-oracle/oracle.go (6 hunks)
- examples/stork.toml (1 hunks)
- oracle/feed_dynamic.go (1 hunks)
- oracle/feed_stork.go (1 hunks)
- oracle/models.go (1 hunks)
- oracle/service.go (13 hunks)
Files skipped from review as they are similar to previous changes (3)
- examples/stork.toml
- oracle/feed_dynamic.go
- oracle/service.go
Additional comments not posted (6)
oracle/models.go (1)
26-28: Review of the new fieldAssetPairsinPriceDatastruct.The addition of
AssetPairsto thePriceDatastruct is aligned with the PR's objective to support Stork Oracle. This field will store asset pairs related to price data, which is crucial for the new functionality.Ensure that the rest of the system properly utilizes this new field, especially in serialization and deserialization processes, as well as in any logic that processes
PriceDatainstances.cmd/injective-price-oracle/options.go (2)
Line range hint
139-156: Review of the new parameterstorkFeedsDirininitExternalFeedsOptionsfunction.The addition of
storkFeedsDirparameter is consistent with the PR's goal to support Stork feeds. It allows the application to accept a new command-line argument for specifying the directory containing Stork feed configurations.Ensure that this parameter is correctly handled wherever
initExternalFeedsOptionsis called, and verify that appropriate error handling is in place if the directory is not accessible or the files within are not correctly formatted.
204-219: New functioninitStorkOracleWebSocketreview.This function initializes WebSocket parameters for connecting to the Stork Oracle. It's a crucial addition for the real-time data feed functionality.
Check the usage of this function in the system to ensure the WebSocket connection is managed efficiently, including reconnection logic and error handling. Also, verify that the
websocketUrlandwebsocketHeaderare secured and validated appropriately to prevent security issues.oracle/feed_stork.go (2)
42-50: Review ofParseStorkFeedConfigfunction.This function parses Stork feed configurations from TOML format. It's well-implemented with proper error handling using
errors.Wrap, which enhances error messages with context.Ensure that the error messages are clear and provide sufficient information for troubleshooting configuration issues. Also, verify that this function is covered by unit tests, especially for edge cases in the TOML format.
52-100: Review ofNewStorkPriceFeedfunction.This function initializes a new Stork price feed based on the provided configuration. The error handling is robust, with specific messages for different failure scenarios, such as invalid pull intervals.
Consider adding more detailed logging at key steps within this function to aid in debugging and operational monitoring. Also, ensure that the minimum pull interval is justified by the system's requirements and that it doesn't lead to excessive load or data freshness issues.
cmd/injective-price-oracle/oracle.go (1)
297-326: Review ofConnectWebSocketfunction.This function manages the WebSocket connection to the Stork Oracle. The retry logic and error handling are well-implemented, ensuring robustness in maintaining the connection.
Ensure that the retry count and delay are configurable to allow fine-tuning based on operational experience. Also, verify that the security of the WebSocket connection is adequately addressed, particularly in the handling of authentication headers.
oracle/feed_stork.go
Outdated
| // PullAssetPair pulls asset pair for an asset id | ||
| func (f *storkPriceFeed) PullAssetPairs(ctx context.Context, conn *websocket.Conn) (assetPairs []*oracletypes.AssetPair, err error) { | ||
| metrics.ReportFuncCall(f.svcTags) | ||
| doneFn := metrics.ReportFuncTiming(f.svcTags) | ||
| defer doneFn() | ||
|
|
||
| err = conn.WriteMessage(websocket.TextMessage, []byte(f.message)) | ||
| if err != nil { | ||
| log.Fatal("Error writing message:", err) | ||
| return []*oracletypes.AssetPair{}, nil | ||
| } | ||
|
|
||
| var msgNeed []byte | ||
| count := 0 | ||
| for count < 2 { | ||
| _, message, err := conn.ReadMessage() | ||
| if err != nil { | ||
| log.Println("Error reading message:", err) | ||
| return []*oracletypes.AssetPair{}, nil | ||
|
|
||
| } | ||
| msgNeed = message | ||
| count += 1 | ||
| } | ||
|
|
||
| var msgResp messageResponse | ||
| if err = json.Unmarshal(msgNeed, &msgResp); err != nil { | ||
| return []*oracletypes.AssetPair{}, nil | ||
| } | ||
| assetIds := make([]string, 0) | ||
| for key := range msgResp.Data { | ||
| assetIds = append(assetIds, key) | ||
| } | ||
|
|
||
| for _, assetId := range assetIds { | ||
| pair := ConvertDataToAssetPair(msgResp.Data[assetId], assetId) | ||
| assetPairs = append(assetPairs, &pair) | ||
| } | ||
|
|
||
| return assetPairs, nil | ||
| } |
There was a problem hiding this comment.
Review of PullAssetPairs function.
This function is responsible for pulling asset pair data via a WebSocket connection. It includes metrics reporting and timing, which is good for performance monitoring.
However, the error handling could be improved. Currently, the function logs errors but still returns an empty slice and nil error, which might lead to silent failures in upstream logic. Consider modifying the error handling to return meaningful errors to the caller.
- return []*oracletypes.AssetPair{}, nil
+ return nil, errCommittable 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. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // PullAssetPair pulls asset pair for an asset id | |
| func (f *storkPriceFeed) PullAssetPairs(ctx context.Context, conn *websocket.Conn) (assetPairs []*oracletypes.AssetPair, err error) { | |
| metrics.ReportFuncCall(f.svcTags) | |
| doneFn := metrics.ReportFuncTiming(f.svcTags) | |
| defer doneFn() | |
| err = conn.WriteMessage(websocket.TextMessage, []byte(f.message)) | |
| if err != nil { | |
| log.Fatal("Error writing message:", err) | |
| return []*oracletypes.AssetPair{}, nil | |
| } | |
| var msgNeed []byte | |
| count := 0 | |
| for count < 2 { | |
| _, message, err := conn.ReadMessage() | |
| if err != nil { | |
| log.Println("Error reading message:", err) | |
| return []*oracletypes.AssetPair{}, nil | |
| } | |
| msgNeed = message | |
| count += 1 | |
| } | |
| var msgResp messageResponse | |
| if err = json.Unmarshal(msgNeed, &msgResp); err != nil { | |
| return []*oracletypes.AssetPair{}, nil | |
| } | |
| assetIds := make([]string, 0) | |
| for key := range msgResp.Data { | |
| assetIds = append(assetIds, key) | |
| } | |
| for _, assetId := range assetIds { | |
| pair := ConvertDataToAssetPair(msgResp.Data[assetId], assetId) | |
| assetPairs = append(assetPairs, &pair) | |
| } | |
| return assetPairs, nil | |
| } | |
| // PullAssetPair pulls asset pair for an asset id | |
| func (f *storkPriceFeed) PullAssetPairs(ctx context.Context, conn *websocket.Conn) (assetPairs []*oracletypes.AssetPair, err error) { | |
| metrics.ReportFuncCall(f.svcTags) | |
| doneFn := metrics.ReportFuncTiming(f.svcTags) | |
| defer doneFn() | |
| err = conn.WriteMessage(websocket.TextMessage, []byte(f.message)) | |
| if err != nil { | |
| log.Fatal("Error writing message:", err) | |
| return nil, err | |
| } | |
| var msgNeed []byte | |
| count := 0 | |
| for count < 2 { | |
| _, message, err := conn.ReadMessage() | |
| if err != nil { | |
| log.Println("Error reading message:", err) | |
| return nil, err | |
| } | |
| msgNeed = message | |
| count += 1 | |
| } | |
| var msgResp messageResponse | |
| if err = json.Unmarshal(msgNeed, &msgResp); err != nil { | |
| return nil, err | |
| } | |
| assetIds := make([]string, 0) | |
| for key := range msgResp.Data { | |
| assetIds = append(assetIds, key) | |
| } | |
| for _, assetId := range assetIds { | |
| pair := ConvertDataToAssetPair(msgResp.Data[assetId], assetId) | |
| assetPairs = append(assetPairs, &pair) | |
| } | |
| return assetPairs, nil | |
| } |
cmd/injective-price-oracle/oracle.go
Outdated
| storkFeedConfigs := make([]*oracle.StorkFeedConfig, 0, 10) | ||
| if len(*storkFeedsDir) > 0 { | ||
| err := filepath.WalkDir(*storkFeedsDir, func(path string, d fs.DirEntry, err error) error { | ||
| if err != nil { | ||
| return err | ||
| } else if d.IsDir() { | ||
| return nil | ||
| } else if filepath.Ext(path) != ".toml" { | ||
| return nil | ||
| } | ||
|
|
||
| cfgBody, err := ioutil.ReadFile(path) | ||
| if err != nil { | ||
| err = errors.Wrapf(err, "failed to read stork feed config") | ||
| return err | ||
| } | ||
|
|
||
| feedCfg, err := oracle.ParseStorkFeedConfig(cfgBody) | ||
| if err != nil { | ||
| log.WithError(err).WithFields(log.Fields{ | ||
| "filename": d.Name(), | ||
| }).Errorln("failed to parse stork feed config") | ||
| return nil | ||
| } | ||
|
|
||
| storkFeedConfigs = append(storkFeedConfigs, feedCfg) | ||
|
|
||
| return nil | ||
| }) | ||
|
|
||
| if err != nil { | ||
| err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *dynamicFeedsDir) | ||
| log.WithError(err).Fatalln("failed to load stork feeds") | ||
| return | ||
| } | ||
|
|
||
| log.Infof("found %d stork feed configs", len(storkFeedConfigs)) | ||
| } |
There was a problem hiding this comment.
Review of Stork feed configuration handling in oracleCmd function.
The handling of Stork feed configurations is well-integrated into the existing system. The use of filepath.WalkDir to load configurations from a directory is efficient and appropriate.
However, there's a copy-paste error in the error handling (line 251). The error message incorrectly references *dynamicFeedsDir instead of *storkFeedsDir. This needs correction to avoid confusion during troubleshooting.
- err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *dynamicFeedsDir)
+ err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *storkFeedsDir)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. Thoroughly test & benchmark the code to ensure it meets the requirements.
| storkFeedConfigs := make([]*oracle.StorkFeedConfig, 0, 10) | |
| if len(*storkFeedsDir) > 0 { | |
| err := filepath.WalkDir(*storkFeedsDir, func(path string, d fs.DirEntry, err error) error { | |
| if err != nil { | |
| return err | |
| } else if d.IsDir() { | |
| return nil | |
| } else if filepath.Ext(path) != ".toml" { | |
| return nil | |
| } | |
| cfgBody, err := ioutil.ReadFile(path) | |
| if err != nil { | |
| err = errors.Wrapf(err, "failed to read stork feed config") | |
| return err | |
| } | |
| feedCfg, err := oracle.ParseStorkFeedConfig(cfgBody) | |
| if err != nil { | |
| log.WithError(err).WithFields(log.Fields{ | |
| "filename": d.Name(), | |
| }).Errorln("failed to parse stork feed config") | |
| return nil | |
| } | |
| storkFeedConfigs = append(storkFeedConfigs, feedCfg) | |
| return nil | |
| }) | |
| if err != nil { | |
| err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *dynamicFeedsDir) | |
| log.WithError(err).Fatalln("failed to load stork feeds") | |
| return | |
| } | |
| log.Infof("found %d stork feed configs", len(storkFeedConfigs)) | |
| } | |
| err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *storkFeedsDir) |
There was a problem hiding this comment.
Actionable comments posted: 1
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (3)
- .env.example (1 hunks)
- examples/stork.toml (1 hunks)
- oracle/feed_stork.go (1 hunks)
Files skipped from review as they are similar to previous changes (2)
- examples/stork.toml
- oracle/feed_stork.go
Additional comments not posted (1)
.env.example (1)
31-31: Well-configured WebSocket URL for Stork oracle.The URL is correctly using the
wss://protocol for secure communication. However, consider adding a comment to remind users to update this URL for production environments.
| @@ -27,3 +27,6 @@ ORACLE_STATSD_ADDR="localhost:8125" | |||
| ORACLE_STATSD_STUCK_DUR="5m" | |||
| ORACLE_STATSD_MOCKING=false | |||
| ORACLE_STATSD_DISABLED=false | |||
|
|
|||
| STORK_WEBSOCKET_URL="wss://dev.api.stork-oracle.network/evm/subscribe" | |||
| STORK_WEBSOCKET_HEADER= | |||
There was a problem hiding this comment.
Provide guidance for STORK_WEBSOCKET_HEADER.
The variable is left empty, which might be confusing for users. Consider adding a comment or example value to clarify what kind of data might be expected here, such as authentication tokens or specific headers required by the Stork oracle.
|
I created this small PR let me know if it makes sense the improvs |
feat/stork-relayer improvs
There was a problem hiding this comment.
Actionable comments posted: 7
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (5)
- cmd/injective-price-oracle/oracle.go (9 hunks)
- oracle/feed_binance.go (2 hunks)
- oracle/feed_dynamic.go (2 hunks)
- oracle/feed_stork.go (1 hunks)
- oracle/service.go (16 hunks)
Files skipped from review due to trivial changes (1)
- oracle/feed_dynamic.go
Additional comments not posted (1)
oracle/feed_binance.go (1)
125-132: LGTM! Improved resource management withdefer.The replacement of
ioutil.ReadAllwithio.ReadAlland the use ofdeferfor closing the response body are good practices. They improve resource management and ensure the response body is closed properly.
oracle/feed_stork.go
Outdated
| err = conn.WriteMessage(websocket.TextMessage, []byte(f.message)) | ||
| if err != nil { | ||
| log.Infoln("Error writing message:", err) | ||
| return | ||
| } |
There was a problem hiding this comment.
Consider optimizing WebSocket message reading.
The current implementation reads messages in a loop with a fixed count. Consider using a more robust mechanism to determine when to stop reading, such as a specific message type or content.
| func ConvertTimestampToSecond(timestamp uint64) uint64 { | ||
| switch { | ||
| // nanosecond | ||
| case timestamp > 1e18: | ||
| return timestamp / uint64(1_000_000_000) | ||
| // microsecond | ||
| case timestamp > 1e15: | ||
| return timestamp / uint64(1_000_000) | ||
| // millisecond | ||
| case timestamp > 1e12: | ||
| return timestamp / uint64(1_000) | ||
| // second | ||
| default: | ||
| return timestamp | ||
| } | ||
| } |
There was a problem hiding this comment.
Optimize timestamp conversion logic.
The ConvertTimestampToSecond function uses a switch-case to convert timestamps. Consider using a more efficient approach, such as logarithmic checks, to determine the scale.
oracle/feed_stork.go
Outdated
| // PullAssetPairs pulls asset pair for an asset id | ||
| func (f *storkPriceFeed) PullAssetPairs(conn *websocket.Conn) (assetPairs []*oracletypes.AssetPair, err error) { | ||
| metrics.ReportFuncCall(f.svcTags) | ||
| doneFn := metrics.ReportFuncTiming(f.svcTags) | ||
| defer doneFn() | ||
|
|
||
| err = conn.WriteMessage(websocket.TextMessage, []byte(f.message)) | ||
| if err != nil { | ||
| log.Infoln("Error writing message:", err) | ||
| return | ||
| } | ||
|
|
||
| var msgNeed []byte | ||
| count := 0 | ||
| for count < 2 { | ||
| _, message, err := conn.ReadMessage() | ||
| if err != nil { | ||
| log.Infoln("Error reading message:", err) | ||
| return | ||
|
|
||
| } | ||
| msgNeed = message | ||
| count += 1 | ||
| } | ||
|
|
||
| var msgResp messageResponse | ||
| if err = json.Unmarshal(msgNeed, &msgResp); err != nil { | ||
| return | ||
| } | ||
|
|
||
| assetIds := make([]string, 0) | ||
| for key := range msgResp.Data { | ||
| assetIds = append(assetIds, key) | ||
| } | ||
|
|
||
| for _, assetId := range assetIds { | ||
| pair := ConvertDataToAssetPair(msgResp.Data[assetId], assetId) | ||
| assetPairs = append(assetPairs, &pair) | ||
| } | ||
|
|
||
| return assetPairs, nil | ||
| } |
There was a problem hiding this comment.
Improve error handling in PullAssetPairs.
The function logs errors but returns an empty slice and nil error, which might lead to silent failures. Consider returning meaningful errors to the caller.
- return []*oracletypes.AssetPair{}, nil
+ return nil, errCommittable suggestion was skipped due to low confidence.
cmd/injective-price-oracle/oracle.go
Outdated
| conn, _, err = websocket.DefaultDialer.DialContext(ctx, u.String(), header) | ||
| if err != nil { | ||
| log.Infof("Failed to connect to WebSocket server: %v", err) | ||
| retries++ | ||
| if retries > oracle.MaxRetriesReConnectWebSocket { | ||
| log.Infof("Reached maximum retries (%d), exiting...", oracle.MaxRetriesReConnectWebSocket) | ||
| return | ||
| } | ||
| log.Infof("Retrying in 5s...") | ||
| time.Sleep(5 * time.Second) | ||
| } else { | ||
| log.Infof("Connected to WebSocket server") | ||
| return | ||
| } | ||
| } |
There was a problem hiding this comment.
Enhance WebSocket connection logging.
Consider adding more detailed logging for each retry attempt, including the retry count and any specific error messages received.
cmd/injective-price-oracle/oracle.go
Outdated
| err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *dynamicFeedsDir) | ||
| log.WithError(err).Fatalln("failed to load stork feeds") |
There was a problem hiding this comment.
Fix copy-paste error in error handling message.
The error message incorrectly references *dynamicFeedsDir instead of *storkFeedsDir. Correct this to avoid confusion.
- err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *dynamicFeedsDir)
+ err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *storkFeedsDir)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. Thoroughly test & benchmark the code to ensure it meets the requirements.
| err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *dynamicFeedsDir) | |
| log.WithError(err).Fatalln("failed to load stork feeds") | |
| err = errors.Wrapf(err, "stork feeds dir is specified, but failed to read from it: %s", *storkFeedsDir) | |
| log.WithError(err).Fatalln("failed to load stork feeds") |
There was a problem hiding this comment.
@ThanhNhann can you accept this please 😉
There was a problem hiding this comment.
Seems like the humans are having a chat. I'll hop back into my burrow for now. If you need me again, just tag @coderabbitai in a new comment, and I'll come hopping out!
oracle/service.go
Outdated
| if pricePuller.OracleType() == oracletypes.OracleType_Stork { | ||
| storkPricePuller, ok := pricePuller.(*storkPriceFeed) | ||
| if !ok { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithFields(log.Fields{ | ||
| "oracle": "Stork Oracle", | ||
| "retries": maxRetriesPerInterval, | ||
| }).WithError(err).Errorln("can not convert to stork price feed") | ||
| continue | ||
| } | ||
| assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket) | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithError(err).Warningln("retrying PullAssetPairs after error") | ||
|
|
||
| for i := 0; i < maxRetriesPerInterval; i++ { | ||
| if assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket); err != nil { | ||
| time.Sleep(time.Second) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| result, err := pricePuller.PullPrice(ctx) | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithError(err).Warningln("retrying PullPrice after error") | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithFields(log.Fields{ | ||
| "retries": maxRetriesPerInterval, | ||
| }).WithError(err).Errorln("failed to fetch asset pairs") | ||
|
|
||
| for i := 0; i < maxRetriesPerInterval; i++ { | ||
| if result, err = pricePuller.PullPrice(ctx); err != nil { | ||
| time.Sleep(time.Second) | ||
| t.Reset(pricePuller.Interval()) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| } else { |
There was a problem hiding this comment.
Improve error handling in processSetPriceFeed.
The retry logic for pulling asset pairs could be enhanced by adding exponential backoff to reduce load and improve success rates.
oracle/service.go
Outdated
| if priceData.OracleType != oracletypes.OracleType_Stork { | ||
| if priceData.Price.IsZero() || priceData.Price.IsNegative() { | ||
| s.logger.WithFields(log.Fields{ | ||
| "ticker": priceData.Ticker, | ||
| "provider": priceData.ProviderName, | ||
| }).Errorln("got negative or zero price, skipping") | ||
| continue | ||
| } | ||
| } else { | ||
| if len(priceData.AssetPairs) == 0 { | ||
| s.logger.WithFields(log.Fields{ | ||
| "ticker": priceData.Ticker, | ||
| "provider": priceData.ProviderName, | ||
| }).Errorln("got zero asset pair for stork oracle, skipping") | ||
| continue | ||
| } |
There was a problem hiding this comment.
Enhance validation and logging in commitSetPrices.
Consider adding more detailed logging for skipped price data, including the reason for skipping and any relevant data points.
There was a problem hiding this comment.
Actionable comments posted: 0
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (1)
- cmd/injective-price-oracle/oracle.go (9 hunks)
Files skipped from review as they are similar to previous changes (1)
- cmd/injective-price-oracle/oracle.go
…rice-oracle into nhan/stork-relayer
There was a problem hiding this comment.
Actionable comments posted: 0
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
Files selected for processing (2)
- go.mod (3 hunks)
- oracle/service.go (14 hunks)
Files skipped from review as they are similar to previous changes (1)
- oracle/service.go
Additional comments not posted (4)
go.mod (4)
5-6: Specifying the Go toolchain version is a good practice.This helps ensure consistency across development environments.
8-8: Promotion ofcosmossdk.io/mathto a direct dependency is acceptable.This indicates active usage within the codebase.
14-14: Addition ofgithub.com/gorilla/websocketas a direct dependency is appropriate.This aligns with the PR objectives of supporting Stork feeds.
125-125: Updatinggithub.com/grpc-ecosystem/grpc-gateway/v2to version 2.16.0 is beneficial.This update can provide bug fixes and new features. Ensure compatibility with the rest of the codebase.
Run the following script to verify compatibility with the updated dependency:
Verification successful
No direct usage of
github.com/grpc-ecosystem/grpc-gateway/v2found in the codebase.The dependency is marked as indirect and not directly imported, suggesting minimal risk of compatibility issues with the update.
Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify compatibility with the updated grpc-gateway dependency. # Test: Search for grpc-gateway usage. Expect: No compatibility issues. rg --type go 'grpc-gateway'Length of output: 27
Script:
#!/bin/bash # Description: Search for imports of grpc-gateway to verify its usage in the codebase. # Use ast-grep to find any import statements for grpc-gateway ast-grep --lang go --pattern 'import $_ "github.com/grpc-ecosystem/grpc-gateway/v2"'Length of output: 165
cmd/injective-price-oracle/oracle.go
Outdated
| storkWebsocket, err := ConnectWebSocket(ctx, *websocketUrl, *websocketHeader) | ||
| if err != nil { | ||
| err = errors.Wrapf(err, "can not connect with stork oracle websocket") | ||
| log.WithError(err).Fatalln("failed to load stork feeds") |
There was a problem hiding this comment.
This should not be a fatal error; if the Stork feed WS is down for whatever reason this would prevent the whole Injective oracle from starting, if we cannot connect it should disable pulling/pushing Stork oracle prices. Ideally, it should try to reconnect so it can eventually resume the operations
There was a problem hiding this comment.
Currently, we will try to reconnect ws with MaxRetriesReConnectWebSocket = 5, so after that times it will return error with the log that can't not connect with ws, I think in case cannot connect it should disable pulling/pushing Stork as your recommend because we dont want to try to reconnect alot of time and make stuck for another price feeder, wdyt?
There was a problem hiding this comment.
yes, this should be fine 👍
cmd/injective-price-oracle/oracle.go
Outdated
| return | ||
| } | ||
| log.Infof("Retrying connect %sth in 5s...", fmt.Sprint(retries)) | ||
| time.Sleep(5 * time.Second) |
There was a problem hiding this comment.
This code is blocking, it should acknowledge the context
| time.Sleep(5 * time.Second) | |
| select { | |
| case <-ctx.Done(): | |
| return nil, ctx.Err() | |
| case <-time.NewTimer(5*time.Second).C: | |
| } |
cmd/injective-price-oracle/oracle.go
Outdated
| retries := 0 | ||
| for { | ||
| conn, _, err = websocket.DefaultDialer.DialContext(ctx, u.String(), header) | ||
| if err != nil { |
There was a problem hiding this comment.
It should retry only in case of no context error
| if err != nil { | |
| if ctx.Err() != nil { | |
| return ctx.Err() | |
| } else if err != nil { |
cmd/injective-price-oracle/oracle.go
Outdated
| log.Fatal("Error parsing URL:", err) | ||
| return &websocket.Conn{}, err |
There was a problem hiding this comment.
this fatal will panic, it should return the error instead
| log.Fatal("Error parsing URL:", err) | |
| return &websocket.Conn{}, err | |
| return &websocket.Conn{}, errors.Wrapf(err, "can not parse WS url %s: %v", websocketUrl, err) |
oracle/feed_stork.go
Outdated
|
|
||
| var oracleType oracletypes.OracleType | ||
| if cfg.OracleType == "" { | ||
| oracleType = oracletypes.OracleType_PriceFeed |
There was a problem hiding this comment.
shouldn't default to OracleType_Stork ?
| oracleType = oracletypes.OracleType_PriceFeed | |
| oracleType = oracletypes.OracleType_Stork |
In fact, I'm confused why the StorkPriceFeed should be able to set any type other than "Stork", what's the rationale behind this?
And moreover seeing that the type is hardcoded in the getter
func (f *storkPriceFeed) OracleType() oracletypes.OracleType {
return oracletypes.OracleType_Stork
}There was a problem hiding this comment.
I set wrong for this oracle type, it should be oracleType = oracletypes.OracleType_Stork as you recommend
|
|
||
| var msgResp messageResponse | ||
| if err = json.Unmarshal(msgNeed, &msgResp); err != nil { | ||
| return |
There was a problem hiding this comment.
| return | |
| f.logger.Warningln("error unmarshalling feed message:", err) | |
| return |
| @@ -208,38 +230,76 @@ func (s *oracleSvc) processSetPriceFeed(ticker, providerName string, pricePuller | |||
| "provider": pricePuller.ProviderName(), | |||
| }) | |||
|
|
|||
| ctx := context.Background() | |||
| symbol := pricePuller.Symbol() | |||
There was a problem hiding this comment.
symbol will be empty in Stork feeds, why not use the ticker in the config? (even if one file may pull multiple tickers)
There was a problem hiding this comment.
will update this for stork
| } | ||
|
|
||
| for _, priceData := range priceBatch { | ||
| msg := &oracletypes.MsgRelayStorkPrices{ |
There was a problem hiding this comment.
the batch includes all the price data structs from all providers, It needs to check the provider type
| msg := &oracletypes.MsgRelayStorkPrices{ | |
| if priceData.OracleType != oracletypes.OracleType_Stork { | |
| continue | |
| } | |
| msg := &oracletypes.MsgRelayStorkPrices{ |
| } | ||
|
|
||
| } else { |
There was a problem hiding this comment.
We can skip pushing an empty update to the channel
| } else { | |
| if len(assetPairs) == 0 { | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } | |
| } else { |
oracle/service.go
Outdated
| s.logger.WithFields(log.Fields{ | ||
| "ticker": priceData.Ticker, | ||
| "provider": priceData.ProviderName, | ||
| }).Errorln("got zero asset pair for stork oracle, skipping") |
There was a problem hiding this comment.
this doesn't seems to be a real error, just something that can happen
| }).Errorln("got zero asset pair for stork oracle, skipping") | |
| }).Debugln("got zero asset pair for stork oracle, skipping") |
we can avoid this check here completely if we do not push a priceData without any pair as explained in the comment in processSetPriceFeed function
oracle/service.go
Outdated
| @@ -50,21 +50,24 @@ type MultiPricePuller interface { | |||
| type oracleSvc struct { | |||
| pricePullers map[string]PricePuller | |||
| supportedPriceFeeds map[string]PriceFeedConfig | |||
| supportedStorkFeeds map[string]StorkFeedConfig | |||
There was a problem hiding this comment.
It looks like this is not used
| err = errors.Wrapf(err, "failed to init stork price feed for ticker %s", ticker) | ||
| return nil, err | ||
| } | ||
| svc.pricePullers[ticker] = pricePuller |
There was a problem hiding this comment.
Warning: this will overwrite other dynamic feed configs if the ticker collides, I'm ok with it, as long as it is documented, maybe in the Readme so we know that Stork feeds in such case
| } | ||
|
|
||
| // PullAssetPairs pulls asset pair for an asset id | ||
| func (f *storkPriceFeed) PullAssetPairs(conn *websocket.Conn) (assetPairs []*oracletypes.AssetPair, err error) { |
There was a problem hiding this comment.
Is there a reason to send the subscription message more than one time?
I imagine that the optimal way to use this websocket connection is to establish the subscription once and keep reading the incoming message asynchronously in a different goroutine, then when PullAssetPairs is called, simply return the latest values, otherwise, we're using the WS as a normal HTTP request
There was a problem hiding this comment.
I tested the code with multiple stork config and this code is not safe, it tries to send a message to the same WS connection from different pullers concurrently and panics
INFO[0028] Connected to stork websocket
INFO[0028] initialized 4 price pullers svc=oracle
INFO[0029] starting pullers for 4 feeds svc=oracle
panic: concurrent write to websocket connection
We either need to synchronize the access to the WS connection or have different connections
There was a problem hiding this comment.
Currently, the logic for stork is the same as your comment, we establish the connection at begin then when we want to get data of assets, just need to make a request
examples/stork.toml
Outdated
| @@ -0,0 +1,5 @@ | |||
| provider = "Stork" | |||
There was a problem hiding this comment.
This file should be committed in a different folder, with the code as it is, we cannot have price feed and stork configs in the same folder, because they are parsed twice, as DynamicFeedConfig and then as StorkFeedConfig too
Ideally, it'll be better if both file types can coexist in the same folder, it'll require less DevOps overhead and it's much clearer to know all the enabled feeds, the oracleType should be used to discriminate which type of config and the puller to get the prices
There was a problem hiding this comment.
First we should separate them into different folders then process your idea in another pr to make this pr can be merged soon
There was a problem hiding this comment.
Actionable comments posted: 2
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (4)
- cmd/injective-price-oracle/oracle.go (9 hunks)
- examples/stork/stork.toml (1 hunks)
- oracle/feed_stork.go (1 hunks)
- oracle/service.go (13 hunks)
Files skipped from review due to trivial changes (1)
- examples/stork/stork.toml
Files skipped from review as they are similar to previous changes (2)
- cmd/injective-price-oracle/oracle.go
- oracle/feed_stork.go
Additional comments not posted (6)
oracle/service.go (6)
Line range hint
146-190: LGTM! But document the warning about ticker collisions.The function correctly initializes the new parameters and handles errors appropriately. However, ensure that the warning about overwriting dynamic feed configs if the ticker collides is documented.
Line range hint
206-210: LGTM!The function correctly includes the new feed provider in the switch case.
378-395: LGTM!The function correctly composes messages for Stork price feeds.
396-401: LGTM!The function correctly includes the new
composeStorkOracleMsgsmethod.
Line range hint
106-133: LGTM!The function correctly retrieves enabled price feeds.
Line range hint
518-533: LGTM!The function correctly recovers from panics and logs errors.
| if priceData.OracleType != oracletypes.OracleType_Stork { | ||
| if priceData.Price.IsZero() || priceData.Price.IsNegative() { | ||
| s.logger.WithFields(log.Fields{ | ||
| "ticker": priceData.Ticker, | ||
| "provider": priceData.ProviderName, | ||
| }).Debugln("got negative or zero price, skipping") | ||
| continue | ||
| } | ||
| } else { | ||
| if len(priceData.AssetPairs) == 0 { | ||
| s.logger.WithFields(log.Fields{ | ||
| "ticker": priceData.Ticker, | ||
| "provider": priceData.ProviderName, | ||
| }).Debugln("got zero asset pair for stork oracle, skipping") | ||
| continue | ||
| } |
There was a problem hiding this comment.
Enhance validation and logging.
The function correctly includes validation for Stork price data. Consider adding more detailed logging for skipped price data, including the reason for skipping and any relevant data points.
Apply this diff to enhance logging for skipped price data:
if priceData.OracleType != oracletypes.OracleType_Stork {
if priceData.Price.IsZero() || priceData.Price.IsNegative() {
s.logger.WithFields(log.Fields{
"ticker": priceData.Ticker,
"provider": priceData.ProviderName,
+ "reason": "negative or zero price",
}).Debugln("got negative or zero price, skipping")
continue
}
} else {
if len(priceData.AssetPairs) == 0 {
s.logger.WithFields(log.Fields{
"ticker": priceData.Ticker,
"provider": priceData.ProviderName,
+ "reason": "zero asset pair",
}).Debugln("got zero asset pair for stork oracle, skipping")
continue
}
}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. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if priceData.OracleType != oracletypes.OracleType_Stork { | |
| if priceData.Price.IsZero() || priceData.Price.IsNegative() { | |
| s.logger.WithFields(log.Fields{ | |
| "ticker": priceData.Ticker, | |
| "provider": priceData.ProviderName, | |
| }).Debugln("got negative or zero price, skipping") | |
| continue | |
| } | |
| } else { | |
| if len(priceData.AssetPairs) == 0 { | |
| s.logger.WithFields(log.Fields{ | |
| "ticker": priceData.Ticker, | |
| "provider": priceData.ProviderName, | |
| }).Debugln("got zero asset pair for stork oracle, skipping") | |
| continue | |
| } | |
| if priceData.OracleType != oracletypes.OracleType_Stork { | |
| if priceData.Price.IsZero() || priceData.Price.IsNegative() { | |
| s.logger.WithFields(log.Fields{ | |
| "ticker": priceData.Ticker, | |
| "provider": priceData.ProviderName, | |
| "reason": "negative or zero price", | |
| }).Debugln("got negative or zero price, skipping") | |
| continue | |
| } | |
| } else { | |
| if len(priceData.AssetPairs) == 0 { | |
| s.logger.WithFields(log.Fields{ | |
| "ticker": priceData.Ticker, | |
| "provider": priceData.ProviderName, | |
| "reason": "zero asset pair", | |
| }).Debugln("got zero asset pair for stork oracle, skipping") | |
| continue | |
| } |
| ctx, cancelFn := context.WithTimeout(context.Background(), maxRespTime) | ||
| defer cancelFn() | ||
| // define price and asset pair to tracking late | ||
| var err error | ||
| price := zeroPrice | ||
| var assetPairs []*oracletypes.AssetPair | ||
|
|
||
| if pricePuller.OracleType() == oracletypes.OracleType_Stork { | ||
| storkPricePuller, ok := pricePuller.(*storkPriceFeed) | ||
| if !ok { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithFields(log.Fields{ | ||
| "oracle": "Stork Oracle", | ||
| "retries": maxRetriesPerInterval, | ||
| }).WithError(err).Errorln("can not convert to stork price feed") | ||
| continue | ||
| } | ||
| assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket) | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithError(err).Warningln("retrying PullAssetPairs after error") | ||
|
|
||
| for i := 0; i < maxRetriesPerInterval; i++ { | ||
| if assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket); err != nil { | ||
| time.Sleep(time.Second) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| result, err := pricePuller.PullPrice(ctx) | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithError(err).Warningln("retrying PullPrice after error") | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithFields(log.Fields{ | ||
| "retries": maxRetriesPerInterval, | ||
| }).WithError(err).Errorln("failed to fetch asset pairs") | ||
|
|
||
| for i := 0; i < maxRetriesPerInterval; i++ { | ||
| if result, err = pricePuller.PullPrice(ctx); err != nil { | ||
| time.Sleep(time.Second) | ||
| t.Reset(pricePuller.Interval()) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| if len(assetPairs) == 0 { | ||
| t.Reset(pricePuller.Interval()) | ||
| continue | ||
| } | ||
| } else { | ||
| price, err = pricePuller.PullPrice(ctx) | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithFields(log.Fields{ | ||
| "symbol": symbol, | ||
| "retries": maxRetriesPerInterval, | ||
| }).WithError(err).Errorln("failed to fetch price") | ||
| feedLogger.WithError(err).Warningln("retrying PullPrice after error") | ||
|
|
||
| for i := 0; i < maxRetriesPerInterval; i++ { | ||
| if price, err = pricePuller.PullPrice(ctx); err != nil { | ||
| time.Sleep(time.Second) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| t.Reset(pricePuller.Interval()) | ||
| continue | ||
| if err != nil { | ||
| metrics.ReportFuncError(s.svcTags) | ||
| feedLogger.WithFields(log.Fields{ | ||
| "symbol": symbol, | ||
| "retries": maxRetriesPerInterval, | ||
| }).WithError(err).Errorln("failed to fetch price") | ||
|
|
||
| t.Reset(pricePuller.Interval()) | ||
| continue | ||
| } |
There was a problem hiding this comment.
Enhance retry logic with exponential backoff.
The function correctly implements the logic for Stork price feeds and includes appropriate error handling and retry logic. However, consider enhancing the retry logic with exponential backoff to reduce load and improve success rates.
Apply this diff to enhance the retry logic with exponential backoff:
for i := 0; i < maxRetriesPerInterval; i++ {
- if assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket); err != nil {
- time.Sleep(time.Second)
+ backoff := time.Second
+ for i := 0; i < maxRetriesPerInterval; i++ {
+ if assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket); err != nil {
+ time.Sleep(backoff)
+ backoff *= 2
continue
}
break
}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. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ctx, cancelFn := context.WithTimeout(context.Background(), maxRespTime) | |
| defer cancelFn() | |
| // define price and asset pair to tracking late | |
| var err error | |
| price := zeroPrice | |
| var assetPairs []*oracletypes.AssetPair | |
| if pricePuller.OracleType() == oracletypes.OracleType_Stork { | |
| storkPricePuller, ok := pricePuller.(*storkPriceFeed) | |
| if !ok { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "oracle": "Stork Oracle", | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("can not convert to stork price feed") | |
| continue | |
| } | |
| assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket) | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithError(err).Warningln("retrying PullAssetPairs after error") | |
| for i := 0; i < maxRetriesPerInterval; i++ { | |
| if assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket); err != nil { | |
| time.Sleep(time.Second) | |
| continue | |
| } | |
| break | |
| } | |
| result, err := pricePuller.PullPrice(ctx) | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithError(err).Warningln("retrying PullPrice after error") | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("failed to fetch asset pairs") | |
| for i := 0; i < maxRetriesPerInterval; i++ { | |
| if result, err = pricePuller.PullPrice(ctx); err != nil { | |
| time.Sleep(time.Second) | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } | |
| break | |
| } | |
| if len(assetPairs) == 0 { | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } | |
| } else { | |
| price, err = pricePuller.PullPrice(ctx) | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "symbol": symbol, | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("failed to fetch price") | |
| feedLogger.WithError(err).Warningln("retrying PullPrice after error") | |
| for i := 0; i < maxRetriesPerInterval; i++ { | |
| if price, err = pricePuller.PullPrice(ctx); err != nil { | |
| time.Sleep(time.Second) | |
| continue | |
| } | |
| break | |
| } | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "symbol": symbol, | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("failed to fetch price") | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } | |
| ctx, cancelFn := context.WithTimeout(context.Background(), maxRespTime) | |
| defer cancelFn() | |
| // define price and asset pair to tracking late | |
| var err error | |
| price := zeroPrice | |
| var assetPairs []*oracletypes.AssetPair | |
| if pricePuller.OracleType() == oracletypes.OracleType_Stork { | |
| storkPricePuller, ok := pricePuller.(*storkPriceFeed) | |
| if !ok { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "oracle": "Stork Oracle", | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("can not convert to stork price feed") | |
| continue | |
| } | |
| assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket) | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithError(err).Warningln("retrying PullAssetPairs after error") | |
| backoff := time.Second | |
| for i := 0; i < maxRetriesPerInterval; i++ { | |
| if assetPairs, err = storkPricePuller.PullAssetPairs(s.storkWebsocket); err != nil { | |
| time.Sleep(backoff) | |
| backoff *= 2 | |
| continue | |
| } | |
| break | |
| } | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("failed to fetch asset pairs") | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } | |
| } | |
| if len(assetPairs) == 0 { | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } | |
| } else { | |
| price, err = pricePuller.PullPrice(ctx) | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithError(err).Warningln("retrying PullPrice after error") | |
| for i := 0; i < maxRetriesPerInterval; i++ { | |
| if price, err = pricePuller.PullPrice(ctx); err != nil { | |
| time.Sleep(time.Second) | |
| continue | |
| } | |
| break | |
| } | |
| if err != nil { | |
| metrics.ReportFuncError(s.svcTags) | |
| feedLogger.WithFields(log.Fields{ | |
| "symbol": symbol, | |
| "retries": maxRetriesPerInterval, | |
| }).WithError(err).Errorln("failed to fetch price") | |
| t.Reset(pricePuller.Interval()) | |
| continue | |
| } |
Change
Update sdk-go version compatible with
MsgRelayStorkMessageAdd support for
MsgRelayStorkMessageTesting
STORK_WEBSOCKET_URandSTORK_WEBSOCKET_HEADERhereinjective-price-oracle start --stork-feeds examples/stork.tomlSummary by CodeRabbit
Summary by CodeRabbit
New Features
Enhancements
PriceDatastructure to include asset pairs from Stork Oracle.PricePullerinterface andoracleSvcservice to support Stork feeds.