fix(da): fix polling fallback when ws not available#3361
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds ChangesDA WebSocket Fallback and Polling
Sequence Diagram(s)sequenceDiagram
participant Caller
participant NewWSClient
participant retryWSLoop
participant tryUpgradeWS
participant Subscriber
Caller->>NewWSClient: NewWSClient(addr)
alt WS dial succeeds
NewWSClient-->>Caller: Client{IsWebSocket:true}
Caller->>Subscriber: Start()
Subscriber->>Subscriber: followLoop (WS subscription)
else WS dial fails
NewWSClient->>retryWSLoop: start background retry goroutine
NewWSClient-->>Caller: Client{IsWebSocket:false}
Caller->>Subscriber: Start()
Subscriber->>Subscriber: pollLoop (HTTP polling)
loop every 30s
retryWSLoop->>tryUpgradeWS: attempt WS upgrade
tryUpgradeWS-->>retryWSLoop: IsWebSocket=true
end
Subscriber->>Subscriber: SupportsSubscribe() now true
Subscriber->>Subscriber: spawn followLoop, exit pollLoop
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Claude finished @julienrbrt's task in 2m 50s —— View job Code Review: fix(da): fix polling fallback when ws not available
The PR correctly addresses the core bug — falling back to HTTP polling instead of staying stuck on a non-functional WebSocket subscribe. The overall architecture (poll loop → follow loop upgrade, background retry) is sound. There are two concurrency issues that need attention before merging. Critical:
|
|
The latest Buf updates on your PR. Results from workflow CI / buf-check (pull_request).
|
julienrbrt
force-pushed
the
julien/fix-polling-fallback
branch
from
e81f6c9 to
4e8d3ba
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@block/internal/da/client.go`:
- Around line 493-494: The SupportsSubscribe() method in
block/internal/da/client.go reads the IsWebSocket field from the da client
without synchronization, while the tryUpgradeWS() method in
pkg/da/jsonrpc/client.go writes to IsWebSocket under mutex protection. This
creates a data race as the background retryWSLoop goroutine calls
tryUpgradeWS(). To fix this, add a thread-safe accessor method
SupportsSubscribe() in the Client type in pkg/da/jsonrpc/client.go that acquires
the mu mutex, reads IsWebSocket, and returns the value, then update the
SupportsSubscribe() method in block/internal/da/client.go to call this new
accessor method instead of directly accessing the IsWebSocket field.
In `@pkg/da/jsonrpc/client.go`:
- Around line 17-25: Replace the `IsWebSocket` bool field in the Client struct
with `atomic.Bool` to eliminate the data race. Then update the write site in the
`tryUpgradeWS` function where `IsWebSocket` is assigned (around line 168) to use
the atomic.Store method instead of direct assignment. Finally, update the read
site in `SupportsSubscribe()` function in `block/internal/da/client.go` (lines
493-495) to use the atomic.Load method instead of reading the field directly,
which ensures race-safe access without requiring mutex synchronization.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 2fc19a96-2370-4ed8-8a01-7c52b64e82aa
📒 Files selected for processing (12)
CHANGELOG.mdapps/evm/server/force_inclusion_test.goblock/internal/da/async_block_retriever_test.goblock/internal/da/client.goblock/internal/da/interface.goblock/internal/da/subscriber.goblock/internal/da/tracing.goblock/internal/da/tracing_test.goblock/internal/syncing/syncer_test.gopkg/da/jsonrpc/client.gotest/mocks/da.gotest/testda/dummy.go
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
pkg/da/jsonrpc/client.go (2)
157-170:⚠️ Potential issue | 🔴 Critical | 🏗️ Heavy liftAvoid swapping RPC internals while callers can read them concurrently.
c.muonly protects this writer; polling/catchup callers can still usec.Bloborc.Headerwhile the retry goroutine replaces theirInternalfields. Use an immutable transport snapshot behind a synchronized/atomic indirection, or route every RPC call through the same lock before publishingIsWebSocket=true.As per coding guidelines, be careful with concurrent access to shared state in Go code.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/da/jsonrpc/client.go` around lines 157 - 170, The code swaps the Internal fields of c.Blob and c.Header without protecting concurrent readers who may be using these fields for RPC calls simultaneously. While c.mu protects this write operation in the retry goroutine, polling and catchup callers can still access c.Blob.Internal and c.Header.Internal concurrently, causing a race condition. To fix this, either implement an immutable transport snapshot pattern where the entire transport is swapped atomically behind a synchronized indirection (like a pointer to a struct containing Blob and Header), or ensure all RPC callers route their calls through c.mu before the swap happens and only publish IsWebSocket=true after the swap is protected by the same lock that guards all RPC read operations.Source: Coding guidelines
23-25:⚠️ Potential issue | 🟠 Major | ⚡ Quick winPrevent WS upgrades after
Closehas started.
Closecancels the retry loop but does not mark the client closed. IftryUpgradeWSalready createdwsClient, it can acquirec.muafterClosecaptures the old closer and install a new WS transport thatClosewill never close.Suggested fix
type Client struct { Blob BlobAPI Header HeaderAPI IsWebSocket atomic.Bool mu sync.Mutex closer jsonrpc.ClientCloser retryCancel context.CancelFunc // stops the background WS retry loop + closed bool } func (c *Client) Close() { if c == nil { return } c.mu.Lock() if c.retryCancel != nil { c.retryCancel() c.retryCancel = nil } closer := c.closer + c.closer = nil + c.closed = true c.mu.Unlock() if closer != nil { closer() } }c.mu.Lock() defer c.mu.Unlock() + if c.closed { + wsClient.Close() + return true + } + // Another goroutine may have already upgraded. if c.IsWebSocket.Load() { wsClient.Close() return true }As per coding guidelines, be mindful of goroutine leaks in Go code.
Also applies to: 34-42, 148-170
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/da/jsonrpc/client.go` around lines 23 - 25, The client struct needs a closed flag to track whether Close has been called, preventing race conditions in goroutine management. Add a boolean field (like closed or isClosed) to the struct alongside mu, closer, and retryCancel. In the Close method, acquire c.mu, set this flag to true, then proceed with canceling the retry loop and closing the existing transport. In the tryUpgradeWS method, check this closed flag while holding c.mu at the beginning and return early if the client is already closed, ensuring no new WS transports are installed after Close has started. This prevents goroutine leaks from WS clients that would never be closed.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@pkg/da/jsonrpc/client.go`:
- Around line 157-170: The code swaps the Internal fields of c.Blob and c.Header
without protecting concurrent readers who may be using these fields for RPC
calls simultaneously. While c.mu protects this write operation in the retry
goroutine, polling and catchup callers can still access c.Blob.Internal and
c.Header.Internal concurrently, causing a race condition. To fix this, either
implement an immutable transport snapshot pattern where the entire transport is
swapped atomically behind a synchronized indirection (like a pointer to a struct
containing Blob and Header), or ensure all RPC callers route their calls through
c.mu before the swap happens and only publish IsWebSocket=true after the swap is
protected by the same lock that guards all RPC read operations.
- Around line 23-25: The client struct needs a closed flag to track whether
Close has been called, preventing race conditions in goroutine management. Add a
boolean field (like closed or isClosed) to the struct alongside mu, closer, and
retryCancel. In the Close method, acquire c.mu, set this flag to true, then
proceed with canceling the retry loop and closing the existing transport. In the
tryUpgradeWS method, check this closed flag while holding c.mu at the beginning
and return early if the client is already closed, ensuring no new WS transports
are installed after Close has started. This prevents goroutine leaks from WS
clients that would never be closed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 1b423264-26b3-4b99-819a-d51141a0dc6b
📒 Files selected for processing (2)
block/internal/da/client.gopkg/da/jsonrpc/client.go
🚧 Files skipped from review as they are similar to previous changes (1)
- block/internal/da/client.go
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #3361 +/- ##
==========================================
- Coverage 60.97% 60.60% -0.37%
==========================================
Files 129 129
Lines 14111 14201 +90
==========================================
+ Hits 8604 8607 +3
- Misses 4556 4639 +83
- Partials 951 955 +4
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
2 participants
Overview
#3146 uses subscribe (websocket) only, and the fallback to http in #3211, just changes the transport but we still use subscribe. when switching back to http we need to keep using getall/get instead.
Summary by CodeRabbit
SupportsSubscribe()capability, enabling syncing to use subscription-based updates when available and switch to polling otherwise.