optimize WS newhead #1615
optimize WS newhead #1615
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| @@ -98,7 +98,8 @@ | |||
| func (a *FilterAPI) NewFilter( | |||
| _ context.Context, | |||
| crit filters.FilterCriteria, | |||
| ) (ethrpc.ID, error) { | |||
| ) (id ethrpc.ID, err error) { | |||
| defer recordMetrics("eth_newFilter", time.Now(), err == nil) | |||
Check warning
Code scanning / CodeQL
Calling the system time Warning
| @@ -113,7 +114,8 @@ | |||
|
|
|||
| func (a *FilterAPI) NewBlockFilter( | |||
| _ context.Context, | |||
| ) (ethrpc.ID, error) { | |||
| ) (id ethrpc.ID, err error) { | |||
| defer recordMetrics("eth_newBlockFilter", time.Now(), err == nil) | |||
Check warning
Code scanning / CodeQL
Calling the system time Warning
| @@ -128,7 +130,8 @@ | |||
| func (a *FilterAPI) GetFilterChanges( | |||
| ctx context.Context, | |||
| filterID ethrpc.ID, | |||
| ) (interface{}, error) { | |||
| ) (res interface{}, err error) { | |||
| defer recordMetrics("eth_getFilterChanges", time.Now(), err == nil) | |||
Check warning
Code scanning / CodeQL
Calling the system time Warning
| @@ -178,7 +181,8 @@ | |||
| func (a *FilterAPI) GetFilterLogs( | |||
| ctx context.Context, | |||
| filterID ethrpc.ID, | |||
| ) ([]*ethtypes.Log, error) { | |||
| ) (res []*ethtypes.Log, err error) { | |||
| defer recordMetrics("eth_getFilterLogs", time.Now(), err == nil) | |||
Check warning
Code scanning / CodeQL
Calling the system time Warning
| @@ -206,7 +210,8 @@ | |||
| func (a *FilterAPI) GetLogs( | |||
| ctx context.Context, | |||
| crit filters.FilterCriteria, | |||
| ) ([]*ethtypes.Log, error) { | |||
| ) (res []*ethtypes.Log, err error) { | |||
| defer recordMetrics("eth_getLogs", time.Now(), err == nil) | |||
Check warning
Code scanning / CodeQL
Calling the system time Warning
| } | ||
|
|
||
| func (a *SubscriptionAPI) NewHeads(ctx context.Context) (*rpc.Subscription, error) { | ||
| func (a *SubscriptionAPI) NewHeads(ctx context.Context) (s *rpc.Subscription, err error) { | ||
| defer recordMetrics("eth_newHeads", time.Now(), err == nil) |
Check warning
Code scanning / CodeQL
Calling the system time Warning
| return rpcSub, nil | ||
| } | ||
|
|
||
| func (a *SubscriptionAPI) Logs(ctx context.Context, filter *filters.FilterCriteria) (*rpc.Subscription, error) { | ||
| func (a *SubscriptionAPI) Logs(ctx context.Context, filter *filters.FilterCriteria) (s *rpc.Subscription, err error) { | ||
| defer recordMetrics("eth_logs", time.Now(), err == nil) |
Check warning
Code scanning / CodeQL
Calling the system time Warning
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #1615 +/- ##
==========================================
+ Coverage 61.62% 61.68% +0.06%
==========================================
Files 365 365
Lines 26135 26169 +34
==========================================
+ Hits 16106 16143 +37
+ Misses 8959 8956 -3
Partials 1070 1070
|
| } | ||
| case <-rpcSub.Err(): | ||
| return | ||
| break OUTER |
There was a problem hiding this comment.
is this the same as continue here?
There was a problem hiding this comment.
no we want to break out of the outer loop (instead of the select clause)
| } | ||
| id, subCh, err := api.subscriptionManager.Subscribe(context.Background(), NewHeadQueryBuilder(), api.subscriptonConfig.subscriptionCapacity) |
There was a problem hiding this comment.
does this need supervision or some mechanism to retry/reconnect (can the subscription die?)
There was a problem hiding this comment.
I think the subscription manager has a retry mechanism (that's what the subscription was canceled, resubscribing log we saw)
Describe your changes and provide context
instead of having each subscription subscribing to tendermint and encode to JSON independently, we will maintain a single subscription with tendermint and a single copy of encoded JSON per head, then fan it out to all outgoing subscriptions
Testing performed to validate your change