From 8350cdcca57a0c7776efa2411d5af177f3c3f42d Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 10 Sep 2026 22:16:26 +0300 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20Hub=20lifecycle=20=E2=80=94=20wg.Add?= =?UTF-8?q?=20before=20Run,=20select-based=20TOCTOU=20fix,=20Go=201.27?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B1: go.mod updated to Go 1.27 (json/v2 requires it). B2: wg.Add(1) moved to NewHub() — prevents Close() racing with Run() startup. B3: Register/Unregister/Broadcast use select with done channel instead of TOCTOU (RLock→check closed→RUnlock→send). No more panic on closed channel. Close() no longer closes register/unregister/broadcast channels. --- go.mod | 2 +- websocket/hub.go | 68 +++++++++++++++++++++--------------------------- 2 files changed, 31 insertions(+), 39 deletions(-) diff --git a/go.mod b/go.mod index d015adb..8449c39 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,3 @@ module github.com/coregx/stream -go 1.25 +go 1.27 diff --git a/websocket/hub.go b/websocket/hub.go index 3b2d2ac..801b65f 100644 --- a/websocket/hub.go +++ b/websocket/hub.go @@ -62,13 +62,15 @@ type Hub struct { // // 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,35 +85,40 @@ func NewHub() *Hub { // - 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) 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) @@ -119,7 +126,6 @@ func (h *Hub) Run() { h.mu.RUnlock() case <-h.done: - // Shutdown return } } @@ -136,14 +142,10 @@ func (h *Hub) Run() { // // 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 +159,10 @@ func (h *Hub) Register(client *Conn) { // 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 +179,10 @@ func (h *Hub) Unregister(client *Conn) { // 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. @@ -275,10 +269,8 @@ func (h *Hub) Close() error { 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 } From f854a50a39ce194f64bec1bff93c60b246de9421 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 10 Sep 2026 22:28:19 +0300 Subject: [PATCH 2/3] fix: Close-without-Run deadlock, SSE TOCTOU, CI 1.27, README cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - WebSocket Hub: added 'started' flag — Close() without Run() no longer deadlocks, double Run() returns immediately - SSE Hub: Register/Unregister/Broadcast use select with done channel (double-select pattern: check done first, then try send) - CI: Go 1.25 → 1.27 in all workflows - README: removed 'timeouts' claim (no deadlines implemented yet) --- .github/workflows/benchmark.yml | 2 +- .github/workflows/test.yml | 8 +++--- README.md | 2 +- sse/hub.go | 51 +++++++++++++++++---------------- websocket/hub.go | 23 +++++++++++---- 5 files changed, 51 insertions(+), 35 deletions(-) diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 8de34cf..9a583e4 100644 --- a/.github/workflows/benchmark.yml +++ b/.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 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index dd55582..9644550 100644 --- a/.github/workflows/test.yml +++ b/.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 diff --git a/README.md b/README.md index 39e580b..cf40e7f 100644 --- a/README.md +++ b/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 diff --git a/sse/hub.go b/sse/hub.go index 81c00b1..15b0eb8 100644 --- a/sse/hub.go +++ b/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. diff --git a/websocket/hub.go b/websocket/hub.go index 801b65f..fcea6ae 100644 --- a/websocket/hub.go +++ b/websocket/hub.go @@ -44,9 +44,10 @@ type Hub struct { 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 @@ -88,6 +89,15 @@ func NewHub() *Hub { // 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.mu.Lock() + if h.started || h.closed { + h.mu.Unlock() + h.wg.Done() + return + } + h.started = true + h.mu.Unlock() + defer h.wg.Done() for { @@ -253,13 +263,16 @@ func (h *Hub) Close() error { 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() From 2120148dcc36fea26b63002a2cd828fc30ed09f3 Mon Sep 17 00:00:00 2001 From: Andrey Kolkov Date: Thu, 10 Sep 2026 22:32:41 +0300 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20remove=20wg.Done=20from=20Run=20guar?= =?UTF-8?q?d=20=E2=80=94=20prevents=20negative=20WaitGroup=20panic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Double Run or Run-after-Close no longer panics. Guard branch returns silently without touching WaitGroup counter. Close without Run returns immediately via started flag. --- websocket/hub.go | 1 - 1 file changed, 1 deletion(-) diff --git a/websocket/hub.go b/websocket/hub.go index fcea6ae..2706efa 100644 --- a/websocket/hub.go +++ b/websocket/hub.go @@ -92,7 +92,6 @@ func (h *Hub) Run() { h.mu.Lock() if h.started || h.closed { h.mu.Unlock() - h.wg.Done() return } h.started = true