-
Notifications
You must be signed in to change notification settings - Fork 19
Not returning on connection close for chainsync, block-fetch and tx-submission protocol #1141
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
6b39f59
c42b242
829e00e
b255d66
3afdaa0
4de8be1
a43fdd1
762277d
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -29,7 +29,10 @@ | |
"io" | ||
"log/slog" | ||
"net" | ||
"os" | ||
"strings" | ||
"sync" | ||
"syscall" | ||
"time" | ||
|
||
"github.com/blinklabs-io/gouroboros/connection" | ||
|
@@ -250,6 +253,100 @@ | |
close(c.errorChan) | ||
} | ||
|
||
// isConnectionReset checks if an error is a connection reset error using proper error type checking | ||
func (c *Connection) isConnectionReset(err error) bool { | ||
if errors.Is(err, io.EOF) { | ||
return true | ||
} | ||
|
||
// Check for connection reset errors using proper error type checking | ||
var opErr *net.OpError | ||
if errors.As(err, &opErr) { | ||
if syscallErr, ok := opErr.Err.(*os.SyscallError); ok { | ||
if errno, ok := syscallErr.Err.(syscall.Errno); ok { | ||
// Check for connection reset (ECONNRESET) or broken pipe (EPIPE) | ||
return errno == syscall.ECONNRESET || errno == syscall.EPIPE | ||
} | ||
} | ||
// Also check for string-based errors as fallback for edge cases | ||
errStr := opErr.Err.Error() | ||
return strings.Contains(errStr, "connection reset") || | ||
strings.Contains(errStr, "broken pipe") | ||
} | ||
|
||
return false | ||
} | ||
|
||
// checkProtocolsDone checks if the protocols are explicitly stopped by the client - treat as normal connection closure | ||
func (c *Connection) checkProtocolsDone() bool { | ||
// Check chain-sync protocol | ||
if c.chainSync != nil { | ||
if (c.chainSync.Client != nil && !c.chainSync.Client.IsDone()) || | ||
(c.chainSync.Server != nil && !c.chainSync.Server.IsDone()) { | ||
return false | ||
} | ||
} | ||
|
||
// Check block-fetch protocol | ||
if c.blockFetch != nil { | ||
if (c.blockFetch.Client != nil && !c.blockFetch.Client.IsDone()) || | ||
(c.blockFetch.Server != nil && !c.blockFetch.Server.IsDone()) { | ||
return false | ||
} | ||
} | ||
|
||
// Check tx-submission protocol | ||
if c.txSubmission != nil { | ||
if (c.txSubmission.Client != nil && !c.txSubmission.Client.IsDone()) || | ||
(c.txSubmission.Server != nil && !c.txSubmission.Server.IsDone()) { | ||
return false | ||
} | ||
} | ||
|
||
// Check local-state-query protocol | ||
if c.localStateQuery != nil { | ||
if (c.localStateQuery.Client != nil && !c.localStateQuery.Client.IsDone()) || | ||
(c.localStateQuery.Server != nil && !c.localStateQuery.Server.IsDone()) { | ||
return false | ||
} | ||
} | ||
|
||
// Check local-tx-monitor protocol | ||
if c.localTxMonitor != nil { | ||
if (c.localTxMonitor.Client != nil && !c.localTxMonitor.Client.IsDone()) || | ||
(c.localTxMonitor.Server != nil && !c.localTxMonitor.Server.IsDone()) { | ||
return false | ||
} | ||
} | ||
|
||
// Check local-tx-submission protocol | ||
if c.localTxSubmission != nil { | ||
if (c.localTxSubmission.Client != nil && !c.localTxSubmission.Client.IsDone()) || | ||
(c.localTxSubmission.Server != nil && !c.localTxSubmission.Server.IsDone()) { | ||
return false | ||
} | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Let's also add |
||
|
||
return true | ||
} | ||
|
||
// handleConnectionError handles connection-level errors centrally | ||
func (c *Connection) handleConnectionError(err error) error { | ||
if err == nil { | ||
return nil | ||
} | ||
|
||
if c.checkProtocolsDone() { | ||
return nil | ||
} | ||
|
||
if errors.Is(err, io.EOF) || c.isConnectionReset(err) { | ||
return err | ||
} | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This block doesn't seem to be useful, since we unconditionally |
||
|
||
return err | ||
} | ||
|
||
// setupConnection establishes the muxer, configures and starts the handshake process, and initializes | ||
// the appropriate mini-protocols | ||
func (c *Connection) setupConnection() error { | ||
|
@@ -285,16 +382,20 @@ | |
if !ok { | ||
return | ||
} | ||
var connErr *muxer.ConnectionClosedError | ||
if errors.As(err, &connErr) { | ||
// Pass through ConnectionClosedError from muxer | ||
c.errorChan <- err | ||
} else { | ||
// Wrap error message to denote it comes from the muxer | ||
c.errorChan <- fmt.Errorf("muxer error: %w", err) | ||
|
||
// Use centralized connection error handling | ||
if handledErr := c.handleConnectionError(err); handledErr != nil { | ||
var connErr *muxer.ConnectionClosedError | ||
if errors.As(handledErr, &connErr) { | ||
// Pass through ConnectionClosedError from muxer | ||
c.errorChan <- handledErr | ||
} else { | ||
// Wrap error message to denote it comes from the muxer | ||
c.errorChan <- fmt.Errorf("muxer error: %w", handledErr) | ||
} | ||
// Close connection on muxer errors | ||
c.Close() | ||
} | ||
// Close connection on muxer errors | ||
c.Close() | ||
} | ||
}() | ||
protoOptions := protocol.ProtocolOptions{ | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We should check for error types rather than substring matches of the
Error()
output. Do we actually get any "connection closed" error that isn'tio.EOF
?