go-sse-endpoint-hang

Answer

Root cause (SHIM-EVENT-001): the /event handler set SSE headers but never called w.WriteHeader(200) + flusher.Flush() before entering its wait loop, and never replayed stored memory_events on subscribe. A late subscriber (no events in flight) never received the 200 response, so its first read blocked forever; even connected subscribers missed all pre-connect history.

Fix — go-sse-endpoint-hang/sse.go:

1. Immediate header flush on connect, in BOTH branches — shared helper called before any wait loop:

// writeSSEHeaders flushes the 200 status and SSE headers immediately on
// connect. This is the core fix: without it the client's first read blocks
// until the first event, which never arrives for late subscribers.
func writeSSEHeaders(w http.ResponseWriter) {
    h := w.Header()
    h.Set("Content-Type", "text/event-stream")
    h.Set("Cache-Control", "no-cache")
    h.Set("Connection", "keep-alive")
    h.Set("X-Accel-Buffering", "no")
    w.WriteHeader(http.StatusOK)
    if f, ok := w.(http.Flusher); ok {
        f.Flush()
    }
}

1. Historical replay on subscribe (id ASC, capped ~50) — atomic snapshot + bus registration under one lock so nothing is missed between replay and subscribe:

func (b *Broker) subscribeWithReplay(sessionID string) ([]memoryEvent, chan memoryEvent, func()) {
    b.mu.Lock()
    defer b.mu.Unlock()
    replay := append([]memoryEvent{}, b.history[sessionID]...)
    sort.SliceStable(replay, func(i, j int) bool { return replay[i].ID < replay[j].ID })
    ch := make(chan memoryEvent, subChanBuffer)
    b.subs[sessionID][ch] = struct{}{}
    return replay, ch, func() { /* delete + close under b.mu */ }
}

Event-bus branch (headers flushed → replay frames → live wait loop):

func (s *Server) ServeEventBus(w http.ResponseWriter, r *http.Request) {
    ...
    writeSSEHeaders(w)                      // FIX: flush before the wait loop
    replay, ch, unsub := s.broker.subscribeWithReplay(sid)
    defer unsub()
    for _, ev := range replay { writeSSEFrame(w, ev) }   // FIX: replay history
    for {
        select {
        case ev, ok := <-ch:
            if !ok { return }
            writeSSEFrame(w, ev)
        case <-r.Context().Done():
            return
        }
    }
}

Poll-fallback branch (same contract — flush first, replay PollSince(sid, 0), then poll PollSince(sid, lastID) on a 250ms ticker).

1. Bounded test helper (sse_test.go) — replaces the broken io.ReadAll (which blocks forever on an infinite SSE stream) with event:/data: block parsing wrapped in a quietReader that resets a 5s idle timer per read:

func (q *quietReader) Read(p []byte) (int, error) {
    ch := make(chan result, 1)
    go func() { n, err := q.r.Read(p); ch <- result{n, err} }()
    q.timer.Reset(q.idle)
    select {
    case res := <-ch:   return res.n, res.err
    case <-q.timer.C:   return 0, errQuietTimeout
    }
}

The original buggy handlers are preserved behind a //go:build buggy tag for A/B verification (handler_buggy.go, sse_buggy.go).