Complete SSE refactor: migrate consumers, extract invalidators, drop singleton

web-admin/package.json

@@ -7,6 +7,7 @@
     "build": "npm run generate:sveltekit -w web-common && vite build && cp _redirects build",
     "preview": "vite preview",
     "test": "playwright test",
+    "test:unit": "vitest run",
     "generate:sveltekit": "svelte-kit sync",
     "test:setup": "E2E_NO_TEARDOWN=true playwright test --project=setup",
     "test:dev": "E2E_NO_SETUP_OR_TEARDOWN=true playwright test --project=e2e",

web-admin/src/features/projects/status/logs/ProjectLogsPage.svelte

@@ -2,11 +2,15 @@
   import { page } from "$app/stores";
   import { onMount, onDestroy } from "svelte";
   import {
-    SSEConnectionManager,
     ConnectionStatus,
-  } from "@rilldata/web-common/runtime-client/sse-connection-manager";
+    createSSEStream,
+  } from "@rilldata/web-common/runtime-client/sse";
   import { useRuntimeClient } from "@rilldata/web-common/runtime-client/v2";
-  import { V1LogLevel, type V1Log } from "@rilldata/web-common/runtime-client";
+  import {
+    V1LogLevel,
+    type V1Log,
+    type V1WatchLogsResponse,
+  } from "@rilldata/web-common/runtime-client";
   import Search from "@rilldata/web-common/components/search/Search.svelte";
   import * as DropdownMenu from "@rilldata/web-common/components/dropdown-menu";
   import CaretDownIcon from "@rilldata/web-common/components/icons/CaretDownIcon.svelte";
@@ -16,6 +20,7 @@
     parseArrayParam,
     parseStringParam,
   } from "@rilldata/web-common/lib/url-filter-sync";
+  import { ProjectLogsStore, type LogEntry } from "./project-logs-store";
 
   const runtimeClient = useRuntimeClient();
 
@@ -29,9 +34,9 @@
     { value: V1LogLevel.LOG_LEVEL_ERROR, label: "Error" },
   ];
 
-  type LogEntry = V1Log & { _id: number };
-  let nextLogId = 0;
-  let logs: LogEntry[] = [];
+  const logStore = new ProjectLogsStore(MAX_LOGS);
+  // Bumped on each addLog so Svelte reactivity re-runs filtering.
+  let logsVersion = 0;
   let logsContainer: HTMLDivElement;
   let connectionError: string | null = null;
   let filterDropdownOpen = false;
@@ -57,29 +62,47 @@
     filterSync.syncToUrl({ q: searchText, level: selectedLevels });
   }
 
-  const logsConnection = new SSEConnectionManager({
-    maxRetryAttempts: 5,
-    retryOnError: true,
-    retryOnClose: true,
+  // Typed decoder layer: `createSSEStream` composes connection + subscriber so
+  // component code stays in UI concerns.
+  const logsStream = createSSEStream<{
+    log: V1WatchLogsResponse;
+    error: { code?: string; message?: string };
+  }>({
+    connection: {
+      maxRetryAttempts: 5,
+      retryOnError: true,
+      retryOnClose: true,
+    },
+    decoders: {
+      log: (data) => JSON.parse(data) as V1WatchLogsResponse,
+      error: (data) => {
+        try {
+          return JSON.parse(data) as { code?: string; message?: string };
+        } catch {
+          return { message: data };
+        }
+      },
+    },
   });
 
-  $: connectionStatus = logsConnection.status;
+  $: connectionStatus = logsStream.status;
   $: isConnected = $connectionStatus === ConnectionStatus.OPEN;
   $: isConnecting = $connectionStatus === ConnectionStatus.CONNECTING;
   $: isClosed = $connectionStatus === ConnectionStatus.CLOSED;
   $: hasConnectionError = isClosed && connectionError !== null;
 
-  $: filteredLogs = logs.filter((log) => {
-    const matchesLevel =
-      selectedLevels.length === 0 || selectedLevels.includes(log.level ?? "");
-    const matchesSearch =
-      !searchText ||
-      (log.message?.toLowerCase().includes(searchText.toLowerCase()) ??
-        false) ||
-      (log.jsonPayload?.toLowerCase().includes(searchText.toLowerCase()) ??
-        false);
-    return matchesLevel && matchesSearch;
-  });
+  let filteredLogs: LogEntry[] = [];
+  let totalLogs = 0;
+  $: {
+    // Depend on logsVersion so Svelte re-runs when new logs arrive; the
+    // reference below keeps the compiler from treating it as unused.
+    void logsVersion;
+    filteredLogs = logStore.getFiltered({
+      levels: selectedLevels,
+      search: searchText,
+    });
+    totalLogs = logStore.getAll().length;
+  }
 
   $: selectedLevelLabel = (() => {
     if (selectedLevels.length === 0) return "All levels";
@@ -105,48 +128,45 @@
     const url = `${host}/v1/instances/${instanceId}/sse?events=log&logs_replay=true&logs_replay_limit=${REPLAY_LIMIT}`;
 
     unsubs = [
-      logsConnection.on("message", handleMessage),
-      logsConnection.on("error", handleError),
-      logsConnection.on("open", handleOpen),
+      logsStream.on("log", handleLogMessage),
+      logsStream.on("error", handleServerError),
+      logsStream.onConnection("error", handleTransportError),
+      logsStream.onConnection("open", handleOpen),
     ];
 
-    logsConnection.start(url, {
+    logsStream.start(url, {
       getJwt: () => runtimeClient.getJwt(),
     });
   });
 
   onDestroy(() => {
     unsubs.forEach((fn) => fn());
-    logsConnection.close(true);
+    logsStream.cleanup();
   });
 
   function isNearBottom(el: HTMLElement, threshold = 50): boolean {
     return el.scrollHeight - el.scrollTop - el.clientHeight <= threshold;
   }
 
-  function handleMessage(message: { data: string; type?: string }) {
-    try {
-      if (message.type && message.type !== "log") return;
+  function handleLogMessage(response: V1WatchLogsResponse) {
+    const log = response.log as V1Log | undefined;
+    if (!log) return;
 
-      const response = JSON.parse(message.data);
-      const log = response.log as V1Log;
-      if (log) {
-        logs = [...logs, { ...log, _id: nextLogId++ }].slice(-MAX_LOGS);
+    logStore.addLog(log);
+    logsVersion++;
 
-        if (logsContainer && isNearBottom(logsContainer)) {
-          requestAnimationFrame(() => {
-            logsContainer.scrollTop = logsContainer.scrollHeight;
-          });
-        }
-      }
-    } catch (e) {
-      if (import.meta.env.DEV) {
-        console.warn("Failed to parse log message:", e);
-      }
+    if (logsContainer && isNearBottom(logsContainer)) {
+      requestAnimationFrame(() => {
+        logsContainer.scrollTop = logsContainer.scrollHeight;
+      });
     }
   }
 
-  function handleError(error: Error) {
+  function handleServerError(payload: { code?: string; message?: string }) {
+    connectionError = payload.message || payload.code || "Server error";
+  }
+
+  function handleTransportError(error: Error) {
     console.error("Logs SSE error:", error);
     connectionError = error.message || "Connection failed";
   }
@@ -161,7 +181,7 @@
 
     connectionError = null;
     const url = `${host}/v1/instances/${instanceId}/sse?events=log&logs_replay=true&logs_replay_limit=${REPLAY_LIMIT}`;
-    logsConnection.start(url, {
+    logsStream.start(url, {
       getJwt: () => runtimeClient.getJwt(),
     });
   }
@@ -300,7 +320,7 @@
         <span class="text-red-600">Connection failed: {connectionError}</span>
         <button class="retry-button" onclick={retryConnection}> Retry </button>
       </div>
-    {:else if logs.length === 0}
+    {:else if totalLogs === 0}
       <div class="empty-state">Waiting for logs...</div>
     {:else if filteredLogs.length === 0}
       <div class="empty-state">No logs match the current filters</div>

web-admin/src/features/projects/status/logs/project-logs-store.spec.ts

@@ -0,0 +1,86 @@
+import { V1LogLevel } from "@rilldata/web-common/runtime-client";
+import { describe, expect, it } from "vitest";
+import { ProjectLogsStore } from "./project-logs-store";
+
+describe("ProjectLogsStore", () => {
+  it("addLog assigns a monotonic _id counter", () => {
+    const store = new ProjectLogsStore(100);
+    const a = store.addLog({ message: "a" });
+    const b = store.addLog({ message: "b" });
+    const c = store.addLog({ message: "c" });
+    expect(a._id).toBe(0);
+    expect(b._id).toBe(1);
+    expect(c._id).toBe(2);
+  });
+
+  it("drops the oldest entry once maxLogs is exceeded (ring buffer)", () => {
+    const store = new ProjectLogsStore(3);
+    store.addLog({ message: "a" });
+    store.addLog({ message: "b" });
+    store.addLog({ message: "c" });
+    store.addLog({ message: "d" });
+    const messages = store.getAll().map((e) => e.message);
+    expect(messages).toEqual(["b", "c", "d"]);
+  });
+
+  it("continues to assign fresh _ids even after buffer wraps", () => {
+    const store = new ProjectLogsStore(2);
+    store.addLog({ message: "a" });
+    store.addLog({ message: "b" });
+    const third = store.addLog({ message: "c" });
+    expect(third._id).toBe(2);
+    expect(store.getAll().map((e) => e._id)).toEqual([1, 2]);
+  });
+
+  it("filters by level", () => {
+    const store = new ProjectLogsStore(100);
+    store.addLog({ message: "a", level: V1LogLevel.LOG_LEVEL_INFO });
+    store.addLog({ message: "b", level: V1LogLevel.LOG_LEVEL_ERROR });
+    store.addLog({ message: "c", level: V1LogLevel.LOG_LEVEL_WARN });
+
+    const errors = store.getFiltered({
+      levels: [V1LogLevel.LOG_LEVEL_ERROR],
+    });
+    expect(errors.map((e) => e.message)).toEqual(["b"]);
+  });
+
+  it("search matches message and jsonPayload case-insensitively", () => {
+    const store = new ProjectLogsStore(100);
+    store.addLog({ message: "Fooooo", jsonPayload: "{}" });
+    store.addLog({ message: "bar", jsonPayload: '{"reason":"foobar"}' });
+    store.addLog({ message: "baz", jsonPayload: "{}" });
+
+    const hits = store.getFiltered({ search: "FOO" });
+    expect(hits.map((e) => e.message)).toEqual(["Fooooo", "bar"]);
+  });
+
+  it("combines level and search as an intersection", () => {
+    const store = new ProjectLogsStore(100);
+    store.addLog({
+      message: "boot",
+      level: V1LogLevel.LOG_LEVEL_ERROR,
+    });
+    store.addLog({
+      message: "boot",
+      level: V1LogLevel.LOG_LEVEL_INFO,
+    });
+    store.addLog({
+      message: "crash",
+      level: V1LogLevel.LOG_LEVEL_ERROR,
+    });
+
+    const hits = store.getFiltered({
+      levels: [V1LogLevel.LOG_LEVEL_ERROR],
+      search: "boot",
+    });
+    expect(hits.map((e) => e.message)).toEqual(["boot"]);
+  });
+
+  it("returns everything when filters are empty", () => {
+    const store = new ProjectLogsStore(100);
+    store.addLog({ message: "a" });
+    store.addLog({ message: "b" });
+    expect(store.getFiltered({}).map((e) => e.message)).toEqual(["a", "b"]);
+    expect(store.getFiltered().map((e) => e.message)).toEqual(["a", "b"]);
+  });
+});

web-admin/src/features/projects/status/logs/project-logs-store.ts

@@ -0,0 +1,53 @@
+import type { V1Log } from "@rilldata/web-common/runtime-client";
+
+export type LogEntry = V1Log & { _id: number };
+
+export interface LogFilters {
+  levels?: string[];
+  search?: string;
+}
+
+/**
+ * In-memory log buffer for ProjectLogsPage. Keeps the state machine out of
+ * the Svelte component so the filtering + ring-buffer behavior can be
+ * tested without a DOM.
+ */
+export class ProjectLogsStore {
+  private readonly entries: LogEntry[] = [];
+  private nextId = 0;
+
+  constructor(public readonly maxLogs: number) {}
+
+  public addLog(log: V1Log): LogEntry {
+    const entry: LogEntry = { ...log, _id: this.nextId++ };
+    this.entries.push(entry);
+    if (this.entries.length > this.maxLogs) {
+      this.entries.shift();
+    }
+    return entry;
+  }
+
+  public getAll(): LogEntry[] {
+    return [...this.entries];
+  }
+
+  public getFiltered(filters: LogFilters = {}): LogEntry[] {
+    const { levels = [], search = "" } = filters;
+    const needle = search.toLowerCase();
+    return this.entries.filter((log) => {
+      const matchesLevel =
+        levels.length === 0 || levels.includes(log.level ?? "");
+      if (!matchesLevel) return false;
+
+      if (!needle) return true;
+      return (
+        (log.message?.toLowerCase().includes(needle) ?? false) ||
+        (log.jsonPayload?.toLowerCase().includes(needle) ?? false)
+      );
+    });
+  }
+
+  public clear(): void {
+    this.entries.length = 0;
+  }
+}

web-admin/src/routes/[organization]/[project]/-/edit/+layout.svelte

@@ -3,9 +3,11 @@
   import {
     createAdminServiceGetCurrentUser,
     createAdminServiceGetProject,
+    getAdminServiceGetProjectQueryKey,
     V1DeploymentStatus,
     type V1Organization,
   } from "@rilldata/web-admin/client";
+  import { useQueryClient } from "@tanstack/svelte-query";
   import {
     branchPathPrefix,
     extractBranchFromPath,
@@ -121,6 +123,20 @@
   $: projectUrl = `/${organization}/${project}`;
   $: branchUrl = `/${organization}/${project}${branchPathPrefix(branch)}`;
 
+  // Invalidating this query refetches a fresh JWT; `runtimeClient.getJwt()`
+  // reads the updated value on the next call. Branch must be part of the
+  // key or the invalidation misses the branch-scoped cache entry.
+  const queryClient = useQueryClient();
+  $: onBeforeReconnect = async () => {
+    await queryClient.invalidateQueries({
+      queryKey: getAdminServiceGetProjectQueryKey(
+        organization,
+        project,
+        branch ? { branch } : undefined,
+      ),
+    });
+  };
+
   onDestroy(() => {
     $editorRoutePrefix = "";
   });
@@ -172,7 +188,8 @@
         <FileAndResourceWatcher
           host={runtimeHost}
           {instanceId}
-          keepAlive
+          lifecycle="none"
+          {onBeforeReconnect}
           errorBody="Lost connection to the editing environment. Try ending the session and starting a new one."
         >
           <div class="flex flex-1 overflow-hidden">