refactor(zetaclient): subscribe to new blocks in scheduler #3228
Conversation
📝 WalkthroughWalkthroughThe pull request introduces significant enhancements to the Changes
Assessment against linked issues
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
Documentation and Community
|
|
!!!WARNING!!! Be very careful about using Only suppress a single rule (or a specific set of rules) within a section of code, while continuing to scan for other problems. To do this, you can list the rule(s) to be suppressed within the #nosec annotation, e.g: /* #nosec G401 */ or //#nosec G201 G202 G203 Pay extra attention to the way |
gartnera
force-pushed
the
zetaclient-subscribe-new-blocks
branch
from
8fab7cd
to
5733a5b
Compare
gartnera
force-pushed
the
zetaclient-subscribe-new-blocks
branch
from
5733a5b
to
276e5ad
Compare
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #3228 +/- ##
========================================
Coverage 62.00% 62.00%
========================================
Files 428 428
Lines 30469 30466 -3
========================================
Hits 18891 18891
+ Misses 10738 10735 -3
Partials 840 840
|
gartnera
requested review from
brewmaster012,
kingpinXD,
lumtis,
ws4charlie,
skosito,
swift1337 and
fbac
as code owners
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Outside diff range and nitpick comments (7)
cmd/zetaclientd/start.go (1)
Line range hint
254-266: Consider addressing the disabled zeta supply checkerThe TODO comment indicates that the zeta supply checker is disabled due to performance concerns with GRPC queries. This might be worth revisiting now that we're improving synchronization performance.
Would you like me to:
- Open a GitHub issue to track the re-enablement of the supply checker?
- Propose an implementation that addresses the performance concerns?
zetaclient/orchestrator/orchestrator_test.go (1)
591-591: Consider enhancing mock client initialization.While the mock client is correctly added to the Orchestrator struct, the current initialization could be improved to better support testing different scenarios.
Consider this enhancement:
- cometbftClient: &cometbft_rpc_client_mock.Client{}, + cometbftClient: func() *cometbft_rpc_client_mock.Client { + mockClient := &cometbft_rpc_client_mock.Client{} + // Initialize with default expectations + mockClient.On("Start").Return(nil) + return mockClient + }(),zetaclient/orchestrator/orchestrator.go (5)
321-330: Simplify Operator Balance Update LogicThe current logic for updating the operator balance includes unnecessary checks and can be streamlined for clarity and efficiency.
diff --git a/zetaclient/orchestrator/orchestrator.go b/zetaclient/orchestrator/orchestrator.go --- a/zetaclient/orchestrator/orchestrator.go +++ b/zetaclient/orchestrator/orchestrator.go @@ -324,13 +324,9 @@ } else { diff := oc.lastOperatorBalance.Sub(balance) - if diff.GT(sdkmath.NewInt(0)) && diff.LT(sdkmath.NewInt(math.MaxInt64)) { + if diff.IsPositive() { oc.ts.AddFeeEntry(bn, diff.Int64()) oc.lastOperatorBalance = balance } }
341-344: Enhance Error Logging for ClarityImprove the error message when failing to retrieve pending cross-chain transactions to provide clearer context for troubleshooting.
-oc.logger.Error().Err(err).Msgf("runScheduler: GetPendingCctxsWithinRatelimit failed") +oc.logger.Error().Err(err).Msg("failed to retrieve pending cross-chain transactions within rate limit")🧰 Tools
🪛 GitHub Check: codecov/patch
[warning] 341-344: zetaclient/orchestrator/orchestrator.go#L341-L344
Added lines #L341 - L344 were not covered by tests
358-360: Provide Additional Context in Error MessagesWhen logging errors during signer resolution, include the chain ID and relevant details to aid in debugging.
oc.logger.Error().Err(err). Int64(logs.FieldChain, chainID). - Msg("runScheduler: unable to resolve signer") + Msgf("unable to resolve signer for chain ID %d", chainID)🧰 Tools
🪛 GitHub Check: codecov/patch
[warning] 358-360: zetaclient/orchestrator/orchestrator.go#L358-L360
Added lines #L358 - L360 were not covered by tests
364-369: Avoid Redundant Error Handling in Observer ResolutionThe error handling logic duplicates context provided by the
errors.Wrapffunction. Streamlining this can enhance readability.if err != nil { - oc.logger.Error().Err(err). - Int64(logs.FieldChain, chainID). - Msg("runScheduler: unable to resolve observer") + oc.logger.Error().Err(err).Msgf("unable to resolve observer for chain ID %d", chainID) continue }🧰 Tools
🪛 GitHub Check: codecov/patch
[warning] 364-369: zetaclient/orchestrator/orchestrator.go#L364-L369
Added lines #L364 - L369 were not covered by tests
299-406: Increase Test Coverage for New Scheduler LogicThe new event-driven scheduler implementation lacks test coverage, as indicated by static analysis. Enhancing tests will improve reliability and prevent regressions.
Would you like me to assist in creating unit tests for the
runSchedulermethod to ensure thorough coverage?🧰 Tools
🪛 GitHub Check: codecov/patch
[warning] 299-302: zetaclient/orchestrator/orchestrator.go#L299-L302
Added lines #L299 - L302 were not covered by tests
[warning] 309-329: zetaclient/orchestrator/orchestrator.go#L309-L329
Added lines #L309 - L329 were not covered by tests
[warning] 333-338: zetaclient/orchestrator/orchestrator.go#L333-L338
Added lines #L333 - L338 were not covered by tests
[warning] 341-344: zetaclient/orchestrator/orchestrator.go#L341-L344
Added lines #L341 - L344 were not covered by tests
[warning] 347-350: zetaclient/orchestrator/orchestrator.go#L347-L350
Added lines #L347 - L350 were not covered by tests
[warning] 353-356: zetaclient/orchestrator/orchestrator.go#L353-L356
Added lines #L353 - L356 were not covered by tests
[warning] 358-360: zetaclient/orchestrator/orchestrator.go#L358-L360
Added lines #L358 - L360 were not covered by tests
[warning] 364-369: zetaclient/orchestrator/orchestrator.go#L364-L369
Added lines #L364 - L369 were not covered by tests
[warning] 373-379: zetaclient/orchestrator/orchestrator.go#L373-L379
Added lines #L373 - L379 were not covered by tests
[warning] 383-384: zetaclient/orchestrator/orchestrator.go#L383-L384
Added lines #L383 - L384 were not covered by tests
[warning] 388-401: zetaclient/orchestrator/orchestrator.go#L388-L401
Added lines #L388 - L401 were not covered by tests
[warning] 406-406: zetaclient/orchestrator/orchestrator.go#L406
Added line #L406 was not covered by tests
📜 Review details
Configuration used: .coderabbit.yaml
Review profile: CHILL
📒 Files selected for processing (3)
cmd/zetaclientd/start.go(3 hunks)zetaclient/orchestrator/orchestrator.go(5 hunks)zetaclient/orchestrator/orchestrator_test.go(2 hunks)
🧰 Additional context used
📓 Path-based instructions (3)
cmd/zetaclientd/start.go (1)
Pattern **/*.go: Review the Go code, point out issues relative to principles of clean code, expressiveness, and performance.
zetaclient/orchestrator/orchestrator.go (1)
Pattern **/*.go: Review the Go code, point out issues relative to principles of clean code, expressiveness, and performance.
zetaclient/orchestrator/orchestrator_test.go (1)
Pattern **/*.go: Review the Go code, point out issues relative to principles of clean code, expressiveness, and performance.
🪛 GitHub Check: codecov/patch
zetaclient/orchestrator/orchestrator.go
[warning] 114-114: zetaclient/orchestrator/orchestrator.go#L114
Added line #L114 was not covered by tests
[warning] 299-302: zetaclient/orchestrator/orchestrator.go#L299-L302
Added lines #L299 - L302 were not covered by tests
[warning] 309-329: zetaclient/orchestrator/orchestrator.go#L309-L329
Added lines #L309 - L329 were not covered by tests
[warning] 333-338: zetaclient/orchestrator/orchestrator.go#L333-L338
Added lines #L333 - L338 were not covered by tests
[warning] 341-344: zetaclient/orchestrator/orchestrator.go#L341-L344
Added lines #L341 - L344 were not covered by tests
[warning] 347-350: zetaclient/orchestrator/orchestrator.go#L347-L350
Added lines #L347 - L350 were not covered by tests
[warning] 353-356: zetaclient/orchestrator/orchestrator.go#L353-L356
Added lines #L353 - L356 were not covered by tests
[warning] 358-360: zetaclient/orchestrator/orchestrator.go#L358-L360
Added lines #L358 - L360 were not covered by tests
[warning] 364-369: zetaclient/orchestrator/orchestrator.go#L364-L369
Added lines #L364 - L369 were not covered by tests
[warning] 373-379: zetaclient/orchestrator/orchestrator.go#L373-L379
Added lines #L373 - L379 were not covered by tests
[warning] 383-384: zetaclient/orchestrator/orchestrator.go#L383-L384
Added lines #L383 - L384 were not covered by tests
[warning] 388-401: zetaclient/orchestrator/orchestrator.go#L388-L401
Added lines #L388 - L401 were not covered by tests
[warning] 406-406: zetaclient/orchestrator/orchestrator.go#L406
Added line #L406 was not covered by tests
🔇 Additional comments (5)
cmd/zetaclientd/start.go (1)
239-239: Verify orchestrator's usage of CometBFT client
The integration of CometBFT client with the orchestrator looks correct. However, let's verify its usage in the orchestrator package.
✅ Verification successful
CometBFT client integration verified and properly implemented
The orchestrator correctly uses the CometBFT client for block subscription:
- Client is properly defined in the Orchestrator struct (
cometbftClient cometbft_rpc_client.Client) - Block subscription is implemented in
zetaclient/orchestrator/orchestrator.gousingSubscribemethod withtm.event='NewBlock' - The subscription is used to receive new block events through
blockEventChan
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify how the CometBFT client is used in the orchestrator package
# Check for block subscription usage
ast-grep --pattern 'func (o *Orchestrator) runScheduler() {
$$$
Subscribe($$$)
$$$
}'
# Check for proper cleanup
ast-grep --pattern 'func (o *Orchestrator) Stop() {
$$$
Stop()
$$$
}'
Length of output: 171
Script:
#!/bin/bash
# Let's try a broader search to understand the orchestrator's implementation
# Find orchestrator files
fd orchestrator
# Check how CometBFT client is used in orchestrator
rg -A 5 "CometClient|cometClient"
# Look for block subscription patterns
rg -A 5 "Subscribe|EventsClient|BlockClient"
# Check orchestrator struct definition
ast-grep --pattern 'type Orchestrator struct {
$$$
}'
Length of output: 16113
zetaclient/orchestrator/orchestrator_test.go (2)
15-15: LGTM: Import addition aligns with the new CometBFT integration.
The addition of the CometBFT RPC client mock import is consistent with the PR's objective of implementing block subscription.
Line range hint 1-675: Verify test coverage for CometBFT integration.
The test file has comprehensive coverage for existing functionality, but we should ensure adequate testing of the new CometBFT integration.
zetaclient/orchestrator/orchestrator.go (2)
333-334: Verify Accuracy of Hot Key Burn Rate Metric
Ensure that the HotKeyBurnRate metric reflects the most recent operator balance to maintain accurate monitoring.
Would you like assistance in implementing a verification method to confirm the metric's accuracy?
🧰 Tools
🪛 GitHub Check: codecov/patch
[warning] 333-338: zetaclient/orchestrator/orchestrator.go#L333-L338
Added lines #L333 - L338 were not covered by tests
373-379: Optimize Metric Updates for Pending Transactions
Updating Prometheus metrics inside the main loop may impact performance. Consider batching metric updates or moving them outside critical execution paths.
[performance]
🧰 Tools
🪛 GitHub Check: codecov/patch
[warning] 373-379: zetaclient/orchestrator/orchestrator.go#L373-L379
Added lines #L373 - L379 were not covered by tests
| cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) | ||
| cometbftClient, err := cometbft_client.New(cometbftURL, "/websocket") | ||
| if err != nil { | ||
| return errors.Wrapf(err, "new cometbft client (%s)", cometbftURL) | ||
| } | ||
| // start websockets | ||
| err = cometbftClient.WSEvents.Start() | ||
| if err != nil { | ||
| return errors.Wrap(err, "cometbft start") | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Enhance CometBFT client configuration and lifecycle management
Several improvements can be made to the CometBFT client setup:
- The port number (26657) should be configurable via the config struct
- WebSocket connection should be properly cleaned up during shutdown
- Error messages could be more descriptive for troubleshooting
Consider applying these changes:
-cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657)
+cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, cfg.CometBFTPort)
cometbftClient, err := cometbft_client.New(cometbftURL, "/websocket")
if err != nil {
- return errors.Wrapf(err, "new cometbft client (%s)", cometbftURL)
+ return errors.Wrapf(err, "failed to create CometBFT client at %s: %v", cometbftURL, err)
}
+defer func() {
+ if err := cometbftClient.WSEvents.Stop(); err != nil {
+ logger.Error().Err(err).Msg("failed to stop CometBFT WebSocket events")
+ }
+}()
+
// start websockets
err = cometbftClient.WSEvents.Start()
if err != nil {
- return errors.Wrap(err, "cometbft start")
+ return errors.Wrap(err, "failed to start CometBFT WebSocket events")
}Committable suggestion skipped: line range outside the PR's diff.
There was a problem hiding this comment.
- we hardcode other port numbers in zetaclient. unrelated refactoring.
- sure
- error messages are fine imo
There was a problem hiding this comment.
@gartnera, understood. Thank you for the clarification. I'll note that in zetaclient, hardcoded port numbers are the standard practice and modifying them is considered unrelated refactoring. I'm glad you're agreeable to ensuring proper cleanup of the WebSocket connection during shutdown.
✏️ Learnings added
Learnt from: gartnera
PR: zeta-chain/node#3228
File: cmd/zetaclientd/start.go:222-231
Timestamp: 2024-11-27T22:02:47.873Z
Learning: In the `zetaclient` codebase, port numbers like `26657` are hardcoded, and changing them to be configurable via the config struct is considered unrelated refactoring.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
| @@ -217,12 +219,24 @@ func Start(_ *cobra.Command, _ []string) error { | |||
| return errors.Wrap(err, "unable to create chain observer map") | |||
| } | |||
|
|
|||
| cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) | |||
There was a problem hiding this comment.
Consider adding this as a function for Config struct.
There was a problem hiding this comment.
I'd prefer to keep this very simple for now and refactor it later. We also instantiate this client internally in zetaclient/zetacore/client_cosmos.go but it really should probably be injected from the top imo.
There was a problem hiding this comment.
Can we open an issue for the future consideration?
| @@ -289,122 +296,114 @@ func (oc *Orchestrator) runScheduler(ctx context.Context) error { | |||
| return err | |||
| } | |||
|
|
|||
| observeTicker := time.NewTicker(3 * time.Second) | |||
| var lastBlockNum int64 | |||
| blockEventChan, err := oc.cometbftClient.Subscribe(ctx, "", "tm.event='NewBlock'") | |||
There was a problem hiding this comment.
Can we define the query filter as a constant?
| metrics.HotKeyBurnRate.Set(float64(oc.ts.HotKeyBurnRate.GetBurnRate().Int64())) | ||
|
|
||
| // get chain ids without zeta chain | ||
| chainIDs := lo.FilterMap(app.ListChains(), func(c zctx.Chain, _ int) (int64, bool) { |
There was a problem hiding this comment.
Could you use the defined FilterChains instead of importing an external library?
I see this library being used in other places in the codebase. Minimizing usage would be a good idea.
There was a problem hiding this comment.
app.FilterChains(...) returns []zctx.Chain not []int64 as required by oc.GetPendingCctxsWithinRateLimit
This is not new logic (it's just unindented).
| if lastBlockNum == 0 { | ||
| lastBlockNum = bn - 1 | ||
|
|
||
| if !app.IsOutboundObservationEnabled() { |
There was a problem hiding this comment.
Can we add a log here and maybe move this checkup?
Do we even need to try to start processing an outbound if it's disabled?
There was a problem hiding this comment.
This logic does seem a bit strange to me too. But it's not new, this is unindenting the existing logic.
| @@ -217,12 +219,24 @@ func Start(_ *cobra.Command, _ []string) error { | |||
| return errors.Wrap(err, "unable to create chain observer map") | |||
| } | |||
|
|
|||
| cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) | |||
There was a problem hiding this comment.
| cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) | |
| cometBFTURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) |
| @@ -217,12 +219,24 @@ func Start(_ *cobra.Command, _ []string) error { | |||
| return errors.Wrap(err, "unable to create chain observer map") | |||
| } | |||
|
|
|||
| cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) | |||
| cometbftClient, err := cometbft_client.New(cometbftURL, "/websocket") | |||
There was a problem hiding this comment.
| cometbftClient, err := cometbft_client.New(cometbftURL, "/websocket") | |
| cometBFTClient, err := cometbft_client.New(cometbftURL, "/websocket") |
| @@ -217,12 +219,24 @@ func Start(_ *cobra.Command, _ []string) error { | |||
| return errors.Wrap(err, "unable to create chain observer map") | |||
| } | |||
|
|
|||
| cometbftURL := fmt.Sprintf("http://%s:%d", cfg.ZetaCoreURL, 26657) | |||
There was a problem hiding this comment.
Can we open an issue for the future consideration?
| @@ -84,6 +89,7 @@ type multiLogger struct { | |||
| func New( | |||
| ctx context.Context, | |||
| client interfaces.ZetacoreClient, | |||
| cometbftClient cometbft_rpc_client.Client, | |||
There was a problem hiding this comment.
Let's integrate this into zetacore client wrapper in zetaclient/zetacore/*
There was a problem hiding this comment.
Isn't it better to ultimately have smaller focused interfaces rather than one giant interface.
There was a problem hiding this comment.
Generally yes, but here it introduces some unnecessary details to the orchestrator. Also mocking an interface for oc.zetacore.NewBlockSubscriber(ctx) (chan ..., error) is easier than for websocket client
| @@ -289,122 +296,114 @@ func (oc *Orchestrator) runScheduler(ctx context.Context) error { | |||
| return err | |||
| } | |||
|
|
|||
| observeTicker := time.NewTicker(3 * time.Second) | |||
| var lastBlockNum int64 | |||
| blockEventChan, err := oc.cometbftClient.Subscribe(ctx, "", "tm.event='NewBlock'") | |||
There was a problem hiding this comment.
oc.zetacore.NewBlockSubscriber(ctx) (chan ..., error)
| @@ -289,122 +296,114 @@ func (oc *Orchestrator) runScheduler(ctx context.Context) error { | |||
| return err | |||
| } | |||
|
|
|||
| observeTicker := time.NewTicker(3 * time.Second) | |||
There was a problem hiding this comment.
This ticker was running every 3s. Now, it runs every block or slower. What would happen if the next tick occurred at N+2 instead of N+1?
There was a problem hiding this comment.
I'm not exactly sure what you're saying. We weren't doing anything until we got a new block.
There was a problem hiding this comment.
Is it possible to "lose" a "tick" and jump from the 100th to the 102nd block?
There was a problem hiding this comment.
Yes that is possible if zetacored restarts. But that would have resulted in a missed tick in the polling style too.
The actual work to do comes from the GetPendingCctxsWithinRateLimit call.
Use cometbft block subscription to improve signer synchronization.
Closes #3227
Performance tests show ~25% improvement in overall runtime and massive reductions p50+ stats:
Before:
After:
I expect improvements on mainnet and testnet but as extreme without some additional latency correction.
Summary by CodeRabbit
Release Notes
New Features
Bug Fixes
Tests