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>
connection_test.go
Outdated
| // Stop the protocol explicitly | ||
| if err := chainSyncProtocol.Client.Stop(); err != nil { | ||
| t.Fatalf("failed to stop chain sync: %s", err) | ||
| } |
There was a problem hiding this comment.
This isn't really a good way to test this. The protocols aren't meant to be arbitrarily started/stopped from the outside, only from within the actual protocol. Instead, you should use the mock connection to send the Done message to shut it down.
connection.go
Outdated
| } | ||
|
|
||
| return false | ||
| } |
There was a problem hiding this comment.
I'm not sure that any of this is necessary. The docs for io.ReadFull() and binary.Read() both say that they'll only return io.EOF or io.ErrUnexpectedEOF. We may want to add a check for io.ErrUnexpectedEOF and remap it to io.EOF in the muxer (links below), and then we only need to check for errors.Is(err, io.EOF) anywhere below.
Lines 304 to 306 in a57c097
Lines 331 to 333 in a57c097
Signed-off-by: Jenita <jkawan@blinklabs.io>
Signed-off-by: Jenita <jkawan@blinklabs.io>
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
WalkthroughCentralizes connection error handling to treat EOF differently based on active protocol state, adds per-protocol state tracking and safe query paths, guards sending of protocol "done" messages when already finished, and adds/updates tests for connection and protocol behaviors. 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.1)
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).
| ouroboros_mock.ProtocolRoleServer, | ||
| []ouroboros_mock.ConversationEntry{ | ||
| // MsgProposeVersions from mock client |
There was a problem hiding this comment.
Handshake role mismatch: make the mock the client when oConn is server.
oConn is created with WithServer(true), so the mock must play the client. Using ProtocolRoleServer here risks a client–client stalemate and CI timeouts.
Apply in both subtests:
- mockConn := ouroboros_mock.NewConnection(
- ouroboros_mock.ProtocolRoleServer,
+ mockConn := ouroboros_mock.NewConnection(
+ ouroboros_mock.ProtocolRoleClient,Also applies to: 141-143
🤖 Prompt for AI Agents
In connection_test.go around lines 39 to 41 (and similarly at lines 141 to 143),
the mock conversation uses ouroboros_mock.ProtocolRoleServer while oConn is
constructed with WithServer(true), causing both sides to act as servers; change
the mock role to ouroboros_mock.ProtocolRoleClient so the mock plays the client
and the handshake can complete. Update both subtests to use ProtocolRoleClient
for the mock conversation entries and ensure the conversation entries reflect
client-initiated messages where appropriate.
| 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.
| // Wait for handshake to complete | ||
| time.Sleep(100 * time.Millisecond) | ||
|
|
||
| // Close connection to generate error | ||
| mockConn.Close() | ||
|
|
||
| // Should receive error since protocols are active | ||
| select { | ||
| case err := <-oConn.ErrorChan(): | ||
| if err == nil { | ||
| t.Fatal("expected connection error, got nil") | ||
| } | ||
| // We can't call t.Fatalf() from a different Goroutine, so we panic instead | ||
| panic(fmt.Sprintf("unexpected Ouroboros connection error: %s", err)) | ||
| }() | ||
| // Close connection | ||
| if err := oConn.Close(); err != nil { | ||
| t.Fatalf("unexpected error when closing Connection object: %s", err) | ||
| 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") | ||
| } |
There was a problem hiding this comment.
Test does not start any protocol but expects an error — contradicts PR semantics.
No mini‑protocol traffic is initiated here, yet the test asserts an error. Per PR, closing when no protocols exchanged messages should not return an error.
Choose one:
- Start at least one protocol (e.g., ChainSync FindIntersect) before closing to assert error propagation, or
- Keep it as handshake‑only and assert no error is emitted.
Example to assert no error:
-// 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")
-}
+select {
+case err := <-oConn.ErrorChan():
+ t.Fatalf("unexpected error with no active protocols: %v", err)
+case <-time.After(750 * time.Millisecond):
+ // OK: no error when no protocols ran
+}📝 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.
| // Wait for handshake to complete | |
| time.Sleep(100 * time.Millisecond) | |
| // Close connection to generate error | |
| mockConn.Close() | |
| // Should receive error since protocols are active | |
| select { | |
| case err := <-oConn.ErrorChan(): | |
| if err == nil { | |
| t.Fatal("expected connection error, got nil") | |
| } | |
| // We can't call t.Fatalf() from a different Goroutine, so we panic instead | |
| panic(fmt.Sprintf("unexpected Ouroboros connection error: %s", err)) | |
| }() | |
| // Close connection | |
| if err := oConn.Close(); err != nil { | |
| t.Fatalf("unexpected error when closing Connection object: %s", err) | |
| 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") | |
| } | |
| // Wait for handshake to complete | |
| time.Sleep(100 * time.Millisecond) | |
| // Close connection to generate error | |
| mockConn.Close() | |
| select { | |
| case err := <-oConn.ErrorChan(): | |
| t.Fatalf("unexpected error with no active protocols: %v", err) | |
| case <-time.After(750 * time.Millisecond): | |
| // OK: no error when no protocols ran | |
| } |
🤖 Prompt for AI Agents
In connection_test.go around lines 242 to 257, the test closes the mock
connection but never starts any mini-protocol yet asserts that an error should
be received, which contradicts the PR semantics; fix by choosing one of two
options: (A) start at least one mini-protocol exchange (for example initiate a
ChainSync FindIntersect request and ensure the protocol becomes active) before
closing the connection so that closing triggers an error and the existing select
remains valid, or (B) keep only the handshake and change the assertion to expect
no error — replace the select to wait briefly and fail if an error is received
(or succeed when timeout elapses with no error) to reflect that closing with no
protocol traffic should not emit an error.
3 participants
Closes #1112
Summary by CodeRabbit
New Features
Improvements
Tests