Skip to content

Improve event-driven reporter with cycle detection #31

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
0xSpuddy merged 6 commits into from
Jun 25, 2026
Merged

Improve event-driven reporter with cycle detection #31

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
f73dd2d
perf: event-driven reporter cycle detection
May 29, 2026
588e48f
Merge branch 'main' into perf/event-driven-cycle-monitor
diorwave Jun 2, 2026
9b12426
Add import for CometBFT RPC HTTP client
diorwave Jun 2, 2026
5e94993
feat: RPC endpoint fallback for WebSocket subscriptions
Jun 3, 2026
38b41d6
reporter: remove duplicate RestorePrimaryEndpointsPeriodically gorout…
diorwave Jun 22, 2026
b63bd2f
Merge branch 'tellor-io:main' into perf/event-driven-cycle-monitor
diorwave Jun 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
159 changes: 91 additions & 68 deletions reporter/client/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,10 @@
"sync"
"sync/atomic"
"time"

Check failure on line 14 in reporter/client/client.go

View workflow job for this annotation

GitHub Actions / golangci-lint 

File is not properly formatted (gci)

@0xSpuddy 0xSpuddy Jun 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please address lint check

"github.com/ethereum/go-ethereum/common"
"github.com/spf13/viper"
rpchttp "github.com/cometbft/cometbft/rpc/client/http"
globalfeetypes "github.com/strangelove-ventures/globalfee/x/globalfee/types"
customquery "github.com/tellor-io/layer-daemons/custom_query"
daemonflags "github.com/tellor-io/layer-daemons/flags"
Expand Down Expand Up @@ -185,6 +186,8 @@
grpcMu sync.RWMutex
grpcConn *grpc.ClientConn
grpcClient daemontypes.GrpcClient

rpcClient *rpchttp.HTTP // direct reference for WebSocket subscriptions

@0xSpuddy 0xSpuddy Jun 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adding this import in a new branch pr-31-review

grpcManager *grpcEndpointManager
rpcMu sync.RWMutex
rpcManager *rpcEndpointManager
Expand Down Expand Up @@ -383,14 +386,13 @@
if err != nil {
return fmt.Errorf("failed to initialize RPC endpoint manager: %w", err)
}
rpcClient, rpcEndpoint, err := rpcManager.currentClient()
rpcClientVal, rpcEndpoint, err := rpcManager.currentClient()
if err != nil {
return fmt.Errorf("failed to create RPC client: %w", err)
}
c.logger.Info("CometBFT RPC client established", "endpoint", rpcEndpoint)
c.rpcManager = rpcManager
c.setRPCClient(rpcClient)

c.setRPCClient(rpcClientVal)
c.logger.Info("CometBFT RPC client established", "endpoint", rpcEndpoint)
encodingConfig := CreateEncodingConfig()
c.cosmosCtx = c.cosmosCtx.WithCodec(encodingConfig.Codec).WithInterfaceRegistry(encodingConfig.InterfaceRegistry).WithTxConfig(encodingConfig.TxConfig)

Expand Down Expand Up @@ -535,56 +537,6 @@
return fmt.Errorf("%s failed on all gRPC endpoints: %w", operation, lastErr)
}

func (c *Client) RestorePrimaryEndpointsPeriodically(ctx context.Context, wg *sync.WaitGroup) {
defer wg.Done()

ticker := time.NewTicker(primaryEndpointCheckInterval)
defer ticker.Stop()

for {
select {
case <-ctx.Done():
return
case <-ticker.C:
c.tryRestorePrimaryRPCEndpoint(ctx)
c.tryRestorePrimaryGRPCEndpoint(ctx)
}
}
}

func (c *Client) tryRestorePrimaryRPCEndpoint(ctx context.Context) {
if c.rpcManager == nil || c.rpcManager.usingPrimary() {
return
}

probeCtx, cancel := context.WithTimeout(ctx, primaryEndpointProbeTimeout)
defer cancel()

rpcClient, endpoint, err := c.rpcManager.primaryClient()
if err != nil {
c.logger.Warn("Primary CometBFT RPC endpoint is not ready", "endpoint", endpoint, "error", err)
return
}
status, err := rpcClient.Status(probeCtx)
if err != nil {
c.logger.Warn("Primary CometBFT RPC endpoint health check failed", "endpoint", endpoint, "error", err)
return
}
chainID := c.chainID()
if status.NodeInfo.Network != chainID {
c.logger.Warn(
"Primary CometBFT RPC endpoint returned unexpected chain ID",
"endpoint", endpoint,
"expected_chain_id", chainID,
"actual_chain_id", status.NodeInfo.Network,
)
return
}

c.setRPCClient(rpcClient)
c.rpcManager.switchToPrimary()
}

func (c *Client) tryRestorePrimaryGRPCEndpoint(ctx context.Context) {
if c.grpcManager == nil || c.grpcManager.usingPrimary() {
return
Expand Down Expand Up @@ -640,25 +592,11 @@
}
}

func (c *Client) setRPCClient(rpcClient client.CometRPC) {
c.rpcMu.Lock()
defer c.rpcMu.Unlock()
c.cosmosCtxMu.Lock()
defer c.cosmosCtxMu.Unlock()
c.cosmosCtx = c.cosmosCtx.WithClient(rpcClient)
}

func (c *Client) rpcContextWithClient(rpcClient client.CometRPC) client.Context {
clientCtx := c.currentCosmosContext()
return clientCtx.WithClient(rpcClient)
}

func (c *Client) currentCosmosContext() client.Context {
c.cosmosCtxMu.RLock()
defer c.cosmosCtxMu.RUnlock()
return c.cosmosCtx
}

func (c *Client) chainID() string {
return c.currentCosmosContext().ChainID
}
Expand Down Expand Up @@ -715,6 +653,9 @@
client.BroadcastTxMsgToChain(ctx)
}()

wg.Add(1)
go client.RestorePrimaryEndpointsPeriodically(ctx, wg)
Comment thread
diorwave marked this conversation as resolved.
Outdated

wg.Add(1)
go client.MonitorCyclelistQuery(ctx, wg)

Expand All @@ -741,6 +682,88 @@
wg.Wait()
}

func (c *Client) setRPCClient(rpcClient client.CometRPC) {
c.rpcMu.Lock()
defer c.rpcMu.Unlock()
c.cosmosCtxMu.Lock()
defer c.cosmosCtxMu.Unlock()
if httpClient, ok := rpcClient.(*rpchttp.HTTP); ok {
c.rpcClient = httpClient
}
c.cosmosCtx = c.cosmosCtx.WithClient(rpcClient)
}

func (c *Client) currentCosmosContext() client.Context {
c.cosmosCtxMu.RLock()
defer c.cosmosCtxMu.RUnlock()
return c.cosmosCtx
}

func (c *Client) RestorePrimaryEndpointsPeriodically(ctx context.Context, wg *sync.WaitGroup) {
defer wg.Done()

if c.rpcManager == nil {
return
}

ticker := time.NewTicker(primaryEndpointCheckInterval)
defer ticker.Stop()

for {
select {
case <-ctx.Done():
return
case <-ticker.C:
c.tryRestorePrimaryRPCEndpoint(ctx)
c.tryRestorePrimaryGRPCEndpoint(ctx)
}
}
}

func (c *Client) tryRestorePrimaryRPCEndpoint(ctx context.Context) {
if c.rpcManager == nil || c.rpcManager.usingPrimary() {
return
}

probeCtx, cancel := context.WithTimeout(ctx, primaryEndpointProbeTimeout)
defer cancel()

rpcClientVal, endpoint, err := c.rpcManager.primaryClient()
if err != nil {
c.logger.Warn("Primary CometBFT RPC endpoint is not ready", "endpoint", endpoint, "error", err)
return
}
httpClient, ok := rpcClientVal.(*rpchttp.HTTP)
if !ok {
return
}
if !httpClient.IsRunning() {
if err := httpClient.Start(); err != nil {
c.logger.Warn("Primary CometBFT RPC endpoint failed to start", "endpoint", endpoint, "error", err)
return
}
}
status, err := httpClient.Status(probeCtx)
if err != nil {
c.logger.Warn("Primary CometBFT RPC endpoint health check failed", "endpoint", endpoint, "error", err)
return
}
chainID := c.currentCosmosContext().ChainID
if status.NodeInfo.Network != chainID {
c.logger.Warn(
"Primary CometBFT RPC endpoint returned unexpected chain ID",
"endpoint", endpoint,
"expected_chain_id", chainID,
"actual_chain_id", status.NodeInfo.Network,
)
return
}

c.setRPCClient(rpcClientVal)
c.rpcManager.switchToPrimary()
c.logger.Info("CometBFT RPC endpoint restored to primary", "endpoint", endpoint)
}

func (c *Client) RefreshGasEstimatesPeriodically(ctx context.Context, wg *sync.WaitGroup) {
defer wg.Done()
ticker := time.NewTicker(c.refreshGasEstimatesInterval)
Expand Down
Loading
Loading