From 7e802bbb145ddf076a11d50608ca2ace11e907f5 Mon Sep 17 00:00:00 2001 From: dtoro Date: Tue, 7 Jul 2026 13:34:32 +0200 Subject: [PATCH] sse: real-time flushing via raw handler overriding the generated route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generated strict-server path could only return an io.Reader that io.Copy drains without flushing, so SSE events sat chunk-buffered instead of streaming in real time. Replace it with a raw http.ResponseWriter handler (serveSSE) that Flush()es after every event. Routing: chi allows a later registration to supersede an earlier one for the same method+path (verified empirically for v5.3.1), so serveSSE is registered on the router AFTER gen.HandlerWithOptions and wins over the generated /events/stream route. It inherits the base middleware chain and applies auth via With(). The generated StreamEvents method now returns an error (never reached) so a routing regression fails loudly rather than silently reverting to buffered delivery. Adds TestSSEStreamRealtimeDelivery: a real httptest.NewServer + streaming client (NewRecorder can't flush) that connects, triggers an event, and asserts delivery within 3s — proving both the override routing and per-event flushing. Passes in <1s. Co-Authored-By: Claude Fable 5 --- internal/httpapi/server.go | 10 ++-- internal/httpapi/sse.go | 88 ++++++++++++++++++----------------- internal/httpapi/sse_test.go | 89 ++++++++++++++++++++++++++++++++++++ 3 files changed, 142 insertions(+), 45 deletions(-) create mode 100644 internal/httpapi/sse_test.go diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 0aeda9e..d1f5afb 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -101,12 +101,16 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand }, }) + // SSE stream: override the generated /events/stream route with a raw + // flushing handler (registered AFTER HandlerWithOptions so chi's last + // registration wins). The strict-server path can't Flush() per event; + // this one uses the real ResponseWriter for real-time delivery. It + // inherits the router's base middleware and applies auth via With(). + r.With(combinedAuth(cfg)).Get("/api/v1/events/stream", s.serveSSE) + // Mount MCP at /mcp (plan R3-10) r.With(combinedAuth(cfg)).Handle("/mcp", mcphandler.NewHandler(pool, cfg.MCPBearerToken)) - // Mount SSE stream — handled by the strict handler's StreamEvents method - // via the OpenAPI-specified /api/v1/events/stream route. - return r } diff --git a/internal/httpapi/sse.go b/internal/httpapi/sse.go index 45452f0..4a5a398 100644 --- a/internal/httpapi/sse.go +++ b/internal/httpapi/sse.go @@ -5,7 +5,6 @@ import ( "context" "encoding/json" "fmt" - "io" "log/slog" "net/http" "strconv" @@ -231,35 +230,32 @@ type ioWriter interface { Write([]byte) (int, error) } -// StreamEvents implements the OpenAPI interface (SSE event stream). +// serveSSE is the streaming handler for GET /api/v1/events/stream, registered +// directly on the chi router in NewHandler so it supersedes the generated +// route. Using the raw http.ResponseWriter lets us Flush() after every event +// (real-time delivery); the generated strict-server path can only hand back +// an io.Reader that io.Copy drains without flushing (chunk-buffered). // -// The strict-server model hands us only an io.Reader Body (copied to the -// client via io.Copy in the generated Visit method), not the raw -// ResponseWriter/Flusher — so we bridge with an io.Pipe. ctx is the request -// context (the generated wrapper passes r.Context()); when the client -// disconnects it cancels, the SSE loop returns, and pw.Close() ends the -// io.Copy. Without threading ctx through, each disconnected client would -// leak a goroutine holding a pipe. -// -// Note: io.Copy does not flush per-write, so delivery is chunk-buffered -// rather than strictly real-time. Adequate for the event feed; a raw -// flushing handler is a possible future refinement. -func (s *Server) StreamEvents(ctx context.Context, req gen.StreamEventsRequestObject) (gen.StreamEventsResponseObject, error) { - pr, pw := io.Pipe() - go func() { - s.serveSSEWriter(ctx, pw, req.Params) - pw.Close() - }() - return gen.StreamEvents200TexteventStreamResponse{ - Body: pr, - ContentLength: 0, // unknown/streaming — omit the Content-Length header - }, nil -} +// Protocol: https://html.spec.whatwg.org/multipage/server-sent-events.html +// 1. If Last-Event-ID is present, replay buffered events (broker, then DB). +// 2. Subscribe and forward events fanned out from pg_notify. +// 3. Colon-comment heartbeat every 15s. +// 4. Unsubscribe on client disconnect (request context cancels). +func (s *Server) serveSSE(w http.ResponseWriter, r *http.Request) { + flusher, ok := w.(http.Flusher) + if !ok { + writeProblem(w, r, http.StatusInternalServerError, "internal error", "streaming not supported") + return + } -// serveSSEWriter runs the SSE loop writing to an io.Writer until the request -// context is cancelled (client disconnect / server shutdown). -func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen.StreamEventsParams) { - ctx, cancel := context.WithCancel(parent) + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering + w.WriteHeader(http.StatusOK) + flusher.Flush() + + ctx, cancel := context.WithCancel(r.Context()) defer cancel() sub := &sseSubscriber{ @@ -277,27 +273,27 @@ func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen. close(sub.done) }() - // Replay on Last-Event-ID - if params.LastEventID != nil && *params.LastEventID != "" { - id, err := strconv.ParseInt(*params.LastEventID, 10, 64) - if err == nil { + // ── 1. Replay on Last-Event-ID ──────────────────────────────── + if lastID := r.Header.Get("Last-Event-ID"); lastID != "" { + if id, err := strconv.ParseInt(lastID, 10, 64); err == nil { + // The in-memory broker only holds recent events; if it doesn't + // cover the whole gap, fall back to the DB for a complete replay. events := s.sseBroker.after(id) - if len(events) > 0 { + complete := len(events) > 0 && events[len(events)-1].ID == s.sseBroker.latestID() + if complete { for _, ev := range events { - if !writeSSE(w, nil, ev) { + if !writeSSE(w, flusher, ev) { return } } - } - // If broker didn't have them all, fetch from DB - if len(events) == 0 || events[len(events)-1].ID != s.sseBroker.latestID() { + } else { q := sqlcgen.New(s.pool) dbEvents, err := q.ListEventsAfter(ctx, sqlcgen.ListEventsAfterParams{ID: id, Limit: 5000}) if err != nil { slog.Error("sse db replay failed", "error", err) } else { for _, ev := range dbEvents { - if !writeSSE(w, nil, ev) { + if !writeSSE(w, flusher, ev) { return } } @@ -306,7 +302,7 @@ func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen. } } - // Subscribe and forward + // ── 2. Subscribe and forward ────────────────────────────────── heartbeat := time.NewTicker(15 * time.Second) defer heartbeat.Stop() @@ -318,18 +314,26 @@ func (s *Server) serveSSEWriter(parent context.Context, w io.Writer, params gen. if !ok { return } - if !writeSSE(w, nil, ev) { + if !writeSSE(w, flusher, ev) { return } case <-heartbeat.C: - _, err := fmt.Fprintf(w, ": heartbeat\n\n") - if err != nil { + if _, err := fmt.Fprintf(w, ": heartbeat\n\n"); err != nil { return } + flusher.Flush() } } } +// StreamEvents satisfies the generated ServerInterface, but the SSE route is +// served by the raw serveSSE handler registered in NewHandler (which wins +// over this generated route). If this method is ever reached, routing has +// regressed — fail loudly rather than silently chunk-buffering. +func (s *Server) StreamEvents(ctx context.Context, req gen.StreamEventsRequestObject) (gen.StreamEventsResponseObject, error) { + return nil, fmt.Errorf("%w: SSE must be served by the raw handler", errNotImplemented) +} + // Ensure pgxpool is imported — used via Acquire. var _ = &pgxpool.Pool{} var _ = pgx.ErrNoRows diff --git a/internal/httpapi/sse_test.go b/internal/httpapi/sse_test.go new file mode 100644 index 0000000..b60122f --- /dev/null +++ b/internal/httpapi/sse_test.go @@ -0,0 +1,89 @@ +package httpapi + +// Real-connection SSE test. httptest.NewRecorder buffers and never flushes, +// so this uses httptest.NewServer + a streaming client to verify that: +// - the raw serveSSE handler (not the generated 501 stub) serves the route, +// - an event created *after* the client connects is delivered in real time +// (i.e. flushed before the connection closes), +// - the SSE `data:` payload is the canonical gen.Event shape. +// Guarded by OIKOS_TEST_DATABASE_URL. + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestSSEStreamRealtimeDelivery(t *testing.T) { + srv := httptest.NewServer(newTestHandler(t, devConfig())) + defer srv.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // Connect to the stream. + req, _ := http.NewRequestWithContext(ctx, "GET", srv.URL+"/api/v1/events/stream", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("connect stream: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatalf("stream status = %d, want 200", resp.StatusCode) + } + if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") { + t.Fatalf("content-type = %q, want text/event-stream", ct) + } + + // Read SSE frames in a goroutine. + dataCh := make(chan map[string]any, 4) + go func() { + sc := bufio.NewScanner(resp.Body) + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "data: ") { + var m map[string]any + if json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &m) == nil { + dataCh <- m + } + } + } + }() + + // Give the subscriber a moment to register, then trigger an event by + // POSTing to the SAME live server (same DB → NOTIFY the listener sees). + time.Sleep(300 * time.Millisecond) + payload, _ := json.Marshal(map[string]any{"slug": "service:sse-rt", "type": "service", "name": "sse-rt"}) + cResp, err := http.Post(srv.URL+"/api/v1/entities", "application/json", bytes.NewReader(payload)) + if err != nil { + t.Fatalf("trigger create: %v", err) + } + cResp.Body.Close() + if cResp.StatusCode != 201 { + t.Fatalf("trigger create status = %d, want 201", cResp.StatusCode) + } + + // The event must arrive in real time (well before the 10s ctx deadline), + // proving the handler flushes rather than buffering until close. + select { + case ev := <-dataCh: + if ev["type"] != "entity.created" { + t.Errorf("event type = %v, want entity.created", ev["type"]) + } + // canonical shape: snake_case + decoded data object + if _, ok := ev["entity_id"]; !ok { + t.Errorf("missing snake_case entity_id: %v", ev) + } + if d, ok := ev["data"].(map[string]any); !ok || d["slug"] != "service:sse-rt" { + t.Errorf("data not a decoded object with slug: %v", ev["data"]) + } + case <-time.After(3 * time.Second): + t.Fatal("SSE event not delivered within 3s (flushing broken?)") + } +}