fix(realtime): repair hub shutdown deadlock and WaitGroup misuse

internal/realtime has timed out at 3m under -race on seven CI runs since
2026-08-21 (#212), on PRs that touch no Go code. Three defects, all in
the Hub start/stop lifecycle:

1. The writer goroutine's cleanup sent on h.unregister unconditionally
   after a racy h.stopped pre-check. Stop() can flip that flag and the
   run loop can return between the load and the send, leaving the writer
   blocked forever on a channel nobody drains — so writerWg.Wait() never
   returns and shutdown wedges. This is the 3m hang. Select on stopCh and
   close the client directly instead; the existing CAS guard keeps that
   safe against the run loop's own stop-path close. HandleWebSocket's
   h.register send had the same shape and gets the same treatment.

2. Run() called h.wg.Add(1) from inside the goroutine while Stop() called
   h.wg.Wait() — an Add from zero concurrent with Wait. Stop() could
   return before the loop had started, leaving it running afterwards.
   The count is now taken in NewHub and consumed by whichever of Run or
   Stop claims runOwner first, so Add always happens-before Wait and a
   hub that is never Run still stops cleanly.

3. HandleWebSocket called h.writerWg.Add(1) with no ordering against
   Stop()'s writerWg.Wait(), same misuse. Writer slots are now reserved
   through admitWriter() under lifecycleMu, which refuses once Stop has
   begun, so every Add is ordered before the Wait.

The dead h.stopped flag is removed with its last reader.

TestStopDoesNotDeadlockAgainstChurn drives connect/disconnect churn
against a concurrent Stop() and fails on the unpatched hub (verified:
DATA RACE) while passing on the fix. Also run: 20x internal/realtime
under -race with GOMAXPROCS=2 (340 tests, green) and the full suite
(1527 tests, 30 packages, green).

Author: aksOps

internal/realtime/hub.go

@@ -91,11 +91,21 @@ type Hub struct {
 	// revision-driven publication the event hub performs (#164).
 	aggregateMode atomic.Bool
 
-	stopCh   chan struct{}
-	stopped  atomic.Bool
-	wg       sync.WaitGroup
-	writerWg sync.WaitGroup // tracks writer goroutines
-	devMode  bool
+	stopCh chan struct{}
+	// runOwner is claimed exactly once, by whichever of Run or Stop gets there
+	// first. It balances the wg count taken in NewHub: Run consumes it when it
+	// starts the loop, Stop consumes it when the hub is stopped without ever
+	// having been run. Taking the count at construction is what makes
+	// wg.Add happen-before wg.Wait, which `go hub.Run()` cannot guarantee.
+	runOwner atomic.Bool
+	// lifecycleMu serialises writerWg.Add against Stop's writerWg.Wait.
+	// Without it the handler can Add from zero while Stop is already
+	// waiting, which is WaitGroup misuse and a real data race.
+	lifecycleMu sync.Mutex
+	closing     bool
+	wg          sync.WaitGroup
+	writerWg    sync.WaitGroup // tracks writer goroutines
+	devMode     bool
 
 	// enforceOrigin and originHosts implement WS_ALLOWED_ORIGINS. Enforcement
 	// is on whenever authentication is enabled or APP_ENV=production; the
@@ -150,12 +160,19 @@ func NewHub(onConnectionChange func(count int)) *Hub {
 	h.logBuffer = h.logPool.Get().([]LogEntry)
 	h.metricBuffer = h.metricPool.Get().([]MetricEntry)
 
+	// Balanced by Run (loop exit) or by Stop (hub never run). See runOwner.
+	h.wg.Add(1)
+
 	return h
 }
 
 // Run starts the hub's main event loop. Should be called in a goroutine.
 func (h *Hub) Run() {
-	h.wg.Add(1)
+	if !h.runOwner.CompareAndSwap(false, true) {
+		// Stop already released the construction-time count, or Run was
+		// called twice. Either way there is no loop to start.
+		return
+	}
 	defer h.wg.Done()
 
 	flushTicker := time.NewTicker(h.flushInterval)
@@ -388,10 +405,31 @@ func (h *Hub) BroadcastMetric(entry MetricEntry) {
 	}
 }
 
+// admitWriter reserves a writer slot on writerWg, refusing once Stop has
+// begun. Every writerWg.Add goes through here so each one is ordered before
+// Stop's Wait by lifecycleMu.
+func (h *Hub) admitWriter() bool {
+	h.lifecycleMu.Lock()
+	defer h.lifecycleMu.Unlock()
+	if h.closing {
+		return false
+	}
+	h.writerWg.Add(1)
+	return true
+}
+
 // Stop gracefully shuts down the hub.
 func (h *Hub) Stop() {
-	h.stopped.Store(true)
+	h.lifecycleMu.Lock()
+	h.closing = true
+	h.lifecycleMu.Unlock()
+
 	close(h.stopCh)
+	if h.runOwner.CompareAndSwap(false, true) {
+		// Run was never called, so nothing will ever call Done for the
+		// construction-time count. Release it here or Wait blocks forever.
+		h.wg.Done()
+	}
 	h.wg.Wait()
 	h.writerWg.Wait()
 	slog.Info("🛑 WebSocket hub stopped")
@@ -448,22 +486,47 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 		tenant: connTenantScope(r),
 	}
 
-	h.register <- c
+	// Reserve the writer slot before the registration handshake so the
+	// count is taken while the hub is provably still accepting.
+	if !h.admitWriter() {
+		releaseSlot()
+		_ = conn.Close(websocket.StatusGoingAway, "server shutting down")
+		return
+	}
+
+	// Registration races Stop(): once the run loop returns on stopCh nothing
+	// drains h.register, so an unconditional send blocks this goroutine
+	// forever and Stop()'s wg.Wait() never returns. Refuse the connection
+	// instead of joining a hub that is going away.
+	select {
+	case h.register <- c:
+	case <-h.stopCh:
+		h.writerWg.Done()
+		releaseSlot()
+		_ = conn.Close(websocket.StatusGoingAway, "server shutting down")
+		return
+	}
 
 	// Writer goroutine
-	h.writerWg.Add(1)
 	go func() { // #nosec G118 -- long-lived WS writer goroutine outlives HTTP request intentionally
 		defer h.writerWg.Done()
 		// Release the admission slot when the writer exits — the writer
 		// outlives the HandleWebSocket reader loop, so this is the last
 		// goroutine alive for this client.
 		defer releaseSlot()
 		defer func() {
-			if !h.stopped.Load() {
-				h.unregister <- c
-			} else if c.closed.CompareAndSwap(false, true) {
-				// Hub already stopped; clean up directly.
-				close(c.send)
+			// An unconditional send here is what wedged Stop(): the run loop
+			// returns on stopCh and then nothing drains h.unregister, so this
+			// goroutine blocks forever and writerWg.Wait() never returns.
+			// Select on stopCh so the send is abandoned the moment the drain
+			// side is gone; the CAS guard makes the direct close safe against
+			// the run loop's own stop-path close.
+			select {
+			case h.unregister <- c:
+			case <-h.stopCh:
+				if c.closed.CompareAndSwap(false, true) {
+					close(c.send)
+				}
 			}
 			_ = conn.Close(websocket.StatusNormalClosure, "closing")
 		}()

internal/realtime/hub_shutdown_race_test.go

@@ -0,0 +1,70 @@
+//go:build !race_off
+
+package realtime
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+)
+
+// TestStopDoesNotDeadlockAgainstChurn is the regression for the hub shutdown
+// deadlock (#212). Registration and unregistration both send on channels that
+// only the run loop drains; the run loop returns as soon as stopCh closes. A
+// client whose writer goroutine reaches its unregister send after that return
+// blocks forever, so Stop()'s writerWg.Wait() never comes back and the whole
+// graceful shutdown wedges.
+//
+// The test drives connect/disconnect churn while Stop() runs, so the send and
+// the run loop's exit interleave, and it fails by timeout rather than hanging
+// the suite for the full go test deadline.
+func TestStopDoesNotDeadlockAgainstChurn(t *testing.T) {
+	for attempt := 0; attempt < 12; attempt++ {
+		hub := NewHub(nil)
+		go hub.Run()
+
+		srv := httptest.NewServer(http.HandlerFunc(hub.HandleWebSocket))
+		wsURL := "ws" + srv.URL[len("http"):]
+
+		var wg sync.WaitGroup
+		for i := 0; i < 8; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer cancel()
+				c, _, err := websocket.Dial(ctx, wsURL, nil)
+				if err != nil {
+					return // refused mid-shutdown is a valid outcome
+				}
+				// Close immediately: the writer goroutine's unregister send
+				// then races the run loop's exit.
+				_ = c.Close(websocket.StatusNormalClosure, "bye")
+			}()
+		}
+
+		// Stop concurrently with the churn — the window under test.
+		time.Sleep(time.Duration(attempt) * time.Millisecond)
+		done := make(chan struct{})
+		go func() {
+			hub.Stop()
+			close(done)
+		}()
+
+		select {
+		case <-done:
+		case <-time.After(20 * time.Second):
+			srv.Close()
+			t.Fatalf("attempt %d: Hub.Stop() deadlocked — a writer or handler is blocked "+
+				"sending on register/unregister with the run loop already returned", attempt)
+		}
+
+		wg.Wait()
+		srv.Close()
+	}
+}