fix: prevent panic on close of closed channel in worker client reconnection (#7837)

* fix: prevent panic on close of closed channel in worker client reconnection

- Use idiomatic Go pattern of setting channels to nil after closing instead of flags
- Extract repeated safe-close logic into safeCloseChannel() helper method
- Call safeCloseChannel() in attemptConnection(), reconnect(), and handleDisconnect()
- In safeCloseChannel(), check if channel is not nil, close it, and set to nil
- Also set streamExit to nil in attemptConnection() when registration fails
- This follows Go best practices for channel management and prevents double-close panics
- Improved code maintainability by eliminating duplication

* fix: prevent panic on close of closed channel in worker client reconnection

- Use idiomatic Go pattern of setting channels to nil after closing instead of flags
- Extract repeated safe-close logic into safeCloseChannel() helper method
- Call safeCloseChannel() in attemptConnection(), reconnect(), and handleDisconnect()
- In safeCloseChannel(), check if channel is not nil, close it, and set to nil
- Also set streamExit to nil in attemptConnection() when registration fails
- Document thread-safety assumptions: function is safe in current usage (serialized
  in managerLoop) but would need synchronization if used in concurrent contexts
- This follows Go best practices for channel management and prevents double-close panics
- Improved code maintainability by eliminating duplication
This commit is contained in:
Chris Lu
2025-12-21 19:29:08 -08:00
committed by GitHub
parent 1dfda78e59
commit 683eef72a6

View File

@@ -98,6 +98,17 @@ func NewGrpcAdminClient(adminAddress string, workerID string, dialOption grpc.Di
return c
}
// safeCloseChannel safely closes a channel and sets it to nil to prevent double-close panics.
// NOTE: This function is NOT thread-safe. It is safe to use in this codebase because all calls
// are serialized within the managerLoop goroutine. If this function is used in concurrent contexts
// in the future, synchronization (e.g., sync.Mutex) should be added.
func (c *GrpcAdminClient) safeCloseChannel(chPtr *chan struct{}) {
if *chPtr != nil {
close(*chPtr)
*chPtr = nil
}
}
func (c *GrpcAdminClient) managerLoop() {
state := &grpcState{shouldReconnect: true}
@@ -221,7 +232,7 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error {
if s.lastWorkerInfo != nil {
// Send registration via the normal outgoing channel and wait for response via incoming
if err := c.sendRegistration(s.lastWorkerInfo); err != nil {
close(s.streamExit)
c.safeCloseChannel(&s.streamExit)
s.streamCancel()
s.conn.Close()
s.connected = false
@@ -240,9 +251,7 @@ func (c *GrpcAdminClient) attemptConnection(s *grpcState) error {
// reconnect attempts to re-establish the connection
func (c *GrpcAdminClient) reconnect(s *grpcState) error {
// Clean up existing connection completely
if s.streamExit != nil {
close(s.streamExit)
}
c.safeCloseChannel(&s.streamExit)
if s.streamCancel != nil {
s.streamCancel()
}
@@ -425,7 +434,7 @@ func (c *GrpcAdminClient) handleDisconnect(cmd grpcCommand, s *grpcState) {
}
// Send shutdown signal to stop reconnection loop
close(s.reconnectStop)
c.safeCloseChannel(&s.reconnectStop)
s.connected = false
s.shouldReconnect = false
@@ -450,7 +459,7 @@ func (c *GrpcAdminClient) handleDisconnect(cmd grpcCommand, s *grpcState) {
}
// Send shutdown signal to stop handlers loop
close(s.streamExit)
c.safeCloseChannel(&s.streamExit)
// Cancel stream context
if s.streamCancel != nil {