fix: Hub lifecycle — wg race, TOCTOU, Go 1.27

.github/workflows/benchmark.yml

@@ -19,7 +19,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.25'
+        go-version: '1.27'
         cache: true
 
     - name: Run benchmarks

.github/workflows/test.yml

@@ -34,7 +34,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
-        go-version: ['1.25']  # Match go.mod requirement
+        go-version: ['1.27']  # Match go.mod requirement
     env:
       GOEXPERIMENT: jsonv2  # Required for encoding/json/v2
 
@@ -63,7 +63,7 @@ jobs:
       run: go test -short -v -race -coverprofile=coverage.txt -covermode=atomic ./...
 
     - name: Upload coverage to Codecov
-      if: matrix.os == 'ubuntu-latest' && matrix.go-version == '1.25'
+      if: matrix.os == 'ubuntu-latest' && matrix.go-version == '1.27'
       uses: codecov/codecov-action@v5
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
@@ -87,7 +87,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.25'
+        go-version: '1.27'
         cache: true
 
     - name: Run golangci-lint
@@ -110,7 +110,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.25'
+        go-version: '1.27'
         cache: true
 
     - name: Check formatting

README.md

@@ -157,7 +157,7 @@ go get github.com/coregx/stream
 - ✅ **Text & Binary** - Both message types supported
 - ✅ **Control Frames** - Ping/Pong, Close handshake
 - ✅ **Broadcasting Hub** - Efficient multi-client messaging
-- ✅ **Connection Management** - Auto cleanup, timeouts
+- ✅ **Connection Management** - Auto cleanup
 - ✅ **Frame Masking** - Client-to-server masking (RFC requirement)
 - ✅ **84.3% Test Coverage** - 99 tests, production-ready
 

go.mod

@@ -1,3 +1,3 @@
 module github.com/coregx/stream
 
-go 1.25
+go 1.27

sse/hub.go

@@ -186,16 +186,17 @@ func (h *Hub[T]) removeClient(client *Conn) {
 //	}
 //	err = hub.Register(conn)
 func (h *Hub[T]) Register(conn *Conn) error {
-	h.mu.RLock()
-	closed := h.closed
-	h.mu.RUnlock()
-
-	if closed {
+	select {
+	case <-h.done:
+		return ErrHubClosed
+	default:
+	}
+	select {
+	case h.register <- conn:
+		return nil
+	case <-h.done:
 		return ErrHubClosed
 	}
-
-	h.register <- conn
-	return nil
 }
 
 // Unregister removes a connection from the hub.
@@ -209,16 +210,17 @@ func (h *Hub[T]) Register(conn *Conn) error {
 //
 //	err := hub.Unregister(conn)
 func (h *Hub[T]) Unregister(conn *Conn) error {
-	h.mu.RLock()
-	closed := h.closed
-	h.mu.RUnlock()
-
-	if closed {
+	select {
+	case <-h.done:
+		return ErrHubClosed
+	default:
+	}
+	select {
+	case h.unregister <- conn:
+		return nil
+	case <-h.done:
 		return ErrHubClosed
 	}
-
-	h.unregister <- conn
-	return nil
 }
 
 // Broadcast sends data to all connected clients.
@@ -236,16 +238,17 @@ func (h *Hub[T]) Unregister(conn *Conn) error {
 //
 //	err := hub.Broadcast("Server restarting in 5 minutes")
 func (h *Hub[T]) Broadcast(data T) error {
-	h.mu.RLock()
-	closed := h.closed
-	h.mu.RUnlock()
-
-	if closed {
+	select {
+	case <-h.done:
+		return ErrHubClosed
+	default:
+	}
+	select {
+	case h.broadcast <- data:
+		return nil
+	case <-h.done:
 		return ErrHubClosed
 	}
-
-	h.broadcast <- data
-	return nil
 }
 
 // BroadcastJSON sends a JSON-encoded value to all connected clients.

websocket/hub.go

@@ -44,9 +44,10 @@
 	broadcast  chan []byte // Broadcast message to all
 
 	// Lifecycle management
-	done   chan struct{}  // Shutdown signal
-	closed bool           // Track if hub is closed
-	wg     sync.WaitGroup // Wait for goroutines
+	done    chan struct{}  // Shutdown signal
+	closed  bool           // Track if hub is closed
+	started bool           // Track if Run() was called
+	wg      sync.WaitGroup // Wait for goroutines
 
 	// Thread-safety for clients map and closed flag
 	mu sync.RWMutex
@@ -62,13 +63,15 @@
 //
 // Returns a ready-to-use Hub with initialized channels.
 func NewHub() *Hub {
-	return &Hub{
+	h := &Hub{
 		clients:    make(map[*Conn]bool),
 		register:   make(chan *Conn),
 		unregister: make(chan *Conn),
 		broadcast:  make(chan []byte, 256), // Buffered for performance
 		done:       make(chan struct{}),
 	}
+	h.wg.Add(1) // Must be before go hub.Run() to prevent race with Close().
+	return h
 }
 
 // Run starts the Hub's event loop.
@@ -83,43 +86,55 @@
 //   - Graceful shutdown
 //
 // Run exits when Close() is called.
+// Run must be called in a goroutine: go hub.Run().
+// Call hub.AddRunning() before starting the goroutine if using wg externally.
 func (h *Hub) Run() {
-	h.wg.Add(1)
+	h.mu.Lock()
+	if h.started || h.closed {
+		h.mu.Unlock()
+		return
+	}
+	h.started = true
+	h.mu.Unlock()
+
 	defer h.wg.Done()
 
 	for {
 		select {
-		case client := <-h.register:
-			// Register new client
+		case client, ok := <-h.register:
+			if !ok {
+				return
+			}
 			h.mu.Lock()
 			h.clients[client] = true
 			h.mu.Unlock()
 
-		case client := <-h.unregister:
-			// Unregister client
+		case client, ok := <-h.unregister:
+			if !ok {
+				return
+			}
 			h.mu.Lock()
-			if _, ok := h.clients[client]; ok {
+			if _, exists := h.clients[client]; exists {
 				delete(h.clients, client)
-				_ = client.Close() // Close connection
+				_ = client.Close()
 			}
 			h.mu.Unlock()
 
-		case message := <-h.broadcast:
-			// Broadcast to all clients
+		case message, ok := <-h.broadcast:
+			if !ok {
+				return
+			}
 			h.mu.RLock()
 			for client := range h.clients {
-				// Send in goroutine to avoid blocking on slow clients
 				go func(c *Conn, msg []byte) {
 					if err := c.Write(BinaryMessage, msg); err != nil {
-						// Auto-unregister on write failure
 						h.Unregister(c)
 					}
 				}(client, message)
 			}
 			h.mu.RUnlock()
 
 		case <-h.done:
-			// Shutdown
 			return
 		}
 	}
@@ -136,14 +151,10 @@
 //
 // Thread-safe: can be called from multiple goroutines.
 func (h *Hub) Register(client *Conn) {
-	h.mu.RLock()
-	if h.closed {
-		h.mu.RUnlock()
-		return
+	select {
+	case h.register <- client:
+	case <-h.done:
 	}
-	h.mu.RUnlock()
-
-	h.register <- client
 }
 
 // Unregister removes a client from the Hub.
@@ -157,14 +168,10 @@
 // Thread-safe: can be called from multiple goroutines.
 // Safe to call multiple times for the same client (no-op after first call).
 func (h *Hub) Unregister(client *Conn) {
-	h.mu.RLock()
-	if h.closed {
-		h.mu.RUnlock()
-		return
+	select {
+	case h.unregister <- client:
+	case <-h.done:
 	}
-	h.mu.RUnlock()
-
-	h.unregister <- client
 }
 
 // Broadcast sends a message to all connected clients.
@@ -181,14 +188,10 @@
 // Thread-safe: can be called from multiple goroutines.
 // Non-blocking: queues message and returns immediately.
 func (h *Hub) Broadcast(message []byte) {
-	h.mu.RLock()
-	if h.closed {
-		h.mu.RUnlock()
-		return
+	select {
+	case h.broadcast <- message:
+	case <-h.done:
 	}
-	h.mu.RUnlock()
-
-	h.broadcast <- message
 }
 
 // BroadcastText sends a text message to all connected clients.
@@ -259,13 +262,16 @@
 		return nil
 	}
 	h.closed = true
+	started := h.started
 	h.mu.Unlock()
 
 	// Signal shutdown to event loop
 	close(h.done)
 
-	// Wait for event loop to exit
-	h.wg.Wait()
+	// Wait for event loop to exit — only if Run() was called.
+	if started {
+		h.wg.Wait()
+	}
 
 	// Close all client connections
 	h.mu.Lock()
@@ -275,10 +281,8 @@
 	h.clients = make(map[*Conn]bool) // Clear map
 	h.mu.Unlock()
 
-	// Close channels (safe now that event loop exited and no new sends)
-	close(h.register)
-	close(h.unregister)
-	close(h.broadcast)
+	// Channels are NOT closed — done signal + select handles shutdown.
+	// Closing channels would cause panic in concurrent senders.
 
 	return nil
 }