dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

phase 2 review: fix SSE deadlock, MCP panic, lifecycle 500, NOT NULL bug

Browse Source 
Reviewed the phase-2 implementation (parts 2–5) end to end. The suite hung
for 600s and several handlers were never exercised because there were no
tests for the new mutation/event/MCP surface. Fixes:

- CRITICAL: sseListener ran on context.Background() and held a pooled
  connection forever, so pool.Close() deadlocked (600s test timeout).
  NewHandler now takes a ctx that governs the listener; ListenAndServe and
  the test helper cancel it before closing the pool.
- CRITICAL: MCP AddTool panicked ("missing input schema") at construction
  under go-sdk v1.6.1 — so NewHandler (and every API handler) panicked.
  Added object input schemas to all 8 tools via an objSchema helper.
- HIGH: PatchEntity parsed lifecycle transitions as map[string][]string but
  the shape is {from:{to:{requires:[]}}}, so every state-change PATCH 500'd.
  Parse the nested shape; allow same-state no-ops.
- HIGH: CreateEntity bound SQL NULL for attributes when omitted, violating
  the NOT NULL column (the default only applies when omitted). Default to
  '{}'.
- MED: serveSSEWriter ignored the request ctx (per-client goroutine leak on
  disconnect) and set an invalid Content-Length: -1. Thread ctx through;
  omit the header. writeSSE now nil-checks the flusher (io.Pipe path passed
  nil → would have panicked on first event).
- MED: SSE `data:` leaked raw sqlcgen.Event (PascalCase, base64 JSONB).
  Emit canonical gen.Event so SSE matches GET /events. Verified live.
- LOW: CreateEntity uses uuid.NewV7 (ADR-0005) + real actor from context in
  audit; removed dead bearerAuth; fixed vet unkeyed-field warnings.

Tests (would have caught all of the above): entity create/patch with
If-Match 409/400, valid+invalid lifecycle transitions, idempotency replay,
duplicate-slug 409, abstract-type 422, event+audit side effects, MCP tool
registration. Live smoke test confirmed NOTIFY→listener→SSE delivery.

Also adds the missing Phase 2 deliverable: Gitea Actions CI (vet,
golangci-lint, govulncheck, generated-code drift guard, race tests against
TimescaleDB, docker build) and wires sqlc into `make generate`.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
authentik Default Admin
2026-07-07 13:26:43 +02:00
parent 6b61495e3b
commit f1b0b65149
10 changed files with 457 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
internal/httpapi/sse.go
View File

@@ -98,114 +98,6 @@ type sseSubscriber struct {
cancel context.CancelFunc
}
// serveSSE is the streaming handler for GET /api/v1/events/stream.
//
// Protocol: https://html.spec.whatwg.org/multipage/server-sent-events.html
//
// 1. If Last-Event-ID is present, replay buffered events from the in-memory
// broker (or fall back to ListEventsAfter for cold start).
// 2. Subscribe via in-memory channel and forward events from pg_notify.
// 3. Send a colon-comment heartbeat every 15 s.
// 4. Unsubscribe and clean up on client disconnect.
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
}
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 nginx buffering
w.WriteHeader(http.StatusOK)
flusher.Flush()
ctx := r.Context()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
sub := &sseSubscriber{
ch: make(chan sqlcgen.Event, 64),
done: make(chan struct{}),
cancel: cancel,
}
s.sseMu.Lock()
s.sseSubs[sub] = struct{}{}
s.sseMu.Unlock()
defer func() {
s.sseMu.Lock()
delete(s.sseSubs, sub)
s.sseMu.Unlock()
close(sub.done)
}()
// ── 1. Replay ─────────────────────────────────────────────────
if lastID := r.Header.Get("Last-Event-ID"); lastID != "" {
id, err := strconv.ParseInt(lastID, 10, 64)
if err == nil {
replayed := 0
// Try in-memory broker first
events := s.sseBroker.after(id)
if len(events) > 0 {
for _, ev := range events {
if !writeSSE(w, flusher, ev) {
return
}
replayed++
}
}
// If broker didn't have them all, fetch from DB
if replayed == 0 || events[len(events)-1].ID != s.sseBroker.latestID() {
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, flusher, ev) {
return
}
}
}
}
}
}
// ── 2. Subscribe and forward ──────────────────────────────────
heartbeat := time.NewTicker(15 * time.Second)
defer heartbeat.Stop()
for {
select {
case <-ctx.Done():
return
case ev, ok := <-sub.ch:
if !ok {
return
}
if !writeSSE(w, flusher, ev) {
return
}
case <-heartbeat.C:
_, err := fmt.Fprintf(w, ": heartbeat\n\n")
if err != nil {
return
}
flusher.Flush()
}
}
}
// sseListener runs in a background goroutine: it opens a dedicated pgx
// connection, LISTENs on oikos_events, and fans out each notification to
// all live subscribers. Runs until ctx is cancelled.
@@ -290,10 +182,36 @@ func (s *Server) sseListener(ctx context.Context) {
}
}
// sqlcEventToGen converts a DB event row to the canonical wire shape so the
// SSE `data:` payload matches GET /events (snake_case keys, decoded data
// object) rather than leaking Go field names and base64-encoded JSONB.
func sqlcEventToGen(ev sqlcgen.Event) gen.Event {
out := gen.Event{
Id: int(ev.ID),
Ts: ev.Ts,
Type: ev.Type,
Severity: gen.EventSeverity(ev.Severity),
Source: ev.Source,
CorrelationId: ev.CorrelationID,
}
if ev.EntityID != nil {
s := ev.EntityID.String()
out.EntityId = &s
}
if len(ev.Data) > 0 {
var data map[string]any
if json.Unmarshal(ev.Data, &data) == nil && len(data) > 0 {
out.Data = &data
}
}
return out
}
// writeSSE writes a single Event as an SSE message. Returns false if the
// write failed (client disconnected).
// write failed (client disconnected). flusher may be nil (io.Pipe path,
// which has no separate flush step).
func writeSSE(w ioWriter, flusher http.Flusher, ev sqlcgen.Event) bool {
data, err := json.Marshal(ev)
data, err := json.Marshal(sqlcEventToGen(ev))
if err != nil {
return true // skip un-serializable events
}
@@ -301,7 +219,9 @@ func writeSSE(w ioWriter, flusher http.Flusher, ev sqlcgen.Event) bool {
if err != nil {
return false
}
flusher.Flush()
if flusher != nil {
flusher.Flush()
}
return true
}
@@ -312,28 +232,34 @@ type ioWriter interface {
}
// StreamEvents implements the OpenAPI interface (SSE event stream).
// Uses io.Pipe to bridge the streaming SSE goroutine to the response body.
//
// 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() {
// Wrap the pipe writer as an http.ResponseWriter-like struct
// that implements http.Flusher via calling flush on the pipe
// (which isn't a real flusher — we use the io.Pipe writer directly
// via writeSSE's ioWriter interface).
s.serveSSEWriter(pw, req.Params)
s.serveSSEWriter(ctx, pw, req.Params)
pw.Close()
}()
return gen.StreamEvents200TexteventStreamResponse{
Body: pr,
ContentLength: -1, // unknown length
ContentLength: 0, // unknown/streaming — omit the Content-Length header
}, nil
}
// serveSSEWriter runs the SSE loop writing to an io.Writer.
func (s *Server) serveSSEWriter(w io.Writer, params gen.StreamEventsParams) {
// No explicit flusher for pipe writes — io.Pipe flushes on each Write.
ctx := context.Background()
ctx, cancel := context.WithCancel(ctx)
// 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)
defer cancel()
sub := &sseSubscriber{