Improve event-driven reporter with cycle detection #31
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -11,6 +11,7 @@ import ( | |
| "sync" | ||
| "time" | ||
|
|
||
| rpchttp "github.com/cometbft/cometbft/rpc/client/http" | ||
| "github.com/ethereum/go-ethereum/accounts/abi" | ||
| "github.com/shirou/gopsutil/v3/process" | ||
| "github.com/spf13/viper" | ||
|
|
@@ -76,6 +77,177 @@ func validatorOperatorAddress(reporterAddr string) (string, string, error) { | |
|
|
||
| func (c *Client) MonitorCyclelistQuery(ctx context.Context, wg *sync.WaitGroup) { | ||
| defer wg.Done() | ||
|
|
||
| // Try to use event-driven detection (reacts within milliseconds of each new block). | ||
| // Falls back to 200ms ticker polling if the WebSocket subscription fails. | ||
|
Contributor
There was a problem hiding this comment. added fallback behavior to next listed rpc in .env in this branch https://github.com/tellor-io/layer-daemons/tree/pr-31-review |
||
| blockCh, unsubscribe, err := c.subscribeNewBlocks(ctx) | ||
| if err != nil { | ||
| c.logger.Warn("block subscription failed, falling back to polling", "error", err) | ||
| c.monitorCyclelistQueryPolling(ctx) | ||
| return | ||
| } | ||
| defer unsubscribe() | ||
| c.logger.Info("MonitorCyclelistQuery: using event-driven block subscription") | ||
|
|
||
| // Fallback ticker: if no block event is received for 2s (e.g. slow node), poll anyway. | ||
| const blockEventTimeout = 2 * time.Second | ||
| fallback := time.NewTimer(blockEventTimeout) | ||
| defer fallback.Stop() | ||
|
|
||
| prevQueryData := []byte{} | ||
|
|
||
| checkCycle := func() { | ||
| queryCtx, cancel := context.WithTimeout(ctx, defaultQueryTimeout) | ||
| querydata, querymeta, err := c.CurrentQuery(queryCtx) | ||
| cancel() | ||
|
|
||
| if err != nil || querymeta == nil { | ||
| c.logger.Error("query failed", "error", err) | ||
| return | ||
| } | ||
|
|
||
| mutex.Lock() | ||
| hasCommited := commitedIds[querymeta.Id] | ||
| mutex.Unlock() | ||
| if bytes.Equal(querydata, prevQueryData) || hasCommited { | ||
| return | ||
| } | ||
|
|
||
| txCtx, cancel := context.WithTimeout(ctx, defaultTxTimeout) | ||
| done := make(chan struct{}) | ||
|
|
||
| c.logger.Info(fmt.Sprintf("starting to generate spot price report at %d", time.Now().Unix())) | ||
| go func() { | ||
| defer close(done) | ||
| if err := c.GenerateAndBroadcastSpotPriceReport(txCtx, querydata, querymeta); err != nil { | ||
| c.logger.Error("report generation failed", "error", err) | ||
| } | ||
| }() | ||
|
|
||
| select { | ||
| case <-done: | ||
| cancel() | ||
| case <-txCtx.Done(): | ||
| c.logger.Error(fmt.Sprintf("report generation timed out at %d", time.Now().Unix())) | ||
| cancel() | ||
| } | ||
|
|
||
| prevQueryData = querydata | ||
| } | ||
|
|
||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case _, ok := <-blockCh: | ||
| if !ok { | ||
| // Channel closed; try to re-subscribe on a fallback RPC endpoint. | ||
| unsubscribe() | ||
| c.logger.Warn("block subscription channel closed, trying fallback RPC endpoint") | ||
| blockCh, unsubscribe, err = c.subscribeNewBlocks(ctx) | ||
|
Contributor
There was a problem hiding this comment. suggest using subscribeNewBlocksFallback instead to make consistent with existing fallback behavior
Contributor
There was a problem hiding this comment. There is a separate resource cleanup issue that we can solve after this is merged, or as part of this work: rpcEndpointManager creates fresh RPC clients, and the WebSocket path starts them with httpClient.Start(), but the cleanup only unsubscribes and never calls Stop(). When a subscription is replaced, a fallback endpoint is tried, Subscribe fails after Start(), or the primary RPC health check starts a client and then rejects it, the old CometBFT client workers/connections appear to be left running. Can we make the owner of each started RPC client stop it when it is no longer used? |
||
| if err != nil { | ||
| c.logger.Warn("block subscription failed on all RPC endpoints, falling back to polling", "error", err) | ||
| c.monitorCyclelistQueryPolling(ctx) | ||
| return | ||
| } | ||
| fallback.Reset(blockEventTimeout) | ||
| continue | ||
| } | ||
| fallback.Reset(blockEventTimeout) | ||
| checkCycle() | ||
| case <-fallback.C: | ||
| // No block event received within timeout; check anyway and reset. | ||
| fallback.Reset(blockEventTimeout) | ||
| checkCycle() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // subscribeNewBlocks subscribes to CometBFT NewBlock events over WebSocket. | ||
| // Returns a channel that receives one value per block, an unsubscribe func, and any error. | ||
| // Tries each RPC endpoint in rpcManager before returning an error. | ||
| func (c *Client) subscribeNewBlocks(ctx context.Context) (<-chan struct{}, func(), error) { | ||
| if c.rpcManager == nil { | ||
| if c.rpcClient == nil { | ||
| return nil, func() {}, fmt.Errorf("rpc client not initialized") | ||
| } | ||
| return c.subscribeNewBlocksWithClient(ctx, c.rpcClient, "current") | ||
| } | ||
|
|
||
| var errs []string | ||
| rpcClientVal, endpoint, err := c.rpcManager.currentClient() | ||
| if err != nil { | ||
| errs = append(errs, fmt.Sprintf("%s: %v", endpoint, err)) | ||
| } else if blockCh, unsubscribe, err := c.subscribeNewBlocksWithClient(ctx, rpcClientVal, endpoint); err == nil { | ||
| c.setRPCClient(rpcClientVal) | ||
| return blockCh, unsubscribe, nil | ||
| } else { | ||
| errs = append(errs, fmt.Sprintf("%s: %v", endpoint, err)) | ||
| } | ||
|
|
||
| for attempt := 0; attempt < c.rpcManager.endpointCount()-1; attempt++ { | ||
| rpcClientVal, endpoint, err = c.rpcManager.nextClient() | ||
| if err != nil { | ||
| errs = append(errs, err.Error()) | ||
| break | ||
| } | ||
| blockCh, unsubscribe, err := c.subscribeNewBlocksWithClient(ctx, rpcClientVal, endpoint) | ||
| if err == nil { | ||
| c.setRPCClient(rpcClientVal) | ||
| return blockCh, unsubscribe, nil | ||
| } | ||
| errs = append(errs, fmt.Sprintf("%s: %v", endpoint, err)) | ||
| } | ||
|
|
||
| return nil, func() {}, fmt.Errorf("subscribing to NewBlock events on RPC endpoints: %s", strings.Join(errs, "; ")) | ||
| } | ||
|
|
||
| func (c *Client) subscribeNewBlocksWithClient(ctx context.Context, rpcClientVal interface{}, endpoint string) (<-chan struct{}, func(), error) { | ||
| httpClient, ok := rpcClientVal.(*rpchttp.HTTP) | ||
| if !ok { | ||
| return nil, func() {}, fmt.Errorf("RPC client for %s does not support WebSocket subscriptions", endpoint) | ||
| } | ||
| if !httpClient.IsRunning() { | ||
| if err := httpClient.Start(); err != nil { | ||
| return nil, func() {}, fmt.Errorf("starting rpc client for WebSocket: %w", err) | ||
| } | ||
| } | ||
| subscriber := fmt.Sprintf("reporter-cycle-monitor-%d", time.Now().UnixNano()) | ||
| eventCh, err := httpClient.Subscribe(ctx, subscriber, "tm.event='NewBlock'") | ||
| if err != nil { | ||
| return nil, func() {}, fmt.Errorf("subscribing to NewBlock events: %w", err) | ||
| } | ||
|
|
||
| blockCh := make(chan struct{}, 1) | ||
| go func() { | ||
| defer close(blockCh) | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case _, ok := <-eventCh: | ||
| if !ok { | ||
| return | ||
| } | ||
| // Non-blocking send: if the consumer is busy processing the previous block, | ||
| // skip this event rather than queuing up a backlog. | ||
| select { | ||
| case blockCh <- struct{}{}: | ||
| default: | ||
| } | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| unsubscribe := func() { | ||
| _ = httpClient.Unsubscribe(ctx, subscriber, "tm.event='NewBlock'") | ||
| } | ||
| return blockCh, unsubscribe, nil | ||
| } | ||
|
|
||
| // monitorCyclelistQueryPolling is the fallback ticker-based implementation used when | ||
| // the WebSocket block subscription is unavailable. | ||
| func (c *Client) monitorCyclelistQueryPolling(ctx context.Context) { | ||
| prevQueryData := []byte{} | ||
| retryDelay := defaultRetryDelay | ||
| ticker := time.NewTicker(defaultRetryDelay) | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
please address lint check