(OraklNode) Refresh fetcher when global aggregate and local aggregate price differ by more than 30% #2242
Conversation
|
Warning Rate limit exceeded@nick-bisonai has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 1 minutes and 34 seconds before requesting another review. How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. WalkthroughWalkthroughThe changes encompass significant modifications to the Changes
Sequence Diagram(s)sequenceDiagram
participant App
participant MessageBus
participant Aggregator
participant Fetcher
App->>MessageBus: Publish Refresh Request
MessageBus->>Aggregator: Handle Refresh
Aggregator->>Aggregator: Check LastRefreshTime
Aggregator-->>MessageBus: Refresh on Cooldown (if applicable)
Aggregator-->>MessageBus: Process Refresh (if allowed)
MessageBus->>Fetcher: Trigger Fetch
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 (
|
nick-bisonai
changed the title
nick-bisonai
force-pushed
the
feat/fetcher-refresh-logic
branch
from
c9ab262 to
2f121d9
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (12)
- node/pkg/aggregator/aggregator.go (6 hunks)
- node/pkg/aggregator/aggregator_test.go (4 hunks)
- node/pkg/aggregator/app.go (1 hunks)
- node/pkg/aggregator/globalaggregatebulkwriter_test.go (1 hunks)
- node/pkg/aggregator/types.go (2 hunks)
- node/pkg/error/sentinel.go (2 hunks)
- node/pkg/fetcher/app.go (3 hunks)
- node/pkg/fetcher/app_test.go (2 hunks)
- node/pkg/fetcher/localaggregatebulkwriter_test.go (1 hunks)
- node/pkg/fetcher/main_test.go (1 hunks)
- node/pkg/fetcher/types.go (2 hunks)
- node/pkg/fetcher/utils_test.go (1 hunks)
Additional comments not posted (27)
node/pkg/fetcher/localaggregatebulkwriter_test.go (4)
24-24: LGTM!Running the application in a goroutine is a good approach to allow concurrent execution.
The code changes are approved.
26-26: LGTM!The sleep period ensures that the bulk writer has enough time to process data before assertions are made.
The code changes are approved.
28-31: LGTM!The database query and error handling are correctly implemented.
The code changes are approved.
32-32: LGTM!The assertion ensures that data is being produced and stored correctly in the database.
The code changes are approved.
node/pkg/fetcher/utils_test.go (1)
120-120: LGTM!The initialization step ensures that the application state is correctly configured before any assertions or further actions are taken.
The code changes are approved.
node/pkg/aggregator/globalaggregatebulkwriter_test.go (1)
89-89: LGTM!The additional parameter likely represents a messaging bus or similar component, enhancing the functionality of the
NewAggregatorfunction.The code changes are approved.
node/pkg/fetcher/types.go (2)
28-28: LGTM!The
RefreshCooldownIntervalconstant is correctly defined.The code change is approved.
95-95: LGTM! But verify the initialization and usage ofLastRefreshTime.The
LastRefreshTimefield is correctly defined. Ensure that this field is properly initialized and updated in the codebase.The code change is approved.
Run the following script to verify the initialization and usage of
LastRefreshTime:Verification successful
The
LastRefreshTimefield is properly initialized and used.The field is initialized with
time.Time{}, checked for zero value, and updated withtime.Now(). This indicates correct usage in managing time-based logic.
- Initialization:
node/pkg/fetcher/app.go- Usage:
node/pkg/fetcher/app.goScripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization and usage of `LastRefreshTime`. # Test: Search for the initialization and usage of `LastRefreshTime`. Expect: Proper initialization and updates. rg --type go 'LastRefreshTime'Length of output: 327
node/pkg/aggregator/aggregator_test.go (5)
33-33: LGTM!The
NewAggregatorfunction call is correctly updated to includenilfor theBusparameter.The code change is approved.
26-26: LGTM! But verify the initialization ofBusin the test setup.The
NewAggregatorfunction call is correctly updated to includetestItems.app.Bus. Ensure that theBusparameter is properly initialized in the test setup.The code change is approved.
Run the following script to verify the initialization of
Busin the test setup:Verification successful
Verification Successful:
BusInitialization in Test SetupThe
Busparameter is properly initialized in the test setup, ensuring its correct usage in theNewAggregatorfunction call. No issues found with the initialization.
- Initialization observed in
node/pkg/fetcher/main_test.goandnode/pkg/aggregator/main_test.go.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization of `Bus` in the test setup. # Test: Search for the initialization of `Bus` in the test setup. Expect: Proper initialization. rg --type go 'Bus' -A 5Length of output: 22908
49-49: LGTM! But verify the initialization ofBusin the test setup.The
NewAggregatorfunction call is correctly updated to includetestItems.app.Bus. Ensure that theBusparameter is properly initialized in the test setup.The code change is approved.
Run the following script to verify the initialization of
Busin the test setup:Verification successful
Verification Successful:
BusInitialization in Test Setup ConfirmedThe
Busparameter is properly initialized in the test setup for theNewAggregatorfunction call. The initialization is consistent across the test files, ensuring correct setup for the tests.
node/pkg/aggregator/aggregator_test.go:Busis passed astestItems.app.Bus.node/pkg/aggregator/main_test.go:Businitialized astestItems.messageBus = mb.node/pkg/admin/tests/main_test.go: Similar initialization pattern observed.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization of `Bus` in the test setup. # Test: Search for the initialization of `Bus` in the test setup. Expect: Proper initialization. rg --type go 'Bus' -A 5Length of output: 22908
73-73: LGTM! But verify the initialization ofBusin the test setup.The
NewAggregatorfunction call is correctly updated to includetestItems.app.Bus. Ensure that theBusparameter is properly initialized in the test setup.The code change is approved.
Run the following script to verify the initialization of
Busin the test setup:Verification successful
Bus Initialization Verified in Test Setup
The
Busparameter is properly initialized in the test setup for theNewAggregatorfunction call. ThemessageBusis correctly assigned and used within the test structure. No further action is required.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization of `Bus` in the test setup. # Test: Search for the initialization of `Bus` in the test setup. Expect: Proper initialization. rg --type go 'Bus' -A 5Length of output: 22908
101-101: LGTM! But verify the initialization ofBusin the test setup.The
NewAggregatorfunction call is correctly updated to includetestItems.app.Bus. Ensure that theBusparameter is properly initialized in the test setup.The code change is approved.
Run the following script to verify the initialization of
Busin the test setup:Verification successful
Bus Initialization Verified in Test Setup
The
Busparameter is properly initialized in the test setup, as confirmed by its presence and assignment in the test files. TheNewAggregatorfunction call correctly includestestItems.app.Bus. No further action is required.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization of `Bus` in the test setup. # Test: Search for the initialization of `Bus` in the test setup. Expect: Proper initialization. rg --type go 'Bus' -A 5Length of output: 22908
node/pkg/aggregator/types.go (2)
189-194: LGTM! But verify the initialization and usage ofroundLocalAggregate.The fields
roundTriggers,roundPrices,roundPriceFixes, androundProofshave been replaced withroundLocalAggregate. Ensure that the new field is properly initialized and used in the codebase.The code change is approved.
Run the following script to verify the initialization and usage of
roundLocalAggregate:Verification successful
Initialization and Usage of
roundLocalAggregateVerifiedThe
roundLocalAggregatefield is properly initialized and used in the codebase. It is initialized as an empty map and is used to store, access, and delete round aggregates based on round IDs.
- Initialization:
roundLocalAggregateis initialized as an empty map inaggregator.go.- Usage: It is updated, accessed, and deleted appropriately in
aggregator.go.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization and usage of `roundLocalAggregate`. # Test: Search for the initialization and usage of `roundLocalAggregate`. Expect: Proper initialization and updates. rg --type go 'roundLocalAggregate'Length of output: 482
203-204: LGTM! But verify the initialization and usage ofbus.The
busfield is correctly defined. Ensure that this field is properly initialized and used in the codebase.The code change is approved.
Run the following script to verify the initialization and usage of
bus:Verification successful
Initialization and Usage of
busVerifiedThe
busfield in theAggregatorstruct is properly initialized and actively used in the codebase. The initialization occurs in theNewfunction, and the field is utilized in various message handling functions, confirming its intended functionality.
- Initialization:
node/pkg/aggregator/app.goin theNewfunction.- Usage: Various message handling functions in
node/pkg/aggregator/app.goandnode/pkg/aggregator/aggregator.go.Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the initialization and usage of `bus`. # Test: Search for the initialization and usage of `bus`. Expect: Proper initialization and updates. rg --type go 'bus'Length of output: 16646
node/pkg/fetcher/app_test.go (2)
118-169: LGTM!The test function
TestAppRefreshis correctly implemented and verifies the refresh mechanism of the application.The code changes are approved.
171-241: LGTM!The test function
TestAppRefreshCooldownis correctly implemented and verifies the cooldown mechanism of the refresh operation.The code changes are approved.
node/pkg/fetcher/main_test.go (1)
265-267: LGTM!The refactoring of the
cleanupfunction to calltestItems.app.stopAll(ctx)instead oftestItems.app.stopAllFetchers(ctx)is appropriate and consolidates functionality under a more general method.The code changes are approved.
node/pkg/aggregator/app.go (1)
130-132: LGTM!The modification to include
a.Busin theNewAggregatorcall enhances the functionality of the aggregator initialization process by allowing interaction with a message bus.The code changes are approved.
node/pkg/fetcher/app.go (2)
28-28: LGTM!The
LastRefreshTimefield is correctly added and initialized.The code changes are approved.
97-101: LGTM!The cooldown mechanism is correctly implemented to prevent frequent refresh operations.
The code changes are approved.
node/pkg/aggregator/aggregator.go (4)
7-10: LGTM!The new imports are necessary for the added functionality.
The code changes are approved.
20-28: LGTM!The
NewAggregatorfunction signature and theAggregatorstruct are correctly updated to include the message bus parameter and field.The code changes are approved.
Also applies to: 65-66
117-118: LGTM!The refined locking mechanism ensures thread safety.
The code changes are approved.
296-322: LGTM!The new logic enhances the robustness of the aggregator by ensuring it can respond to inconsistencies in aggregate values effectively.
The code changes are approved.
node/pkg/error/sentinel.go (2)
95-95: LGTM!The new error variable provides a specific error message for scenarios where there is a mismatch between global and local prices.
The code changes are approved.
160-160: LGTM!The new error variable improves the robustness of the Fetcher service's error reporting by handling cases where a data refresh attempt fails due to a cooldown period not being completed.
The code changes are approved.
There was a problem hiding this comment.
Actionable comments posted: 2
Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Files selected for processing (1)
- node/pkg/aggregator/aggregator.go (6 hunks)
Additional comments not posted (2)
node/pkg/aggregator/aggregator.go (2)
117-118: LGTM!The changes enhance thread safety by holding the lock for the entire method execution.
The code changes are approved.
Line range hint
319-334: LGTM!The function is correctly implemented.
The code changes are approved.
| @@ -102,16 +114,14 @@ func (n *Aggregator) HandleTriggerMessage(ctx context.Context, msg raft.Message) | |||
| return err | |||
| } | |||
| defer n.leaveOnlyLast10Entries(triggerMessage.RoundID) | |||
|
|
|||
| n.mu.Lock() | |||
| defer n.mu.Unlock() | |||
There was a problem hiding this comment.
there are two locks in this function both of which are released when function returns. I wonder if it'd be same result having a single lock 🤔
There was a problem hiding this comment.
the scope which n.mu.Lock() and n.roundTriggers.mu.Lock() are having are different so it probably shouldn't have the same result 😅
to be more precise, it might be better to have separate mutex for roundLocalAggregate value
There was a problem hiding this comment.
i thought n.mu.Lock() can do the job of n.roundTriggers.mu.Lock() but seems like it's worth keeping both since it introduces more clarity to the logic?
|
I'll keep this open for a while, I'm not sure if this approach is a good idea yet |
| @@ -142,6 +152,7 @@ func (n *Aggregator) HandleTriggerMessage(ctx context.Context, msg raft.Message) | |||
| // if not enough messages collected from HandleSyncReplyMessage, it will hang in certain round | |||
| value = -1 | |||
| } else { | |||
| n.roundLocalAggregate[triggerMessage.RoundID] = localAggregate.Value | |||
| value = localAggregate.Value | |||
There was a problem hiding this comment.
n.roundLocalAggregate[triggerMessage.RoundID] = value
Agreed. This is quite a bit of automation, might easily spiral out of control. |
nick-bisonai
changed the title
|
will close this pr since this kind of situation are rarely happening, and even if it occurs, always temporary |
Description
It'll auto restart fetcher if difference over 30% between local aggregate and global aggregate is detected
The problem with this approach is, that it will restart fetchers from properly working nodes too.
better approach might be comparing with given local aggregate values from other nodes. yet there aren't enough nodes atm to check if current node's value is majority or not
Type of change
Please delete options that are not relevant.
Checklist before requesting a review
Deployment