Not returning on connection close for chainsync, block-fetch and tx-submission protocol #1141
Conversation
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Closes #1112 |
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
WalkthroughCentralizes connection error handling and context management, refactors protocol state handling with guarded state queries and done detection, adds guards to avoid sending protocol "done" messages when already finished, and adds/updates comprehensive protocol and connection tests. Changes
Sequence Diagram(s)sequenceDiagram
participant App as Application
participant Conn as Connection
participant Mux as Muxer
participant Proto as Protocol(s)
Note over Mux,Conn: Muxer emits error (e.g., EOF)
Mux->>Conn: Error(err)
rect rgba(200,200,200,0.08)
Note over Conn: Centralized handling
Conn->>Conn: handleConnectionError(err)
Conn->>Proto: Query protocol states (IsDone / CurrentState)
Proto-->>Conn: state responses
end
alt All protocols done
Conn->>App: Treat as normal closure (no error)
else Any protocol active
Conn->>Conn: Wrap/propagate error
Conn->>App: Emit error via ErrorChan
end
Conn->>Conn: Close connection
sequenceDiagram
participant Client as Protocol Client
participant Proto as Protocol State Machine
participant Mux as Muxer
Client->>Proto: Start()
Proto->>Proto: stateLoop() (set initial currentState under lock)
loop Active lifecycle
Client->>Proto: Send messages
Proto->>Mux: Write messages
Proto->>Proto: transitionState() (uses stateRespChan to query current state)
Proto->>Proto: update currentState (locked)
end
Client->>Client: Stop()
alt Not Done
Client->>Proto: Send Done message
Proto->>Mux: Write done
else Already Done
Client->>Client: Skip sending done
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes
Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
♻️ Duplicate comments (1)
protocol/protocol.go (1)
139-151: IsDone() behavior matches intended semanticsTreating AgencyNone and InitialState as done addresses idle/never‑started cases cleanly.
🧹 Nitpick comments (6)
protocol/blockfetch/client.go (1)
112-118: Consider locking Stop() like chainsync to avoid interleavingAlign with chainsync by guarding the Done send with busyMutex to avoid racing with GetBlock/GetBlockRange.
Apply:
func (c *Client) Stop() error { var err error c.onceStop.Do(func() { c.Protocol.Logger(). Debug("stopping client protocol", "component", "network", "protocol", ProtocolName, "connection_id", c.callbackContext.ConnectionId.String(), ) - if !c.IsDone() { + c.busyMutex.Lock() + defer c.busyMutex.Unlock() + if !c.IsDone() { msg := NewMsgClientDone() if err = c.SendMessage(msg); err != nil { return } } }) return err }protocol/blockfetch/blockfetch_test.go (2)
51-59: Avoid Read() returning (0, nil); block or EOF to prevent busy spinReturning 0, nil violates net.Conn expectations and can spin the muxer. Make Read block until close, then return EOF.
Apply:
-func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil } +func (c *testConn) Read(b []byte) (n int, err error) { + <-c.closeChan + return 0, io.EOF +}
159-165: Increase timeout to reduce flakiness100ms can be tight on CI. Consider 500ms–1s to avoid false negatives.
connection_test.go (2)
151-156: Typo: “stoppeds”.Minor nit.
- // Protocol is stoppeds + // Protocol is stopped
192-210: Prefer waiting for protocol init instead of fixed sleep.Replace fixed sleep with the same polling used above to reduce flakes under CI load.
protocol/txsubmission/txsubmission_test.go (1)
154-167: Good message-send assertion; consider shorter timeout.Looks fine; 2s is generous, but with a running muxer you can trim to 500ms.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
connection.go(2 hunks)connection_test.go(2 hunks)protocol/blockfetch/blockfetch_test.go(1 hunks)protocol/blockfetch/client.go(1 hunks)protocol/chainsync/chainsync_test.go(1 hunks)protocol/chainsync/client.go(1 hunks)protocol/protocol.go(8 hunks)protocol/txsubmission/client.go(1 hunks)protocol/txsubmission/txsubmission_test.go(1 hunks)
🧰 Additional context used
🪛 GitHub Actions: go-test
connection_test.go
[error] 74-74: TestErrorHandlingWithActiveProtocols: unexpected error when creating Connection object: handshake: timeout waiting on transition from protocol state Propose
[error] 132-132: TestErrorHandlingWithActiveProtocols: unexpected error when creating Connection object: handshake: timeout waiting on transition from protocol state Propose
🔇 Additional comments (6)
protocol/chainsync/client.go (1)
150-154: Good guard: only send Done when not already donePrevents redundant Done and aligns Stop with protocol state. Locking via busyMutex is appropriate.
protocol/protocol.go (1)
491-540: State tracking and readiness signalling look solidLocking around currentState updates/reads and using stateEntry for agency/timeout is correct. Timeout message uses CurrentState() safely.
If any protocol has multiple AgencyNone states, ensure StateMap defines exactly one terminal state; otherwise GetDoneState() can be ambiguous.
Also applies to: 550-551, 556-569, 592-595, 630-631
protocol/chainsync/chainsync_test.go (1)
151-166: Good coverage for client Done send.Starting the muxer and asserting a write to the fake conn verifies the basic send path.
protocol/txsubmission/client.go (1)
86-92: No-op change.Formatting-only; safe to merge.
protocol/txsubmission/txsubmission_test.go (2)
176-181: Server start/stop smoke test LGTM.No issues.
16-29: Import nettest sync; prepare for safer conn behavior.You’ll add sync for mutex/once; rest unchanged.
import ( "io" "log/slog" "net" + "sync" "testing" "time"Likely an incorrect or invalid review comment.
| // handleConnectionError handles connection-level errors centrally | ||
| func (c *Connection) handleConnectionError(err error) error { | ||
| if err == nil { | ||
| return nil | ||
| } | ||
|
|
||
| // Only propagate EOF errors when acting as a client with active server-side protocols | ||
| if errors.Is(err, io.EOF) { | ||
| // Check if we have any active server-side protocols | ||
| if c.server { | ||
| return err | ||
| } | ||
|
|
||
| // For clients, only propagate EOF if we have active server protocols | ||
| hasActiveServerProtocols := false | ||
| if c.chainSync != nil && c.chainSync.Server != nil && !c.chainSync.Server.IsDone() { | ||
| hasActiveServerProtocols = true | ||
| } | ||
| if c.blockFetch != nil && c.blockFetch.Server != nil && !c.blockFetch.Server.IsDone() { | ||
| hasActiveServerProtocols = true | ||
| } | ||
| if c.txSubmission != nil && c.txSubmission.Server != nil && !c.txSubmission.Server.IsDone() { | ||
| hasActiveServerProtocols = true | ||
| } | ||
| if c.localStateQuery != nil && c.localStateQuery.Server != nil && !c.localStateQuery.Server.IsDone() { | ||
| hasActiveServerProtocols = true | ||
| } | ||
| if c.localTxMonitor != nil && c.localTxMonitor.Server != nil && !c.localTxMonitor.Server.IsDone() { | ||
| hasActiveServerProtocols = true | ||
| } | ||
| if c.localTxSubmission != nil && c.localTxSubmission.Server != nil && !c.localTxSubmission.Server.IsDone() { | ||
| hasActiveServerProtocols = true | ||
| } | ||
|
|
||
| if hasActiveServerProtocols { | ||
| return err | ||
| } | ||
|
|
||
| // EOF with no active server protocols is normal connection closure | ||
| return nil | ||
| } | ||
|
|
||
| // For non-EOF errors, always propagate | ||
| return err | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
EOF handling logic inverted for server/client; also consider ErrUnexpectedEOF
- As server, we should NOT surface EOF if all server-side protocols (chainsync, block-fetch, tx-submission; plus local protocols in NtC) are done or never started. Current code always returns err when c.server is true.
- As client, we should check our client-side protocols, not server-side, to decide whether EOF is unexpected.
- Also handle io.ErrUnexpectedEOF (or ensure the muxer remaps it to io.EOF).
Proposed fix:
func (c *Connection) handleConnectionError(err error) error {
if err == nil {
return nil
}
- // Only propagate EOF errors when acting as a client with active server-side protocols
- if errors.Is(err, io.EOF) {
- // Check if we have any active server-side protocols
- if c.server {
- return err
- }
-
- // For clients, only propagate EOF if we have active server protocols
- hasActiveServerProtocols := false
- if c.chainSync != nil && c.chainSync.Server != nil && !c.chainSync.Server.IsDone() {
- hasActiveServerProtocols = true
- }
- if c.blockFetch != nil && c.blockFetch.Server != nil && !c.blockFetch.Server.IsDone() {
- hasActiveServerProtocols = true
- }
- if c.txSubmission != nil && c.txSubmission.Server != nil && !c.txSubmission.Server.IsDone() {
- hasActiveServerProtocols = true
- }
- if c.localStateQuery != nil && c.localStateQuery.Server != nil && !c.localStateQuery.Server.IsDone() {
- hasActiveServerProtocols = true
- }
- if c.localTxMonitor != nil && c.localTxMonitor.Server != nil && !c.localTxMonitor.Server.IsDone() {
- hasActiveServerProtocols = true
- }
- if c.localTxSubmission != nil && c.localTxSubmission.Server != nil && !c.localTxSubmission.Server.IsDone() {
- hasActiveServerProtocols = true
- }
-
- if hasActiveServerProtocols {
- return err
- }
-
- // EOF with no active server protocols is normal connection closure
- return nil
- }
+ // Treat EOF/UnexpectedEOF as connection closed, decide based on active protocols for our role
+ if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+ hasActive := false
+ if c.server {
+ // Server: check server-side protocols
+ if c.chainSync != nil && c.chainSync.Server != nil && !c.chainSync.Server.IsDone() {
+ hasActive = true
+ }
+ if c.blockFetch != nil && c.blockFetch.Server != nil && !c.blockFetch.Server.IsDone() {
+ hasActive = true
+ }
+ if c.txSubmission != nil && c.txSubmission.Server != nil && !c.txSubmission.Server.IsDone() {
+ hasActive = true
+ }
+ if c.localStateQuery != nil && c.localStateQuery.Server != nil && !c.localStateQuery.Server.IsDone() {
+ hasActive = true
+ }
+ if c.localTxMonitor != nil && c.localTxMonitor.Server != nil && !c.localTxMonitor.Server.IsDone() {
+ hasActive = true
+ }
+ if c.localTxSubmission != nil && c.localTxSubmission.Server != nil && !c.localTxSubmission.Server.IsDone() {
+ hasActive = true
+ }
+ } else {
+ // Client: check client-side protocols
+ if c.chainSync != nil && c.chainSync.Client != nil && !c.chainSync.Client.IsDone() {
+ hasActive = true
+ }
+ if c.blockFetch != nil && c.blockFetch.Client != nil && !c.blockFetch.Client.IsDone() {
+ hasActive = true
+ }
+ if c.txSubmission != nil && c.txSubmission.Client != nil && !c.txSubmission.Client.IsDone() {
+ hasActive = true
+ }
+ if c.localStateQuery != nil && c.localStateQuery.Client != nil && !c.localStateQuery.Client.IsDone() {
+ hasActive = true
+ }
+ if c.localTxMonitor != nil && c.localTxMonitor.Client != nil && !c.localTxMonitor.Client.IsDone() {
+ hasActive = true
+ }
+ if c.localTxSubmission != nil && c.localTxSubmission.Client != nil && !c.localTxSubmission.Client.IsDone() {
+ hasActive = true
+ }
+ }
+ // Propagate only if a protocol is still active; otherwise swallow as normal closure
+ if hasActive {
+ return err
+ }
+ return nil
+ }
// For non-EOF errors, always propagate
return err
}To confirm muxer behavior for UnexpectedEOF remapping, run:
Also applies to: 335-347
🏁 Script executed:
#!/bin/bash
# Check if muxer remaps io.ErrUnexpectedEOF to io.EOF
rg -nP 'ErrUnexpectedEOF|UnexpectedEOF' muxer/ --context 2 || trueLength of output: 0
🏁 Script executed:
#!/bin/bash
# 1. Check if muxer directory exists and list structure
echo "=== Checking muxer directory ==="
fd --type d 'muxer' . || echo "No muxer directory found"
# 2. Search for EOF handling patterns more broadly in codebase
echo -e "\n=== Searching for EOF handling patterns ==="
rg -n 'io\.EOF|ErrUnexpectedEOF' --type go -B2 -A2 || echo "No EOF patterns found"
# 3. Verify current connection.go exists and show lines 253-297
echo -e "\n=== Verifying connection.go current state ==="
if [ -f connection.go ]; then
sed -n '253,297p' connection.go
else
echo "connection.go not found at root; searching for it..."
fd --type f 'connection.go' . --exec head -20 {}
fiLength of output: 5839
🌐 Web query:
Go io.ErrUnexpectedEOF error type documentation
💡 Result:
io.ErrUnexpectedEOF
-
What it is: a package-level error variable defined as
var ErrUnexpectedEOF = errors.New("unexpected EOF")
meaning EOF was encountered in the middle of reading a fixed-size block or structured data. [1] -
Where it’s declared: package io in the Go standard library. [1]
-
When it’s returned: commonly by helpers that require reading an exact number of bytes (e.g. io.ReadFull, io.ReadAtLeast) and by packages that read fixed-size binary structures (e.g. encoding/binary) when EOF occurs after some but not all bytes were read. If no bytes were read, functions typically return io.EOF instead. [2][3]
-
How to check it: since Go 1.13 errors may be wrapped, use errors.Is(err, io.ErrUnexpectedEOF) to detect it robustly. (Older code sometimes used err == io.ErrUnexpectedEOF.) [4]
Sources:
[1] io package docs (ErrUnexpectedEOF definition).
[2] io.ReadFull / ReadAtLeast behavior (returns ErrUnexpectedEOF when partial read then EOF).
[3] encoding/binary.Read behavior (returns io.ErrUnexpectedEOF if EOF after some bytes).
[4] Go error-handling FAQ (use errors.Is for comparisons).
🏁 Script executed:
#!/bin/bash
# Get current state of connection.go lines 250-350 for full context
echo "=== git/connection.go lines 250-350 ==="
sed -n '250,350p' git/connection.goLength of output: 100
🏁 Script executed:
#!/bin/bash
# Find connection.go and show lines 250-350
echo "=== Locating connection.go ==="
CONN_FILE=$(fd -t f 'connection\.go$' 2>/dev/null | head -1)
echo "Found at: $CONN_FILE"
if [ -n "$CONN_FILE" ]; then
echo -e "\n=== $CONN_FILE lines 250-350 ==="
sed -n '250,350p' "$CONN_FILE"
else
echo "connection.go not found"
fiLength of output: 3089
🏁 Script executed:
#!/bin/bash
# Check lines 335-347 for the "also applies to" section
CONN_FILE=$(fd -t f 'connection\.go$' 2>/dev/null | head -1)
if [ -n "$CONN_FILE" ]; then
echo "=== $CONN_FILE lines 330-355 (context for 335-347) ==="
sed -n '330,355p' "$CONN_FILE"
fiLength of output: 802
Fix EOF handling logic: server/client role checks are inverted
The current code has the logic backwards:
- Server (line 266): returns EOF unconditionally without checking active server-side protocols
- Client (lines 271–287): checks server-side protocol status instead of client-side
This means servers always propagate EOF (even when protocols are active), and clients check the wrong role's protocols. The proposed fix correctly swaps the logic: server checks Server fields, client checks Client fields. Additionally, handle io.ErrUnexpectedEOF alongside io.EOF, as the protocol layer already uses it for partial reads.
Apply the proposed diff and ensure the muxer behavior aligns with the corrected logic.
🤖 Prompt for AI Agents
In connection.go around lines 253 to 297, the EOF handling logic is inverted and
misses io.ErrUnexpectedEOF; change the EOF check to treat both io.EOF and
io.ErrUnexpectedEOF the same, and swap the role-specific checks so that when
c.server is true you inspect the protocol .Server fields for active server-side
protocols, and when c.server is false (client) you inspect the protocol .Client
fields for active client-side protocols; if any active protocols are found
return the error, otherwise treat EOF/ErrUnexpectedEOF as normal closure and
return nil.
| import ( | ||
| "io" | ||
| "log/slog" | ||
| "net" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/blinklabs-io/gouroboros/connection" | ||
| "github.com/blinklabs-io/gouroboros/muxer" | ||
| "github.com/blinklabs-io/gouroboros/protocol" | ||
| "github.com/blinklabs-io/gouroboros/protocol/common" | ||
| "github.com/stretchr/testify/assert" | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Import sync; avoid busy-read behavior.
You’ll add sync for mutex/once and fix Read to block/EOF on close to prevent spin loops in muxer readers.
Apply this diff:
import (
"io"
"log/slog"
"net"
+ "sync"
"testing"
"time"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| import ( | |
| "io" | |
| "log/slog" | |
| "net" | |
| "testing" | |
| "time" | |
| "github.com/blinklabs-io/gouroboros/connection" | |
| "github.com/blinklabs-io/gouroboros/muxer" | |
| "github.com/blinklabs-io/gouroboros/protocol" | |
| "github.com/blinklabs-io/gouroboros/protocol/common" | |
| "github.com/stretchr/testify/assert" | |
| ) | |
| import ( | |
| "io" | |
| "log/slog" | |
| "net" | |
| "sync" | |
| "testing" | |
| "time" | |
| "github.com/blinklabs-io/gouroboros/connection" | |
| "github.com/blinklabs-io/gouroboros/muxer" | |
| "github.com/blinklabs-io/gouroboros/protocol" | |
| "github.com/blinklabs-io/gouroboros/protocol/common" | |
| "github.com/stretchr/testify/assert" | |
| ) |
🤖 Prompt for AI Agents
In protocol/chainsync/chainsync_test.go around lines 17 to 29, add the missing
"sync" import and modify the muxer reader Read implementation to avoid
busy-looping: introduce a sync.Mutex (and sync.Once where appropriate) and a
condition or channel to block Read until data is available or the stream is
closed; on close, ensure pending Reads unblock and return io.EOF (after draining
any buffered data) instead of spinning. Update any close logic to signal the
condition/channel and use the mutex to protect shared state so Read properly
blocks and exits with EOF on close.
| type testConn struct { | ||
| writeChan chan []byte | ||
| closed bool | ||
| closeChan chan struct{} | ||
| } | ||
|
|
||
| func newTestConn() *testConn { | ||
| return &testConn{ | ||
| writeChan: make(chan []byte, 100), | ||
| closeChan: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil } | ||
| func (c *testConn) Write(b []byte) (n int, err error) { | ||
| select { | ||
| case c.writeChan <- b: | ||
| return len(b), nil | ||
| case <-c.closeChan: | ||
| return 0, io.EOF | ||
| } | ||
| } | ||
| func (c *testConn) Close() error { | ||
| if !c.closed { | ||
| close(c.closeChan) | ||
| c.closed = true | ||
| } | ||
| return nil | ||
| } | ||
| func (c *testConn) LocalAddr() net.Addr { return testAddr{} } | ||
| func (c *testConn) RemoteAddr() net.Addr { return testAddr{} } | ||
| func (c *testConn) SetDeadline(t time.Time) error { return nil } | ||
| func (c *testConn) SetReadDeadline(t time.Time) error { return nil } | ||
| func (c *testConn) SetWriteDeadline(t time.Time) error { return nil } |
There was a problem hiding this comment.
Harden testConn: thread‑safe close/write and non‑spinning Read.
Current Read returns (0,nil) and Close/Write aren’t synchronized. This can spin a reader and race on close.
Apply this diff:
type testConn struct {
- writeChan chan []byte
- closed bool
- closeChan chan struct{}
+ writeChan chan []byte
+ closed bool
+ closeChan chan struct{}
+ mu sync.Mutex
+ closeOnce sync.Once
}
func newTestConn() *testConn {
return &testConn{
writeChan: make(chan []byte, 100),
closeChan: make(chan struct{}),
}
}
-func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil }
+func (c *testConn) Read(b []byte) (n int, err error) {
+ // Block until closed; avoids busy-loop in readers
+ <-c.closeChan
+ return 0, io.EOF
+}
func (c *testConn) Write(b []byte) (n int, err error) {
select {
case c.writeChan <- b:
return len(b), nil
case <-c.closeChan:
return 0, io.EOF
}
}
func (c *testConn) Close() error {
- if !c.closed {
- close(c.closeChan)
- c.closed = true
- }
+ c.closeOnce.Do(func() {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ close(c.closeChan)
+ c.closed = true
+ })
return nil
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| type testConn struct { | |
| writeChan chan []byte | |
| closed bool | |
| closeChan chan struct{} | |
| } | |
| func newTestConn() *testConn { | |
| return &testConn{ | |
| writeChan: make(chan []byte, 100), | |
| closeChan: make(chan struct{}), | |
| } | |
| } | |
| func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil } | |
| func (c *testConn) Write(b []byte) (n int, err error) { | |
| select { | |
| case c.writeChan <- b: | |
| return len(b), nil | |
| case <-c.closeChan: | |
| return 0, io.EOF | |
| } | |
| } | |
| func (c *testConn) Close() error { | |
| if !c.closed { | |
| close(c.closeChan) | |
| c.closed = true | |
| } | |
| return nil | |
| } | |
| func (c *testConn) LocalAddr() net.Addr { return testAddr{} } | |
| func (c *testConn) RemoteAddr() net.Addr { return testAddr{} } | |
| func (c *testConn) SetDeadline(t time.Time) error { return nil } | |
| func (c *testConn) SetReadDeadline(t time.Time) error { return nil } | |
| func (c *testConn) SetWriteDeadline(t time.Time) error { return nil } | |
| type testConn struct { | |
| writeChan chan []byte | |
| closed bool | |
| closeChan chan struct{} | |
| mu sync.Mutex | |
| closeOnce sync.Once | |
| } | |
| func newTestConn() *testConn { | |
| return &testConn{ | |
| writeChan: make(chan []byte, 100), | |
| closeChan: make(chan struct{}), | |
| } | |
| } | |
| func (c *testConn) Read(b []byte) (n int, err error) { | |
| // Block until closed; avoids busy-loop in readers | |
| <-c.closeChan | |
| return 0, io.EOF | |
| } | |
| func (c *testConn) Write(b []byte) (n int, err error) { | |
| select { | |
| case c.writeChan <- b: | |
| return len(b), nil | |
| case <-c.closeChan: | |
| return 0, io.EOF | |
| } | |
| } | |
| func (c *testConn) Close() error { | |
| c.closeOnce.Do(func() { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| close(c.closeChan) | |
| c.closed = true | |
| }) | |
| return nil | |
| } | |
| func (c *testConn) LocalAddr() net.Addr { return testAddr{} } | |
| func (c *testConn) RemoteAddr() net.Addr { return testAddr{} } | |
| func (c *testConn) SetDeadline(t time.Time) error { return nil } | |
| func (c *testConn) SetReadDeadline(t time.Time) error { return nil } | |
| func (c *testConn) SetWriteDeadline(t time.Time) error { return nil } |
🤖 Prompt for AI Agents
In protocol/chainsync/chainsync_test.go around lines 36 to 69, the testConn
implementation is unsafe: Read currently returns (0, nil) (which can spin a
reader) and Write/Close race on the closed flag; fix by making Read block and
return data or io.EOF (read from the existing writeChan and copy into the
provided buffer, returning the number of bytes read and nil, and return io.EOF
when closeChan is closed), and make Close/Write thread-safe by protecting the
closed state with synchronization (use a mutex or atomics) or by relying on the
closeChan only for signaling so writes detect closed and return io.EOF and Close
becomes idempotent and safe to call concurrently. Ensure no goroutine spins when
connection is closed and that Write returns appropriate errors when closed.
| func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil } | ||
| func (c *testConn) Close() error { | ||
| c.closeOnce.Do(func() { | ||
| c.mu.Lock() | ||
| defer c.mu.Unlock() | ||
| close(c.closeChan) | ||
| c.closed = true | ||
| }) | ||
| return nil | ||
| } |
There was a problem hiding this comment.
Make Read blocking and Close idempotent/thread‑safe.
Avoid (0,nil) Read and data races on close.
-func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil }
+func (c *testConn) Read(b []byte) (n int, err error) {
+ <-c.closeChan
+ return 0, io.EOF
+}
func (c *testConn) Close() error {
c.closeOnce.Do(func() {
c.mu.Lock()
defer c.mu.Unlock()
close(c.closeChan)
c.closed = true
})
return nil
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func (c *testConn) Read(b []byte) (n int, err error) { return 0, nil } | |
| func (c *testConn) Close() error { | |
| c.closeOnce.Do(func() { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| close(c.closeChan) | |
| c.closed = true | |
| }) | |
| return nil | |
| } | |
| func (c *testConn) Read(b []byte) (n int, err error) { | |
| <-c.closeChan | |
| return 0, io.EOF | |
| } | |
| func (c *testConn) Close() error { | |
| c.closeOnce.Do(func() { | |
| c.mu.Lock() | |
| defer c.mu.Unlock() | |
| close(c.closeChan) | |
| c.closed = true | |
| }) | |
| return nil | |
| } |
| func getTestProtocolOptions(conn net.Conn) protocol.ProtocolOptions { | ||
| mux := muxer.New(conn) | ||
| return protocol.ProtocolOptions{ | ||
| ConnectionId: connection.ConnectionId{ | ||
| LocalAddr: testAddr{}, | ||
| RemoteAddr: testAddr{}, | ||
| }, | ||
| Muxer: mux, | ||
| Logger: slog.Default(), | ||
| } | ||
| } |
There was a problem hiding this comment.
Start/stop muxer so writes reach the fake conn.
Without starting the muxer, client/server SendMessage may not hit conn.Write.
func getTestProtocolOptions(conn net.Conn) protocol.ProtocolOptions {
- mux := muxer.New(conn)
- return protocol.ProtocolOptions{
+ mux := muxer.New(conn)
+ go mux.Start()
+ go func() {
+ <-conn.(*testConn).closeChan
+ mux.Stop()
+ }()
+ return protocol.ProtocolOptions{
ConnectionId: connection.ConnectionId{
LocalAddr: testAddr{},
RemoteAddr: testAddr{},
},
Muxer: mux,
Logger: slog.Default(),
+ Mode: protocol.ProtocolModeNodeToNode,
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func getTestProtocolOptions(conn net.Conn) protocol.ProtocolOptions { | |
| mux := muxer.New(conn) | |
| return protocol.ProtocolOptions{ | |
| ConnectionId: connection.ConnectionId{ | |
| LocalAddr: testAddr{}, | |
| RemoteAddr: testAddr{}, | |
| }, | |
| Muxer: mux, | |
| Logger: slog.Default(), | |
| } | |
| } | |
| func getTestProtocolOptions(conn net.Conn) protocol.ProtocolOptions { | |
| mux := muxer.New(conn) | |
| go mux.Start() | |
| go func() { | |
| <-conn.(*testConn).closeChan | |
| mux.Stop() | |
| }() | |
| return protocol.ProtocolOptions{ | |
| ConnectionId: connection.ConnectionId{ | |
| LocalAddr: testAddr{}, | |
| RemoteAddr: testAddr{}, | |
| }, | |
| Muxer: mux, | |
| Logger: slog.Default(), | |
| Mode: protocol.ProtocolModeNodeToNode, | |
| } | |
| } |
🤖 Prompt for AI Agents
In protocol/txsubmission/txsubmission_test.go around lines 81 to 91, the created
muxer is never started so writes from client/server won't reach the fake
connection; after creating the muxer call its Start() (or equivalent start
method) before returning the ProtocolOptions and ensure tests stop it (call
Close()/Stop() or register t.Cleanup to stop the muxer) so the muxer is running
for the duration of the test and is properly torn down afterwards.
Signed-off-by: Jenita <jkawan@blinklabs.io>
There was a problem hiding this comment.
Actionable comments posted: 6
♻️ Duplicate comments (1)
connection_test.go (1)
225-231: NtN handshake: align roles and remove client–client pattern.oConn is a client (no WithServer(true)), but mock is also ProtocolRoleClient and sends a handshake response. That’s client–client with the response from the wrong side and can stall the Propose state. This mirrors earlier feedback.
Fix one of the two ways; simplest here: keep oConn as client, make mock the server, and only send the NtN response:
- mockConn := ouroboros_mock.NewConnection( - ouroboros_mock.ProtocolRoleClient, + mockConn := ouroboros_mock.NewConnection( + ouroboros_mock.ProtocolRoleServer, []ouroboros_mock.ConversationEntry{ - ouroboros_mock.ConversationEntryHandshakeRequestGeneric, - ouroboros_mock.ConversationEntryHandshakeNtNResponse, + ouroboros_mock.ConversationEntryHandshakeNtNResponse, }, )
🧹 Nitpick comments (2)
connection_test.go (2)
34-36: Reduce goleak false positives by ignoring known mock runner goroutine (optional).If leaks persist after proper shutdown, ignore the mock runner top‑function to stabilize CI.
Example:
defer goleak.VerifyNone(t, goleak.IgnoreTopFunction("github.com/blinklabs-io/ouroboros-mock.(*Connection).run"), )Use only if cleanup is correct and a stable residual goroutine remains from the mock.
Also applies to: 223-229, 263-266
103-112: Avoid spin‑wait; replace with a deterministic wait helper.Polling oConn.ChainSync() is flaky. Prefer a helper that blocks on a ready signal (e.g., from the callback barrier) or exposes a handshake‑finished channel.
I can provide a small test helper like waitForChainSyncServer(t, oConn, 1*time.Second) that uses a barrier instead of sleeps.
Also applies to: 185-194
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
connection_test.go(2 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
connection_test.go (8)
connection.go (2)
NewConnection(101-124)New(127-129)protocol/handshake/messages.go (2)
NewMsgProposeVersions(64-80)NewMsgAcceptVersion(88-102)protocol/versions.go (2)
ProtocolVersionMap(24-24)ProtocolVersionNtCOffset(20-20)protocol/versiondata.go (1)
VersionDataNtC9to14(48-48)protocol/chainsync/messages.go (1)
NewMsgFindIntersect(224-232)protocol/chainsync/chainsync.go (4)
ProtocolIdNtC(31-31)New(232-240)NewConfig(246-261)ChainSync(194-197)protocol/common/types.go (1)
Point(23-28)connection_options.go (5)
WithConnection(36-40)WithNetworkMagic(50-54)WithServer(64-68)WithChainSyncConfig(131-135)WithNodeToNode(78-82)
🪛 GitHub Actions: go-test
connection_test.go
[error] 205-219: TestErrorHandlingWithActiveProtocols: timed out waiting for protocol to stop. Found large set of unexpected goroutines traces in the test output.
[error] 254-260: TestErrorHandlingWithMultipleProtocols: Received connection error with multiple active protocols: EOF. Unexpected goroutines reported in test.
[error] 300-300: TestBasicErrorHandling: Unexpected goroutines reported during test teardown (goroutine leaks detected).
| chainsync.WithFindIntersectFunc( | ||
| func(ctx chainsync.CallbackContext, points []common.Point) (common.Point, chainsync.Tip, error) { | ||
| // We need to block here to keep the protocol active | ||
| time.Sleep(5 * time.Second) | ||
| return common.Point{}, chainsync.Tip{}, fmt.Errorf("context cancelled") | ||
| }, | ||
| ), | ||
| ), |
There was a problem hiding this comment.
Remove 5s sleep; block on protocol context and signal readiness to avoid leaks/flakes.
time.Sleep(5s) keeps goroutines alive and causes goleak failures. Use a barrier to know CS is active, then block on context cancellation so shutdown is prompt.
Apply:
- chainsync.WithFindIntersectFunc(
- func(ctx chainsync.CallbackContext, points []common.Point) (common.Point, chainsync.Tip, error) {
- // We need to block here to keep the protocol active
- time.Sleep(5 * time.Second)
- return common.Point{}, chainsync.Tip{}, fmt.Errorf("context cancelled")
- },
- ),
+ chainsync.WithFindIntersectFunc(
+ func(ctx chainsync.CallbackContext, points []common.Point) (common.Point, chainsync.Tip, error) {
+ // Signal test that ChainSync is active
+ if entered := ctx.Value("enteredFindIntersect"); entered != nil {
+ if ch, ok := entered.(chan struct{}); ok {
+ select { case <-ch: default: close(ch) }
+ }
+ }
+ // Wait for shutdown instead of sleeping
+ type hasContext interface{ Context() interface{ Done() <-chan struct{}; Err() error } }
+ if hc, ok := any(ctx).(hasContext); ok {
+ <-hc.Context().Done()
+ return common.Point{}, chainsync.Tip{}, hc.Context().Err()
+ }
+ // Fallback: short wait to keep protocol "active" without leaking
+ <-time.After(250 * time.Millisecond)
+ return common.Point{}, chainsync.Tip{}, fmt.Errorf("closed")
+ },
+ ),And create/await the barrier near the test:
- // Wait a bit for protocol to start
- time.Sleep(100 * time.Millisecond)
+ // Barrier to know callback ran
+ enteredFindIntersect := make(chan struct{})
+ oConn.SetValue("enteredFindIntersect", enteredFindIntersect) // helper on Connection to pass into ctx; or thread via cfg if available
+ select {
+ case <-enteredFindIntersect:
+ case <-time.After(1 * time.Second):
+ t.Fatal("timeout waiting for FindIntersect to start")
+ }If SetValue isn’t available, thread the channel via your ChainSync config’s context or a package-level var for the test only. Based on learnings.
Also applies to: 118-123
🤖 Prompt for AI Agents
In connection_test.go around lines 89-96 (and similarly at 118-123), replace the
hardcoded time.Sleep(5 * time.Second) with a synchronization barrier that
signals when the ChainSync callback is active and then blocks on the
protocol/test context; specifically, create a ready channel or waitgroup before
starting the protocol, have the FindIntersectFunc close/send on that channel to
indicate readiness, and then wait on ctx.Done() (or the passed protocol context)
instead of sleeping so the goroutine exits promptly on cancellation; if you
cannot SetValue on the context, pass the ready channel into the ChainSync test
config or use a test-only package-level channel to coordinate.
| // Send Done message to stop the protocol | ||
| ouroboros_mock.ConversationEntryOutput{ | ||
| ProtocolId: chainsync.ProtocolIdNtC, | ||
| Messages: []protocol.Message{chainsync.NewMsgDone()}, | ||
| }, | ||
| }, | ||
| ) |
There was a problem hiding this comment.
Ensure ChainSync actually starts before Done; assert no error after stopped.
Currently only Done is sent; if the server loop never starts, DoneChan may not close, causing the timeout. Also, the test logs an error instead of failing when one is received.
Apply:
- // Send Done message to stop the protocol
- ouroboros_mock.ConversationEntryOutput{
- ProtocolId: chainsync.ProtocolIdNtC,
- Messages: []protocol.Message{chainsync.NewMsgDone()},
- },
+ // Start ChainSync, then stop it
+ ouroboros_mock.ConversationEntryOutput{
+ ProtocolId: chainsync.ProtocolIdNtC,
+ Messages: []protocol.Message{
+ chainsync.NewMsgFindIntersect(
+ []common.Point{{Slot: 21600, Hash: []byte("19297addad3da631einos029")}},
+ ),
+ chainsync.NewMsgDone(),
+ },
+ },- // Wait for protocol to be done (Done message from mock should trigger this)
+ // Wait for protocol to be done (Done message should trigger this)
select {
case <-chainSyncProtocol.Server.DoneChan():
- // Protocol is stoppeds
- case <-time.After(1 * time.Second):
+ // Protocol is stopped
+ case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for protocol to stop")
}- // Now close the mock connection to generate an error
+ // Now close the mock connection; errors after all protocols stopped should be ignored
mockConn.Close()
select {
- case err := <-oConn.ErrorChan():
- t.Logf("Received error during shutdown: %s", err)
+ case err := <-oConn.ErrorChan():
+ t.Fatalf("unexpected connection error after protocols stopped: %v", err)
case <-time.After(500 * time.Millisecond):
- t.Log("No connection error received (expected when protocols are stopped)")
+ // OK: no error
}Also applies to: 200-206, 210-215
| // Protocol is stoppeds | ||
| case <-time.After(1 * time.Second): |
There was a problem hiding this comment.
Minor: fix typo in comment.
“stoppeds” → “stopped”.
- // Protocol is stoppeds
+ // Protocol is stopped📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // Protocol is stoppeds | |
| case <-time.After(1 * time.Second): | |
| // Protocol is stopped | |
| case <-time.After(1 * time.Second): |
🤖 Prompt for AI Agents
In connection_test.go around lines 203 to 204, there's a typo in the comment
("stoppeds"); change the comment text to "stopped" by replacing "stoppeds" with
"stopped" so the comment reads correctly.
| select { | ||
| case err := <-oConn.ErrorChan(): | ||
| t.Logf("Received error during shutdown: %s", err) | ||
| case <-time.After(500 * time.Millisecond): | ||
| t.Log("No connection error received (expected when protocols are stopped)") | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Tighten assertions on ErrorChan to match PR behavior.
Tests should fail on unexpected errors rather than log them, ensuring semantics don’t regress silently.
See diffs above where logging was replaced with t.Fatalf on any received error in “stopped/no‑traffic” scenarios.
Also applies to: 249-257
🤖 Prompt for AI Agents
In connection_test.go around lines 210-215, the test currently logs any error
received from oConn.ErrorChan() during shutdown, but the review requires
tightening assertions so unexpected errors fail the test; change the select
branch that currently calls t.Logf("Received error during shutdown: %s", err) to
call t.Fatalf with the same message to fail on errors, and make the identical
replacement for the other occurrence mentioned at lines ~249-257 so both
“stopped/no-traffic” scenarios fail the test when an error is received instead
of merely logging it.
Signed-off-by: Jenita <jkawan@blinklabs.io>
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
protocol/chainsync/client.go (1)
122-137: Fix panic: send on closed channel in ChainSync clientreadyForNextBlockChan is closed on Done, but handleRollForward/Backward can still send, causing “send on closed channel” (seen in CI). Don’t close the channel; instead, make syncLoop exit on Done.
Apply:
@@ func (c *Client) Start() { - go func() { - <-c.DoneChan() - close(c.readyForNextBlockChan) - }() + go func() { + <-c.DoneChan() + // No channel close; syncLoop exits on Done + }() @@ func (c *Client) syncLoop() { - for { - // Wait for a block to be received - if ready, ok := <-c.readyForNextBlockChan; !ok { - // Channel is closed, which means we're shutting down - return - } else if !ready { - // Sync was cancelled - return - } + for { + // Wait for a block to be received or shutdown + var ready bool + select { + case <-c.DoneChan(): + return + case ready = <-c.readyForNextBlockChan: + // continue + } + if !ready { + // Sync was cancelled + return + } @@ func (c *Client) handleRollForward(...) error { - // Signal that we're ready for the next block - c.readyForNextBlockChan <- true + // Signal that we're ready for the next block + c.readyForNextBlockChan <- true return nil @@ func (c *Client) handleRollBackward(msg protocol.Message) error { - // Signal that we're ready for the next block - c.readyForNextBlockChan <- true + // Signal that we're ready for the next block + c.readyForNextBlockChan <- true return nilNote: Since we no longer close the channel, no additional guards are required at send sites; syncLoop exits via Done.
Also applies to: 440-456, 724-736, 765-767
♻️ Duplicate comments (4)
connection_test.go (3)
207-214: Tighten assertion: any error after protocols stopped should failWhen protocols are stopped, closing the connection must not surface an error. Fail the test if any error is received instead of logging.
- select { - case err := <-oConn.ErrorChan(): - t.Logf("Received error during shutdown: %s", err) - case <-time.After(500 * time.Millisecond): - t.Log("No connection error received (expected when protocols are stopped)") - } + select { + case err := <-oConn.ErrorChan(): + t.Fatalf("unexpected connection error after protocols stopped: %v", err) + case <-time.After(500 * time.Millisecond): + // OK: no error + }
224-256: Fix NtN handshake role mismatch and test semantics; expect no error when no trafficThe mock sends a handshake response while oConn is not configured as server, causing client–client stalemate risk. Also, this test asserts an error without exchanging any mini‑protocol messages, which contradicts the PR behavior (“started but no messages exchanged” → no error).
Apply:
mockConn := ouroboros_mock.NewConnection( ouroboros_mock.ProtocolRoleClient, []ouroboros_mock.ConversationEntry{ - ouroboros_mock.ConversationEntryHandshakeRequestGeneric, - ouroboros_mock.ConversationEntryHandshakeNtNResponse, + // Client proposes; server (our oConn) will respond + ouroboros_mock.ConversationEntryHandshakeRequestGeneric, }, ) oConn, err := ouroboros.New( ouroboros.WithConnection(mockConn), ouroboros.WithNetworkMagic(ouroboros_mock.MockNetworkMagic), - ouroboros.WithNodeToNode(true), + ouroboros.WithNodeToNode(true), + ouroboros.WithServer(true), ) @@ -// Wait for handshake to complete -time.Sleep(100 * time.Millisecond) +// Optionally wait briefly or add a handshake barrier if available +time.Sleep(100 * time.Millisecond) @@ -// Should receive error since protocols are active -select { -case err := <-oConn.ErrorChan(): - if err == nil { - t.Fatal("expected connection error, got nil") - } - t.Logf("Received connection error with multiple active protocols: %s", err) -case <-time.After(2 * time.Second): - t.Error("timed out waiting for connection error") -} +// No mini‑protocol traffic: closing should not emit an error +select { +case err := <-oConn.ErrorChan(): + t.Fatalf("unexpected error with no active protocol traffic: %v", err) +case <-time.After(750 * time.Millisecond): + // OK: no error +}
199-205: Typo in comment“stoppeds” → “stopped”.
- // Protocol is stoppeds + // Protocol is stoppedconnection.go (1)
256-300: Fix inverted EOF handling logic and include ErrUnexpectedEOF as connection closureThe logic for determining when to propagate EOF is backwards. When acting as a server (
c.server=true), the code unconditionally returns the error instead of checking if server-side protocols are active. Conversely, when acting as a client, it checks server-side protocol states instead of client-side. This inverts the intended behavior and will suppress errors that should propagate.Additionally,
io.ErrUnexpectedEOFshould be treated alongsideio.EOFas a normal connection closure signal, since both indicate the connection is closed (protocol.go currently handles it as an incomplete message state, but at the connection level it signals closure).Apply:
func (c *Connection) handleConnectionError(err error) error { if err == nil { return nil } - // Only propagate EOF errors when acting as a client with active server-side protocols - if errors.Is(err, io.EOF) { - // Check if we have any active server-side protocols - if c.server { - return err - } - - // For clients, only propagate EOF if we have active server protocols - hasActiveServerProtocols := false - if c.chainSync != nil && c.chainSync.Server != nil && !c.chainSync.Server.IsDone() { - hasActiveServerProtocols = true - } - if c.blockFetch != nil && c.blockFetch.Server != nil && !c.blockFetch.Server.IsDone() { - hasActiveServerProtocols = true - } - if c.txSubmission != nil && c.txSubmission.Server != nil && !c.txSubmission.Server.IsDone() { - hasActiveServerProtocols = true - } - if c.localStateQuery != nil && c.localStateQuery.Server != nil && !c.localStateQuery.Server.IsDone() { - hasActiveServerProtocols = true - } - if c.localTxMonitor != nil && c.localTxMonitor.Server != nil && !c.localTxMonitor.Server.IsDone() { - hasActiveServerProtocols = true - } - if c.localTxSubmission != nil && c.localTxSubmission.Server != nil && !c.localTxSubmission.Server.IsDone() { - hasActiveServerProtocols = true - } - - if hasActiveServerProtocols { - return err - } - - // EOF with no active server protocols is normal connection closure - return nil - } + // Treat EOF/UnexpectedEOF as normal closure unless our role still has active protocols + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + hasActive := false + if c.server { + // Server: check server-side protocols + if c.chainSync != nil && c.chainSync.Server != nil && !c.chainSync.Server.IsDone() { hasActive = true } + if c.blockFetch != nil && c.blockFetch.Server != nil && !c.blockFetch.Server.IsDone() { hasActive = true } + if c.txSubmission != nil && c.txSubmission.Server != nil && !c.txSubmission.Server.IsDone() { hasActive = true } + if c.localStateQuery != nil && c.localStateQuery.Server != nil && !c.localStateQuery.Server.IsDone() { hasActive = true } + if c.localTxMonitor != nil && c.localTxMonitor.Server != nil && !c.localTxMonitor.Server.IsDone() { hasActive = true } + if c.localTxSubmission != nil && c.localTxSubmission.Server != nil && !c.localTxSubmission.Server.IsDone() { hasActive = true } + } else { + // Client: check client-side protocols + if c.chainSync != nil && c.chainSync.Client != nil && !c.chainSync.Client.IsDone() { hasActive = true } + if c.blockFetch != nil && c.blockFetch.Client != nil && !c.blockFetch.Client.IsDone() { hasActive = true } + if c.txSubmission != nil && c.txSubmission.Client != nil && !c.txSubmission.Client.IsDone() { hasActive = true } + if c.localStateQuery != nil && c.localStateQuery.Client != nil && !c.localStateQuery.Client.IsDone() { hasActive = true } + if c.localTxMonitor != nil && c.localTxMonitor.Client != nil && !c.localTxMonitor.Client.IsDone() { hasActive = true } + if c.localTxSubmission != nil && c.localTxSubmission.Client != nil && !c.localTxSubmission.Client.IsDone() { hasActive = true } + } + if hasActive { + return err + } + return nil + } // For non-EOF errors, always propagate return err }
🧹 Nitpick comments (1)
connection_test.go (1)
117-119: Avoid sleep-based readiness; use a barrier from the callbackReplace time.Sleep with a readiness signal from the FindIntersect callback to make the test deterministic and goleak-safe.
Example:
- // Wait a bit for protocol to start - time.Sleep(100 * time.Millisecond) + // Barrier to ensure callback entered + started := make(chan struct{}, 1) + // pass via config context or test-scoped var; e.g., wrap FindIntersectFunc to close(started) + // then here: + select { + case <-started: + case <-time.After(1 * time.Second): + t.Fatal("timeout waiting for ChainSync to start") + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
connection.go(5 hunks)connection_test.go(2 hunks)protocol/chainsync/chainsync.go(2 hunks)protocol/chainsync/client.go(2 hunks)protocol/chainsync/server.go(1 hunks)protocol/protocol.go(9 hunks)
🧰 Additional context used
🧬 Code graph analysis (4)
connection_test.go (6)
connection.go (2)
NewConnection(104-127)New(130-132)protocol/protocol.go (2)
ProtocolRoleClient(91-91)New(121-132)protocol/chainsync/messages.go (1)
NewMsgFindIntersect(224-232)protocol/chainsync/chainsync.go (6)
ProtocolIdNtC(32-32)New(234-242)NewConfig(248-263)WithFindIntersectFunc(291-297)CallbackContext(214-219)ChainSync(195-198)protocol/common/types.go (1)
Point(23-28)connection_options.go (5)
WithConnection(36-40)WithNetworkMagic(50-54)WithServer(64-68)WithChainSyncConfig(131-135)WithNodeToNode(78-82)
connection.go (3)
protocol/chainsync/server.go (1)
Server(27-33)muxer/muxer.go (2)
ConnectionClosedError(72-75)Muxer(57-70)protocol/protocol.go (1)
ProtocolOptions(96-106)
protocol/protocol.go (2)
protocol/state.go (3)
State(32-35)StateMap(69-69)AgencyNone(26-26)protocol/message.go (1)
Message(18-22)
protocol/chainsync/client.go (1)
protocol/chainsync/messages.go (1)
NewMsgDone(270-277)
🪛 GitHub Actions: go-test
connection.go
[error] 81-81: panic: send on closed channel
🔇 Additional comments (2)
protocol/protocol.go (1)
134-153: State tracking implementation looks goodcurrentState with RW lock, CurrentState()/IsDone() (including initial state) and guarded updates in stateLoop align with lifecycle checks needed by centralized error handling.
Also applies to: 493-503, 552-571, 592-599
protocol/chainsync/chainsync.go (1)
18-26: Context propagation in callbacks looks goodEmbedding context.Context in CallbackContext enables clean cancellation and aligns with the connection’s context wiring.
Also applies to: 213-219
| s.callbackContext = CallbackContext{ | ||
| Context: protoOptions.Context, | ||
| Server: s, | ||
| ConnectionId: protoOptions.ConnectionId, | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
Verify Context is used in callbacks to differentiate clean shutdowns from errors.
The Context propagation to CallbackContext looks correct. However, to achieve the PR's objective of not returning errors on connection close when protocols are explicitly stopped, ensure that callback functions (e.g., RequestNextFunc, FindIntersectFunc) properly check s.callbackContext.Context to distinguish between clean shutdowns (context cancellation) and actual errors.
Run the following script to verify how the Context is used in callback implementations:
🏁 Script executed:
#!/bin/bash
# Description: Check if callbacks use Context to detect clean shutdowns vs errors
# Search for RequestNextFunc and FindIntersectFunc implementations
echo "=== Searching for callback function implementations ==="
rg -n -A 10 "RequestNextFunc.*func" --type=go
echo -e "\n=== Checking Context usage in chainsync callbacks ==="
rg -n -C 5 "callbackContext\.Context|ctx\.Done\(\)|ctx\.Err\(\)" --type=go -g '**/chainsync/**'
echo -e "\n=== Checking for connection close error handling ==="
rg -n -C 5 "connection.*close|ErrConnectionClosed" --type=go -g '**/chainsync/**'Length of output: 2331
🏁 Script executed:
#!/bin/bash
# Description: Deep dive into Context usage and error handling
# 1. Find and display handleRequestNext() implementation
echo "=== handleRequestNext() implementation ==="
rg -n -A 20 "func.*handleRequestNext" --type=go -g '**/chainsync/**'
# 2. Show CallbackContext struct definition
echo -e "\n=== CallbackContext struct definition ==="
rg -n -A 10 "type CallbackContext struct" --type=go -g '**/chainsync/**'
# 3. Search for all uses of callbackContext variable in server.go
echo -e "\n=== All uses of callbackContext in server.go ==="
rg -n "callbackContext" --type=go protocol/chainsync/server.go
# 4. Check if Context is referenced anywhere in callbacks or error returns
echo -e "\n=== Check Context field references in entire chainsync package ==="
rg -n "\.Context" --type=go -g '**/chainsync/**' | head -30Length of output: 3090
🏁 Script executed:
#!/bin/bash
# Description: Verify struct initialization syntax and error handling
# 1. Check if code compiles (look for build/test results)
echo "=== Checking CallbackContext initialization pattern in both files ==="
rg -n -B 2 -A 5 "CallbackContext{" --type=go -g '**/chainsync/**'
# 2. Search for handleDone() to see if Context is checked there
echo -e "\n=== handleDone() implementation ==="
rg -n -A 20 "func.*handleDone" --type=go -g '**/chainsync/**'
# 3. Check if error handling distinguishes between clean close and errors
echo -e "\n=== Error handling in protocol message handlers ==="
rg -n -B 2 "ctx\.Done\(\)|context\.Cause\(|Select\(" --type=go -g '**/chainsync/**'
# 4. Look for any context cancellation checks
echo -e "\n=== Context cancellation usage ==="
rg -n "Err\(\)|IsCanceled|IsDone" --type=go -g '**/chainsync/**'Length of output: 1097
🏁 Script executed:
#!/bin/bash
# Description: Final verification of Context initialization and usage
# 1. Check ProtocolOptions struct to verify Context field exists
echo "=== ProtocolOptions struct definition ==="
rg -n -A 15 "type ProtocolOptions struct" --type=go
# 2. Check how Context is used elsewhere in similar protocol implementations
echo -e "\n=== Context usage in other protocol servers ==="
rg -n "protoOptions\.Context" --type=go
# 3. Look for actual error suppression or clean shutdown logic
echo -e "\n=== Error suppression or clean shutdown checks ==="
rg -n -C 3 "Stop\(\)|shutdown|clean" --type=go -g '**/protocol/**' | head -40
# 4. Verify if embedded context.Context in struct works with named field init
echo -e "\n=== Check Go pattern for embedded context.Context field ==="
rg -n "context\.Context" --type=go -g '**/protocol/**' -B 2 -A 2 | head -30Length of output: 3477
Fix embedded context.Context initialization syntax and implement error handling to achieve PR objectives.
The code has two critical issues:
-
Syntax Error (line 48):
CallbackContextembedscontext.Contextas an unnamed field, but you're initializing it with named field syntaxContext: protoOptions.Context,. For embedded interfaces in Go, you cannot use named field syntax. Either omit the field name or check Go's struct literal rules for embedded types. -
No Error Handling Implementation: The
Contextfield is added but never used. To achieve the PR's objective of not returning errors on clean connection close, the callback functions andhandleDone()must checks.callbackContext.Err()or use context cancellation patterns to distinguish between explicit shutdowns (context cancellation) and actual errors. Currently, no code checks the context for cancellation.
🤖 Prompt for AI Agents
In protocol/chainsync/server.go around lines 47 to 51, the struct literal
incorrectly uses a named field for an embedded context.Context and there is no
handling of context cancellation to treat clean shutdowns as non-errors; fix by
initializing the embedded context using the embedded-field form (i.e. include
protoOptions.Context as the value for the anonymous embedded field, not
Context:), and update the callback functions and handleDone() to check
s.callbackContext.Err() (or select on s.callbackContext.Done()) and treat
context.Canceled / context.DeadlineExceeded as a clean close (do not return them
as errors), ensuring any error returned by operations is only propagated when
the context indicates a real failure rather than an intentional shutdown.
| ) | ||
| } | ||
| // Create context for connection | ||
| c.ctx, c.cancelCtx = context.WithCancel(context.Background()) |
There was a problem hiding this comment.
We already have an async shutdown method, so we don't need to add another one.
Closes #1112
Summary by CodeRabbit
New Features
Improvements
Tests