[OTE-823] Fix FNS onchain events staging + retrieval logic #2318
Conversation
WalkthroughThe pull request introduces several changes across multiple files. Key modifications include the addition of a Changes
Possibly related PRs
Suggested labels
Suggested reviewers
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 using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
| @@ -163,10 +163,6 @@ func (f *Flags) Validate() error { | |||
|
|
|||
| // Grpc streaming | |||
| if f.GrpcStreamingEnabled { | |||
| if f.OptimisticExecutionEnabled { | |||
There was a problem hiding this comment.
Previously added to prevent both flags from being accidentally set true
| @@ -177,6 +177,16 @@ func (am AppModule) PreBlock(ctx context.Context) (appmodule.ResponsePreBlock, e | |||
| }, nil | |||
| } | |||
|
|
|||
| // BeginBlock executes all ABCI BeginBlock logic respective to the clob module. | |||
| func (am AppModule) Precommit(ctx context.Context) error { | |||
There was a problem hiding this comment.
Callback was not previously implemented, so ClobModule.Precommit was not being run
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Outside diff range and nitpick comments (1)
protocol/x/clob/module.go (1)
181-188: LGTM! Consider adding a comment explaining the purpose of the Precommit method.The implementation of the
Precommitmethod looks good and follows the established patterns in the module. It correctly implements theappmodule.HasPrecommitinterface and uses telemetry for performance monitoring.Consider adding a brief comment explaining the purpose of the
Precommitmethod and its role in the ABCI lifecycle. This would improve code documentation and help future developers understand its significance. For example:// Precommit executes all ABCI Precommit logic respective to the clob module. // This method is called after the block is committed but before the state is finalized. func (am AppModule) Precommit(ctx context.Context) error { // ... (existing implementation) }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
📒 Files selected for processing (5)
- protocol/app/app.go (1 hunks)
- protocol/app/flags/flags.go (0 hunks)
- protocol/app/flags/flags_test.go (1 hunks)
- protocol/streaming/full_node_streaming_manager.go (8 hunks)
- protocol/x/clob/module.go (1 hunks)
💤 Files not reviewed due to no reviewable changes (1)
- protocol/app/flags/flags.go
🔇 Additional comments not posted (4)
protocol/x/clob/module.go (1)
Line range hint
19-26: Verify interface implementation for appmodule.HasPrecommitThe addition of the
Precommitmethod suggests that theAppModuleshould now implement theappmodule.HasPrecommitinterface. Please ensure that this interface is properly added to the list of implemented interfaces at the beginning of the file.Run the following script to check if the
appmodule.HasPrecommitinterface is properly implemented:If the script doesn't find the
appmodule.HasPrecommitinterface, please add it to the list of implemented interfaces.protocol/app/flags/flags_test.go (1)
116-125: Fix typo in test case name and clarify behavior change.
There's a typo in the test case name: "canbe" should be "can be".
This test case implies a significant change in behavior, allowing optimistic execution to be enabled alongside gRPC streaming. Could you please provide more context on this change?
- Was this previously not allowed? If so, why?
- What are the implications of enabling both optimistic execution and gRPC streaming?
- Are there any potential risks or performance considerations we should be aware of?
The flag values seem consistent with other test cases. However, it would be helpful to understand the reasoning behind the specific values chosen for
GrpcStreamingMaxBatchSize,GrpcStreamingFlushIntervalMs, andGrpcStreamingMaxChannelBufferSize.To ensure this change doesn't introduce unintended side effects, please run the following script:
This will help us identify if there are any other parts of the codebase that need to be updated to reflect this new behavior.
protocol/streaming/full_node_streaming_manager.go (2)
903-903: Assess the security implications of using an infinite gas meter.In line 903, the context is set with an infinite gas meter using
ctx = ctx.WithGasMeter(ante_types.NewFreeInfiniteGasMeter()). While this allows unmetered operations, ensure that it doesn't introduce security risks like potential denial-of-service attacks due to unbounded computation.Review the necessity of an infinite gas meter in this context and consider implementing limits or safeguards if appropriate.
432-435:⚠️ Potential issueEnsure all calls to 'getStagedFinalizeBlockEventsFromStore' pass the new 'cdc' parameter.
The function
getStagedFinalizeBlockEventsFromStorenow requires acdc codec.BinaryCodecparameter (lines 432-435). Check that all invocations of this function are updated to include thecdcargument.Execute the following script to find and review all calls to
getStagedFinalizeBlockEventsFromStore:#!/bin/bash # Description: Verify all calls to 'getStagedFinalizeBlockEventsFromStore' include the 'cdc' parameter. # Find all calls to 'getStagedFinalizeBlockEventsFromStore' rg --type go 'getStagedFinalizeBlockEventsFromStore\(' # Manually inspect each call to ensure 'cdc' is passed
| @@ -2095,6 +2095,7 @@ func getFullNodeStreamingManagerFromOptions( | |||
| appFlags.GrpcStreamingMaxChannelBufferSize, | |||
| appFlags.FullNodeStreamingSnapshotInterval, | |||
| streamingManagerTransientStoreKey, | |||
| cdc, | |||
There was a problem hiding this comment.
💡 Codebase verification
Remove unused cdc parameter from NewFullNodeStreamingManager.
The cdc parameter is currently not utilized within the NewFullNodeStreamingManager function. If it's not required, please remove it to simplify the function signature and maintain code clarity.
🔗 Analysis chain
LGTM. Verify usage of new parameter in NewFullNodeStreamingManager.
The addition of the cdc parameter to NewFullNodeStreamingManager looks good. This change likely introduces codec functionality for serialization/deserialization in the streaming manager.
Please ensure that the NewFullNodeStreamingManager function properly utilizes the new cdc parameter. Run the following script to verify:
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify the NewFullNodeStreamingManager function signature and usage of cdc parameter
# Check the function signature
echo "Checking NewFullNodeStreamingManager function signature:"
rg --type go "func NewFullNodeStreamingManager\(.*codec\.Codec.*\)" ./protocol
# Check for usage of cdc within the function
echo "Checking usage of cdc parameter within NewFullNodeStreamingManager:"
rg --type go -A 10 "func NewFullNodeStreamingManager\(" ./protocol | rg "cdc"
Length of output: 502
Script:
#!/bin/bash
# Description: Verify all usages of the cdc parameter within NewFullNodeStreamingManager and the codebase
# Check for any usage of cdc within the NewFullNodeStreamingManager function
echo "Searching for cdc usage within NewFullNodeStreamingManager:"
rg --type go "func NewFullNodeStreamingManager\(" ./protocol | rg -A 20 "cdc"
# Search for any external usage of the cdc parameter in the codebase
echo "Searching for external usages of cdc parameter:"
rg --type go "\bcdc\b" ./protocol
Length of output: 91461
| stagedEvent := clobtypes.StagedFinalizeBlockEvent{ | ||
| Event: &clobtypes.StagedFinalizeBlockEvent_SubaccountUpdate{ | ||
| SubaccountUpdate: &subaccountUpdate, | ||
| }, | ||
| } | ||
| sm.stageFinalizeBlockEvent( | ||
| ctx, | ||
| clobtypes.Amino.MustMarshal(stagedEvent), | ||
| sm.cdc.MustMarshal(&stagedEvent), |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Replace 'MustMarshal' with error-handling 'Marshal' to prevent panics.
In StageFinalizeBlockSubaccountUpdate (line 406) and StageFinalizeBlockFill (line 428), sm.cdc.MustMarshal(&stagedEvent) is used. MustMarshal will panic if an error occurs during marshaling. It's safer to use Marshal with proper error handling to improve robustness.
Apply this change to handle marshaling errors gracefully:
- sm.cdc.MustMarshal(&stagedEvent)
+ marshaledEvent, err := sm.cdc.Marshal(&stagedEvent)
+ if err != nil {
+ sm.logger.Error("Failed to marshal staged event", "error", err)
+ return // Handle the error appropriately
+ }
+ sm.stageFinalizeBlockEvent(ctx, marshaledEvent)Also applies to: 428-428
| count := getStagedEventsCount(store) | ||
| events := make([]clobtypes.StagedFinalizeBlockEvent, count) | ||
| store = prefix.NewStore(store, []byte(StagedEventsKeyPrefix)) | ||
| for i := uint32(0); i < count; i++ { | ||
| var event clobtypes.StagedFinalizeBlockEvent | ||
| bytes := store.Get(lib.Uint32ToKey(i)) | ||
| clobtypes.Amino.MustUnmarshal(bytes, &event) | ||
| cdc.MustUnmarshal(bytes, &event) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Handle errors from 'Unmarshal' instead of using 'MustUnmarshal'.
At line 442, cdc.MustUnmarshal(bytes, &event) is called. MustUnmarshal panics on error, which can cause the entire application to crash. It's better to use Unmarshal with error checking to make the code more resilient.
Modify the code to handle unmarshaling errors:
- cdc.MustUnmarshal(bytes, &event)
+ if err := cdc.Unmarshal(bytes, &event); err != nil {
+ sm.logger.Error("Failed to unmarshal staged event", "error", err)
+ continue // or handle the error as appropriate
+ }📝 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.
| cdc.MustUnmarshal(bytes, &event) | |
| if err := cdc.Unmarshal(bytes, &event); err != nil { | |
| sm.logger.Error("Failed to unmarshal staged event", "error", err) | |
| continue // or handle the error as appropriate | |
| } |
| @@ -27,6 +28,7 @@ | |||
| type FullNodeStreamingManagerImpl struct { | |||
| sync.Mutex | |||
|
|
|||
| cdc codec.BinaryCodec | |||
There was a problem hiding this comment.
💡 Codebase verification
Update all calls to NewFullNodeStreamingManager to include the cdc parameter.
- File:
protocol/app/app.go- Line: (exact line number not provided)
Ensure that the cdc parameter is passed when instantiating NewFullNodeStreamingManager in all instances.
🔗 Analysis chain
Update all usages to accommodate the new 'cdc' parameter.
The cdc field of type codec.BinaryCodec has been added to the FullNodeStreamingManagerImpl struct (line 31). The constructor NewFullNodeStreamingManager now requires a cdc parameter (line 100), which is assigned during initialization (line 119). Ensure that:
- All calls to
NewFullNodeStreamingManagerinclude the newcdcparameter. - Any instances where
FullNodeStreamingManagerImplis directly instantiated are updated to initialize thecdcfield appropriately.
Run the following script to verify that all calls to NewFullNodeStreamingManager have been updated with the cdc parameter:
Also applies to: 100-100, 119-119
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify all calls to 'NewFullNodeStreamingManager' include the 'cdc' parameter.
# Search for calls to 'NewFullNodeStreamingManager' lacking the 'cdc' parameter
rg --type go 'NewFullNodeStreamingManager\(' -A 5 | grep -v 'cdc'
# Alternatively, list all calls to inspect them manually
rg --type go -A 5 'NewFullNodeStreamingManager\('
Length of output: 1246
| @@ -113,14 +113,17 @@ func TestValidate(t *testing.T) { | |||
| OptimisticExecutionEnabled: true, | |||
| }, | |||
| }, | |||
| "failure - optimistic execution cannot be enabled with gRPC streaming": { | |||
| "success - optimistic execution canbe enabled with gRPC streaming": { | |||
|
@Mergifyio backport release/protocol/v7.x |
✅ Backports have been created
|
(cherry picked from commit 3019562)
Changelist
Fix FNS onchain events staging + retrieval logic.
Test Plan
Tested by running FNS + load-testing locally.
Author/Reviewer Checklist
state-breakinglabel.indexer-postgres-breakinglabel.PrepareProposalorProcessProposal, manually add the labelproposal-breaking.feature:[feature-name].backport/[branch-name].refactor,chore,bug.Summary by CodeRabbit
New Features
Precommitfor enhanced ABCI Precommit logic in the clob module.Bug Fixes
Tests