From f526f182788e9a3d374215a76416aea1e8370e1a Mon Sep 17 00:00:00 2001 From: patel-lyzr Date: Sat, 9 May 2026 00:31:08 +0530 Subject: [PATCH] feat: enhance SSE handling and API integration with improved event streaming and buffering configurations --- Makefile | 7 +- pkg/api/agents.go | 10 +- pkg/api/server.go | 31 ++++- pkg/execevents/bus.go | 135 ++++++++++++++-------- pkg/executors/build.go | 18 ++- web/app/executions/view/page.tsx | 49 +++++--- web/components/canvas/pipeline-canvas.tsx | 36 ++++-- web/lib/api.ts | 32 ++++- web/nginx.conf | 23 +++- 9 files changed, 256 insertions(+), 85 deletions(-) diff --git a/Makefile b/Makefile index 9dde67e..f4b0d48 100644 --- a/Makefile +++ b/Makefile @@ -33,7 +33,12 @@ web: cd web && (test -d node_modules || npm install --no-fund --no-audit --loglevel=error) && npm run build web-dev: - cd web && (test -d node_modules || npm install --no-fund --no-audit --loglevel=error) && npm run dev + # NEXT_PUBLIC_FLOW_API_URL points the SSE EventSource straight at the + # Go API so streams skip Next's trailingSlash 308 redirect (which + # EventSource doesn't follow). Override at the command line if your + # Go API runs elsewhere. + cd web && (test -d node_modules || npm install --no-fund --no-audit --loglevel=error) \ + && NEXT_PUBLIC_FLOW_API_URL=$${NEXT_PUBLIC_FLOW_API_URL:-http://localhost:8090} npm run dev # `make dev` runs the Next.js dev server (auto-installs deps). # In another terminal run `make watch` to hot-reload the Go API on :8090; diff --git a/pkg/api/agents.go b/pkg/api/agents.go index 2de7052..0d5473c 100644 --- a/pkg/api/agents.go +++ b/pkg/api/agents.go @@ -538,13 +538,19 @@ func (s *Server) handleGitHubWebhook(w http.ResponseWriter, r *http.Request) { return } + // Normalize the ref to the bare branch name (`main`, not `refs/heads/main`) + // so downstream nodes — particularly Build's git clone --branch — don't + // have to know about Git's internal ref namespace. `fullRef` is kept for + // nodes that want the original. + branch := github.BranchFromRef(push.Ref) trigger := []map[string]any{{ "source": "github_push", "agentId": a.ID, "agentName": a.Name, "repoUrl": a.RepoURL, - "ref": push.Ref, - "branch": github.BranchFromRef(push.Ref), + "ref": branch, + "branch": branch, + "fullRef": push.Ref, "commit": push.After, "pusher": push.Pusher.Name, }} diff --git a/pkg/api/server.go b/pkg/api/server.go index 65fdbfb..0cd5e16 100644 --- a/pkg/api/server.go +++ b/pkg/api/server.go @@ -27,6 +27,14 @@ import ( "github.com/lyzrai/flow/pkg/storage" ) +// ssePrimer is a >2 KB SSE comment block sent as the first chunk of every +// stream response. Buffer-aware proxies (Cloudflare quick-tunnels, some CDN +// edges, http/2 windowing on slow links) hold back small chunks until they +// reach a flush threshold. Padding past that threshold makes node-by-node +// events arrive immediately instead of in one burst at the end of the run. +// Lines beginning with `:` are SSE comments — clients ignore them silently. +var ssePrimer = ":" + strings.Repeat(" ", 2049) + "\n\n" + // FlowSummary is the list-shape returned to the dashboard. type FlowSummary struct { ID string `json:"id"` @@ -133,10 +141,18 @@ func (s *Server) routes() { s.mux.HandleFunc("DELETE /api/workflows/{id}", s.handleDeleteFlow) s.mux.HandleFunc("POST /api/workflows/execute", s.handleExecuteWorkflow) s.mux.HandleFunc("GET /api/executions", s.handleListExecutions) + // Stream routes register both with-and-without trailing slash so the + // Next dev server's `trailingSlash: true` rewrite (which appends "/") + // reaches the same handler as a direct call. Without this the SSE + // stream returns the literal redirect text via the Next proxy and the + // UI never sees node_started/log events. s.mux.HandleFunc("GET /api/runs/stream", s.handleRunsStream) + s.mux.HandleFunc("GET /api/runs/stream/", s.handleRunsStream) s.mux.HandleFunc("GET /api/executions/{id}", s.handleGetExecution) s.mux.HandleFunc("GET /api/executions/{id}/stream", s.handleStreamExecution) + s.mux.HandleFunc("GET /api/executions/{id}/stream/", s.handleStreamExecution) s.mux.HandleFunc("GET /api/executions/{id}/logs/{node}", s.handleNodeLog) + s.mux.HandleFunc("GET /api/executions/{id}/logs/{node}/", s.handleNodeLog) s.mux.HandleFunc("POST /api/executions/{id}/resume", s.handleResumeExecution) // Agents — Langship-style agent registry (git URL + PAT) @@ -535,16 +551,19 @@ func (s *Server) handleStreamExecution(w http.ResponseWriter, r *http.Request) { } w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Cache-Control", "no-cache, no-transform") w.Header().Set("Connection", "keep-alive") - w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering + w.Header().Set("X-Accel-Buffering", "no") // disable proxy buffering (nginx) w.WriteHeader(http.StatusOK) ch, cancel := s.events.Subscribe(id) defer cancel() - // Tell the client which execution it's subscribed to (also primes the - // SSE pipe so flushers in the middle don't withhold the first byte). + // Prime the stream with a >2 KB comment so intermediate proxies that + // buffer based on byte threshold (Cloudflare quick-tunnel, some CDN + // edges) flush past the threshold immediately. SSE comments start with + // `:` and are ignored by EventSource. + _, _ = fmt.Fprint(w, ssePrimer) _, _ = fmt.Fprintf(w, "event: open\ndata: {\"execution_id\":%q}\n\n", id) flusher.Flush() @@ -586,7 +605,7 @@ func (s *Server) handleRunsStream(w http.ResponseWriter, r *http.Request) { return } w.Header().Set("Content-Type", "text/event-stream") - w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Cache-Control", "no-cache, no-transform") w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") w.WriteHeader(http.StatusOK) @@ -594,6 +613,8 @@ func (s *Server) handleRunsStream(w http.ResponseWriter, r *http.Request) { ch, cancel := s.runsBus.Subscribe() defer cancel() + // See handleStreamExecution for why this padding is required. + _, _ = fmt.Fprint(w, ssePrimer) _, _ = fmt.Fprint(w, "event: open\ndata: {}\n\n") flusher.Flush() diff --git a/pkg/execevents/bus.go b/pkg/execevents/bus.go index 99ca221..fd0f138 100644 --- a/pkg/execevents/bus.go +++ b/pkg/execevents/bus.go @@ -10,98 +10,143 @@ package execevents import ( "context" "sync" + "time" "github.com/lyzrai/flow/pkg/engine" ) +// retainPerExec is the cap on per-execution event history. We keep the +// most-recent N events so a subscriber that arrives mid-run can replay +// what it missed (node_started lifecycle events especially — those fire +// fast, often before the UI has connected). +const retainPerExec = 500 + +// retainAfterDone is how long to hold the buffer for an execution after +// the terminal `done` event arrives. Late subscribers (e.g. a user who +// opens the run page right after success) get the full replay. +const retainAfterDone = 5 * time.Minute + // MemoryBus is a single-process publish/subscribe bus keyed by execution ID. -// Subscribers receive every event published for their execution until they -// unsubscribe or the channel buffer fills (slow subscribers are dropped to -// keep the publisher non-blocking). +// Subscribers receive every event published for their execution; on +// subscribe they additionally receive a backlog replay of events emitted +// before they connected. Bounded retention per exec keeps memory in check. type MemoryBus struct { - mu sync.RWMutex - subscribers map[string][]*subscription - // terminal stores the final event per exec so a subscriber that arrives - // late still gets a "done" / "error" event and closes cleanly. - terminal map[string]engine.ExecutionEvent + mu sync.RWMutex + streams map[string]*execStream +} + +type execStream struct { + subs []*subscription + history []engine.ExecutionEvent + done bool + doneAt time.Time } type subscription struct { - ch chan engine.ExecutionEvent - closed bool - once sync.Once + ch chan engine.ExecutionEvent + once sync.Once } -// NewMemoryBus returns a fresh in-memory bus. +// NewMemoryBus returns a fresh in-memory bus and starts a background +// sweeper that drops stale streams. The returned bus has no Close — the +// process owns the lifecycle. func NewMemoryBus() *MemoryBus { - return &MemoryBus{ - subscribers: map[string][]*subscription{}, - terminal: map[string]engine.ExecutionEvent{}, - } + b := &MemoryBus{streams: map[string]*execStream{}} + go b.sweep() + return b } -// Emit implements engine.Emitter. Non-blocking: if a subscriber's channel is -// full we drop the event for that subscriber (publisher must not stall). +// Emit implements engine.Emitter. Non-blocking: if a subscriber's channel +// is full the event is dropped for that subscriber (publisher must never +// stall) but stays in the per-exec history so a fresh subscriber can still +// see it. func (b *MemoryBus) Emit(_ context.Context, execID string, e engine.ExecutionEvent) { if execID == "" { return } b.mu.Lock() - subs := append([]*subscription(nil), b.subscribers[execID]...) - if isTerminal(e.Type) { - b.terminal[execID] = e + st := b.streams[execID] + if st == nil { + st = &execStream{} + b.streams[execID] = st } + // Append + cap. + st.history = append(st.history, e) + if over := len(st.history) - retainPerExec; over > 0 { + st.history = st.history[over:] + } + if e.Type == engine.EventDone { + st.done = true + st.doneAt = time.Now() + } + subs := append([]*subscription(nil), st.subs...) b.mu.Unlock() for _, s := range subs { select { case s.ch <- e: default: - // drop — slow subscriber + // slow subscriber — drop } } } -// Subscribe returns a channel that receives every event for execID. The -// caller must call the returned cancel func when done. If a terminal event -// was already published before subscribe, it is replayed once so the caller -// can shut down cleanly. +// Subscribe registers for events for execID. On subscribe the caller +// receives every event already retained for this execution (in order), +// followed by every new event. Returns the channel and a cancel func. func (b *MemoryBus) Subscribe(execID string) (<-chan engine.ExecutionEvent, func()) { - s := &subscription{ch: make(chan engine.ExecutionEvent, 32)} + // Buffer ≥ history cap so the initial replay never drops. + s := &subscription{ch: make(chan engine.ExecutionEvent, retainPerExec+32)} b.mu.Lock() - b.subscribers[execID] = append(b.subscribers[execID], s) - term, hadTerm := b.terminal[execID] + st := b.streams[execID] + if st == nil { + st = &execStream{} + b.streams[execID] = st + } + st.subs = append(st.subs, s) + // Snapshot history under the lock. + backlog := append([]engine.ExecutionEvent(nil), st.history...) b.mu.Unlock() - if hadTerm { - // non-blocking — buffer is fresh - s.ch <- term + // Replay outside the lock. Buffer is sized so this never blocks. + for _, e := range backlog { + s.ch <- e } cancel := func() { s.once.Do(func() { b.mu.Lock() - cur := b.subscribers[execID] - out := cur[:0] - for _, x := range cur { - if x != s { - out = append(out, x) + st := b.streams[execID] + if st != nil { + out := st.subs[:0] + for _, x := range st.subs { + if x != s { + out = append(out, x) + } } - } - if len(out) == 0 { - delete(b.subscribers, execID) - } else { - b.subscribers[execID] = out + st.subs = out } b.mu.Unlock() - s.closed = true close(s.ch) }) } return s.ch, cancel } -func isTerminal(t engine.EventType) bool { - return t == engine.EventDone +// sweep periodically GC's streams that are done + past the retention +// window AND have no live subscribers. +func (b *MemoryBus) sweep() { + t := time.NewTicker(1 * time.Minute) + defer t.Stop() + for range t.C { + now := time.Now() + b.mu.Lock() + for id, st := range b.streams { + if st.done && len(st.subs) == 0 && now.Sub(st.doneAt) > retainAfterDone { + delete(b.streams, id) + } + } + b.mu.Unlock() + } } diff --git a/pkg/executors/build.go b/pkg/executors/build.go index 981cc4d..00da397 100644 --- a/pkg/executors/build.go +++ b/pkg/executors/build.go @@ -76,7 +76,11 @@ func (e *BuildExecutor) Execute(ctx context.Context, node models.NodeDef, inputs } commitSHA, _ := trigger["commit"].(string) - ref := strFirst(strFromAny(trigger["ref"]), a.Ref, "main") + // `git clone --branch` wants a bare name like "main"; if a webhook + // payload (or older trigger record) carried "refs/heads/main", trim it + // so the clone doesn't fail with "Remote branch refs/heads/main not + // found in upstream origin". + ref := stripRefsHeads(strFirst(strFromAny(trigger["ref"]), a.Ref, "main")) cloneDir, cleanup, err := cloneRepo(ctx, a, ref, commitSHA, time.Duration(timeoutSec)*time.Second) if err != nil { @@ -511,6 +515,18 @@ func oneLineSummary(s string) string { return "" } +// stripRefsHeads turns "refs/heads/main" into "main"; passes any other +// shape through unchanged. Tags ("refs/tags/v1") would still need a +// different clone strategy (--branch works for both branches and tags so +// we leave those alone). +func stripRefsHeads(s string) string { + const p = "refs/heads/" + if strings.HasPrefix(s, p) { + return s[len(p):] + } + return s +} + func sanitizeEnvValue(s string) string { r := strings.NewReplacer("\n", " ", "\r", " ", "\x00", "") return r.Replace(s) diff --git a/web/app/executions/view/page.tsx b/web/app/executions/view/page.tsx index ac4ddb4..5e44f2a 100644 --- a/web/app/executions/view/page.tsx +++ b/web/app/executions/view/page.tsx @@ -75,6 +75,12 @@ function ExecutionView() { const [run, setRun] = useState(null); const [pipelineDef, setPipelineDef] = useState(null); const [nodeStatuses, setNodeStatuses] = useState({}); + // Per-node tick when we first marked it running. Used to enforce a + // minimum visible "running" duration so the user always sees the + // spinner — even for instantaneous nodes (Trigger, NoOp). Without + // this, fast nodes flicker pending → success in one render batch and + // the running state is invisible. + const runningSinceRef = useRef>({}); const [nodeLogs, setNodeLogs] = useState({}); const [nodeDurations, setNodeDurations] = useState({}); const [streamConnected, setStreamConnected] = useState(false); @@ -130,22 +136,39 @@ function ExecutionView() { return; } if (ev.node) { - setNodeStatuses((prev) => { - const next: NodeStatus = - ev.type === "node_started" - ? "running" - : ev.type === "node_completed" - ? "success" - : ev.type === "node_error" - ? "failed" - : (prev[ev.node!] ?? "pending"); - return { ...prev, [ev.node!]: next }; - }); - if (ev.type === "node_completed" || ev.type === "node_error") { + const node = ev.node; + const minVisibleMs = 400; + + if (ev.type === "node_started") { + runningSinceRef.current[node] = Date.now(); + setNodeStatuses((prev) => ({ ...prev, [node]: "running" })); + } else if (ev.type === "node_completed" || ev.type === "node_error") { + const final: NodeStatus = + ev.type === "node_completed" ? "success" : "failed"; + const startedAt = runningSinceRef.current[node]; + const elapsed = startedAt ? Date.now() - startedAt : Infinity; + + // Make sure the user actually sees a "running" frame. If we + // never recorded a start (subscriber arrived after the start + // event flushed) we apply the terminal state immediately. + if (startedAt === undefined || elapsed >= minVisibleMs) { + setNodeStatuses((prev) => ({ ...prev, [node]: final })); + } else { + // Briefly show "running" first if we missed it, then flip. + setNodeStatuses((prev) => ({ + ...prev, + [node]: prev[node] === "running" ? "running" : "running", + })); + setTimeout(() => { + setNodeStatuses((prev) => ({ ...prev, [node]: final })); + }, minVisibleMs - elapsed); + } + delete runningSinceRef.current[node]; + if (typeof ev.duration_ms === "number") { setNodeDurations((prev) => ({ ...prev, - [ev.node!]: ev.duration_ms!, + [node]: ev.duration_ms!, })); } } diff --git a/web/components/canvas/pipeline-canvas.tsx b/web/components/canvas/pipeline-canvas.tsx index 336f0e4..93ad51e 100644 --- a/web/components/canvas/pipeline-canvas.tsx +++ b/web/components/canvas/pipeline-canvas.tsx @@ -70,6 +70,8 @@ function CanvasInner({ fullBleed, nodeStatuses, }: PipelineCanvasProps) { + // (Hook order: nodes/edges state declared below so this comment sits at + // the top of the component for context.) // Compute initial RF state once. The canvas owns it from here on. const initial = useMemo(() => toReactFlow(initialValue ?? null), []); // eslint-disable-next-line react-hooks/exhaustive-deps -- intentional: load-once @@ -78,6 +80,28 @@ function CanvasInner({ const [edges, setEdges, onEdgesChange] = useEdgesState(initial.edges); const [selectedId, setSelectedId] = useState(null); + // Merge external runStatus into the RF-owned node state. We can't just + // pass a freshly-mapped `nodes` prop to because useNodesState + // makes RF the source of truth — external props get overridden by the + // internal store on the next render. Instead we patch the store directly + // whenever nodeStatuses changes. Skips updates when the value is + // unchanged so we don't churn React Flow on every poll tick. + useEffect(() => { + if (!nodeStatuses) return; + setNodes((cur) => + cur.map((n) => { + const next = nodeStatuses[n.id] ?? "pending"; + const prev = + (n.data as FlowNodeData & { runStatus?: string }).runStatus ?? "pending"; + if (prev === next) return n; + return { + ...n, + data: { ...(n.data as FlowNodeData), runStatus: next }, + }; + }) + ); + }, [nodeStatuses, setNodes]); + const wrapperRef = useRef(null); const { screenToFlowPosition } = useReactFlow(); @@ -233,17 +257,7 @@ function CanvasInner({ onDrop={onDrop} > ({ - ...n, - data: { - ...n.data, - runStatus: nodeStatuses[n.id] ?? "pending", - }, - })) - : nodes - } + nodes={nodes} edges={edges} onNodesChange={onNodesChange} onEdgesChange={onEdgesChange} diff --git a/web/lib/api.ts b/web/lib/api.ts index 3a04fc0..60b3425 100644 --- a/web/lib/api.ts +++ b/web/lib/api.ts @@ -61,7 +61,24 @@ export type ServerConfig = { orchestratorEnabled: boolean; }; -const base = ""; // same-origin +// REST + page calls go same-origin (Next rewrite proxies /api → Go in dev, +// nginx proxies /api → flow:8090 in prod). +const base = ""; + +// SSE base for streaming endpoints. +// +// Default: same-origin (works behind any reverse proxy that doesn't buffer +// — nginx with `proxy_buffering off`, our prod config; Cloudflare tunnels; +// most production setups). +// +// Dev override: set NEXT_PUBLIC_FLOW_API_URL=http://localhost:8090 to hit +// the Go server directly, bypassing Next's dev rewrite (which buffers +// chunked responses, breaking node-by-node updates) and Next's 308 redirect +// from `trailingSlash: true` (which EventSource won't follow). +const sseBase = + (typeof process !== "undefined" && + process.env?.NEXT_PUBLIC_FLOW_API_URL) || + ""; async function handle(res: Response): Promise { if (!res.ok) { @@ -168,11 +185,16 @@ export const api = { ), // --- runs --- - /** Returns the EventSource URL for SSE streaming of an execution. */ - executionStreamURL: (id: string) => `${base}/api/executions/${id}/stream`, + /** Returns the EventSource URL for SSE streaming of an execution. + * Uses `sseBase` so dev can hit the Go API directly (skipping Next's + * trailingSlash 308 which EventSource won't follow). Trailing slash on + * the path keeps things consistent if the user does proxy through Next + * or nginx; the Go mux registers both forms either way. */ + executionStreamURL: (id: string) => `${sseBase}/api/executions/${id}/stream/`, - /** Global runs feed — fires once per dispatched run. */ - runsStreamURL: () => `${base}/api/runs/stream`, + /** Global runs feed — fires once per dispatched run. Same dev-bypass + * reasoning as executionStreamURL. */ + runsStreamURL: () => `${sseBase}/api/runs/stream/`, listRuns: (params?: { pipelineId?: string; limit?: number }) => { const qs = new URLSearchParams(); diff --git a/web/nginx.conf b/web/nginx.conf index 39babbc..19eab08 100644 --- a/web/nginx.conf +++ b/web/nginx.conf @@ -12,8 +12,27 @@ server { try_files $uri =404; } - # Proxy API calls to the Go service. - # `flow` is the service name on the docker-compose network. + # SSE endpoints. Buffering off + long read timeout so events flush + # node-by-node instead of getting stuck in nginx's buffer. Heartbeats + # every 15s on the Go side keep the connection alive. + location ~ ^/api/(executions/[^/]+/stream|runs/stream)/?$ { + proxy_pass http://flow:8090; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header Connection ""; + + # Disable everything that would prevent immediate event flushing. + proxy_buffering off; + proxy_cache off; + proxy_read_timeout 24h; + proxy_send_timeout 24h; + chunked_transfer_encoding off; + } + + # Regular JSON API. location /api/ { proxy_pass http://flow:8090; proxy_http_version 1.1;