console: keep the redesigned cluster page's live data in TanStack DB collections

console/doc/design/20260902_sync_engine.md

@@ -69,7 +69,7 @@ count stays the same, and a collection can even be fed from an existing atom
 | `SubscribeManager` | `api/materialize/SubscribeManager.ts` | The SUBSCRIBE protocol: buffers rows per timestamp, folds the upsert stream into the current row set, exposes snapshots via `onChange`/`getSnapshot`. Holds its last snapshot through a resubscribe so the UI does not flash empty. |
 | `WebsocketConnectionManager` | `api/materialize/WebsocketConnectionManager.ts` | Connection lifecycle: subscribes to environment health and the current region on the jotai store, connects when healthy, reconnects with backoff, reconnects when the region's address changes, and tears down a socket left pointing at a departed region. |
 | Subscribe sessions | `api/materialize/subscribeSession.ts` | Composition: one constructor wires the manager, the connection manager, the sink, the region reset, and cache hydration, in one defined order. |
-| Hooks | `api/materialize/useSubscribe.ts` | Lifecycle adapters: a single effect creates the session on mount and destroys it on unmount. |
+| Hooks | `api/materialize/useSubscribe.ts`, `useScopedSubscribeCollection.ts` | Lifecycle adapters: a single effect creates the session on mount and destroys it on unmount. |
 | Collection bridge | `api/materialize/subscribeCollection.ts` | `createSubscribeCollection`: turns manager snapshots into incremental insert/update/delete writes on a TanStack DB collection, owns the scoped localStorage cache, and mirrors status (error, snapshotComplete) into a companion atom. |
 | Cache scope | `store/syncEngineCache.ts` | Derives the cache scope `organizationId\|regionId` as an atom, so hydration can be driven from the store. |
 
@@ -176,11 +176,13 @@ What the engine is not:
   sinks are transport-agnostic, which is the seam a shared upstream
   (cross-tab or cross-client dedup) can plug into without touching surfaces,
   but the engine alone does not provide that.
-- **Not for polled telemetry.** Expensive introspection queries
-  (utilization history, arrangement sizes) are react-query territory, with
-  polling intervals and its persistence tooling. The engine serves
-  catalog-shaped data that a SUBSCRIBE can maintain.
-- **Not for page-scoped, request-driven subscribes.** Those use
+- **Not for polled telemetry.** Introspection data no SUBSCRIBE maintains,
+  such as utilization beyond 24h or on environments that predate the indexed
+  views, is react-query territory, with polling intervals and its persistence
+  tooling. The engine serves data that a SUBSCRIBE can maintain, including
+  per-key page data (see per-key scoped collections below).
+- **Not for one-off, request-driven subscribes.** Subscribes whose rows
+  nothing reads after the component unmounts use
   `useSubscribe`/`useSubscribeManager`, which tie a manager to a component
   and expose state via `useSyncExternalStore`.
 
@@ -194,9 +196,9 @@ queries over the objects collection (fed from the app-wide `allObjects` atom)
 and the namespaces collection (its own subscribe), and renders from cache on
 revisit before the live snapshot lands.
 
-Declare configuration once at module scope. The hooks require a referentially
-stable options object, and a new object identity restarts the session, which
-makes accidental instability visible instead of masked.
+Declare configuration once at module scope, or memoize it. The hooks require
+a referentially stable options object, and a new object identity restarts the
+session, which makes accidental instability visible instead of masked.
 
 Atom-backed data set:
 
@@ -278,6 +280,33 @@ component keyed by a path pattern survives param-only navigation, including
 region switches, which is why region handling lives in the session rather
 than in mount/unmount.
 
+## Per-key scoped collections
+
+Page data keyed by an id, such as a cluster's utilization or a replica's
+arrangement sizes, lives in `createScopedSubscribeCollections`: one in-memory
+collection per tenant scope and key, dropped when the scope changes and never
+persisted. `useScopedSubscribeCollection` resolves the scope, picks the key's
+collection and runs the subscribe into it while mounted. The rows outlive the
+component, so a return to the page serves them at once, with `isRefreshing`
+set until the new subscribe's snapshot lands, and a view derived from them,
+such as a chart window, can change without resubscribing.
+
+```ts
+const samplesByCluster = createScopedSubscribeCollections<Sample>({
+  id: "samples",
+  getKey: (sample) => sample.id,
+});
+
+export function useLiveSamples(clusterId: string) {
+  // A new builder restarts the subscribe, so memoize it.
+  const subscribe = React.useMemo(
+    () => buildSubscribeQuery(buildSamplesQuery(clusterId), { upsertKey: "id" }),
+    [clusterId],
+  );
+  return useScopedSubscribeCollection(samplesByCluster, clusterId, subscribe);
+}
+```
+
 ## Testing
 
 Sessions and the collection bridge are plain objects, so their behavior is

console/src/api/materialize/cluster/largestClusterReplica.ts

@@ -40,6 +40,7 @@ export function buildLargestClusterReplicaQuery(
     )
     .leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size")
     .select([
+      "cr.id",
       "cr.name",
       "cr.size",
       sql<string | null>`${replicaHeapLimit}::text`.as("heapLimit"),

console/src/api/materialize/cluster/largestMaintainedQueries.ts

@@ -15,6 +15,7 @@ import {
   executeSqlV2,
   queryBuilder,
 } from "~/api/materialize";
+import { buildSubscribeQuery } from "~/api/materialize/buildSubscribeQuery";
 
 export type LargestMaintainedQueriesParams = {
   replicaHeapLimit: number | null;
@@ -164,6 +165,31 @@ export function buildLargestMaintainedObjectSizesQuery({
   );
 }
 
+/** One object's arrangement size on a replica, in bytes. */
+export interface ReplicaObjectSize {
+  objectId: string;
+  size: number | null;
+}
+
+/**
+ * Live arrangement sizes of every object on a replica. Filtering on the
+ * literal replica id makes this a lookup on `mz_object_arrangement_sizes_ind`.
+ */
+export function buildReplicaObjectSizesSubscribe(replicaId: string) {
+  return buildSubscribeQuery<ReplicaObjectSize>(
+    queryBuilder
+      .selectFrom("mz_object_arrangement_sizes")
+      .select([
+        "object_id as objectId",
+        // NOTE: The cast is required. An int8 decodes to a BigInt, which the
+        // collection bridge's JSON.stringify row signature throws on.
+        sql<number | null>`size::float8`.as("size"),
+      ])
+      .where("replica_id", "=", replicaId),
+    { upsertKey: "objectId" },
+  );
+}
+
 export async function fetchLargestMaintainedObjectSizes({
   params,
   queryKey,

console/src/api/materialize/cluster/replicaUtilizationBinning.ts

@@ -477,3 +477,17 @@ export const hasUtilizationReading = (point: DataPoint) =>
 export const hasNoUtilizationMetrics = (graphData: ReplicaData[]) =>
   graphData.length > 0 &&
   graphData.every(({ data }) => !data.some(hasUtilizationReading));
+
+/**
+ * Server-binned rows inside a window that starts at `startDate`, in time order.
+ * Held rows from a wider window would otherwise stretch the chart past the
+ * selected range, and ENVELOPE UPSERT yields them unordered.
+ */
+export function clipBinnedRowsToWindow<T extends { bucketStart: Date }>(
+  rows: T[],
+  startDate: Date,
+): T[] {
+  return rows
+    .filter((row) => row.bucketStart.getTime() >= startDate.getTime())
+    .sort((a, b) => a.bucketStart.getTime() - b.bucketStart.getTime());
+}

console/src/api/materialize/subscribeCollection.test.ts

@@ -10,6 +10,7 @@
 import { getStore } from "~/jotai";
 
 import {
+  createScopedSubscribeCollections,
   createSubscribeCollection,
   syncEngineCacheKey,
 } from "./subscribeCollection";
@@ -257,6 +258,39 @@ describe("createSubscribeCollection", () => {
     expect(getStore().get(statusAtom).snapshotComplete).toBe(false);
   });
 
+  it("marks held rows as refreshing until a new subscribe's snapshot lands", () => {
+    const { applySnapshot, statusAtom } = freshCollection();
+    applySnapshot(state([{ id: "1", name: "a" }]));
+    expect(getStore().get(statusAtom).isRefreshing).toBe(false);
+
+    // A new subscribe opens with an empty pre-snapshot.
+    applySnapshot(state([], false));
+    expect(getStore().get(statusAtom)).toMatchObject({
+      snapshotComplete: true,
+      isRefreshing: true,
+    });
+
+    applySnapshot(state([{ id: "1", name: "b" }]));
+    expect(getStore().get(statusAtom).isRefreshing).toBe(false);
+  });
+
+  it("marks the rows a reopened socket holds as refreshing", async () => {
+    const { collection, applySnapshot, statusAtom } = freshCollection();
+    applySnapshot(state([{ id: "1", name: "a" }]));
+
+    // A reopened socket re-emits its held rows until the new snapshot lands.
+    applySnapshot({ ...state([{ id: "1", name: "a" }]), resubscribing: true });
+    await flush();
+    expect(collection.get("1")?.name).toBe("a");
+    expect(getStore().get(statusAtom)).toMatchObject({
+      snapshotComplete: true,
+      isRefreshing: true,
+    });
+
+    applySnapshot(state([{ id: "1", name: "b" }]));
+    expect(getStore().get(statusAtom).isRefreshing).toBe(false);
+  });
+
   it("surfaces error and snapshotComplete on the status atom", () => {
     const { applySnapshot, statusAtom } = freshCollection();
 
@@ -302,3 +336,18 @@ describe("createSubscribeCollection", () => {
     }
   });
 });
+
+describe("createScopedSubscribeCollections", () => {
+  it("keeps one collection per scope and key, dropping other scopes' on a scope change", () => {
+    const collections = createScopedSubscribeCollections<Row>({
+      id: "scoped-test",
+      getKey: (row) => row.id,
+    });
+    const first = collections.get("org|regionA", "u1");
+    expect(collections.get("org|regionA", "u1")).toBe(first);
+    expect(collections.get("org|regionA", "u2")).not.toBe(first);
+
+    expect(collections.get("org|regionB", "u1")).not.toBe(first);
+    expect(collections.get("org|regionA", "u1")).not.toBe(first);
+  });
+});

console/src/api/materialize/subscribeCollection.ts

@@ -57,8 +57,21 @@ function writePersistedRows<T>(key: string, rows: T[]) {
 export interface SubscribeCollectionStatus {
   error: SubscribeError | undefined;
   snapshotComplete: boolean;
+  /** Rows from an earlier snapshot are served, and no running subscribe has
+   * delivered a newer one yet. */
+  isRefreshing: boolean;
 }
 
+const IDLE_STATUS: SubscribeCollectionStatus = {
+  error: undefined,
+  snapshotComplete: false,
+  isRefreshing: false,
+};
+
+/** The status of a collection that nothing has fed, for a consumer whose
+ * collection is not known yet. */
+export const idleCollectionStatusAtom = atom(() => IDLE_STATUS);
+
 export interface SubscribeCollection<T extends object> {
   collection: Collection<T, string>;
   statusAtom: PrimitiveAtom<SubscribeCollectionStatus>;
@@ -103,10 +116,7 @@ export function createSubscribeCollection<T extends object>(options: {
   // `hydrate`, then kept current by `applySnapshot`.
   let desired = new Map<string, T>();
 
-  const statusAtom = atom<SubscribeCollectionStatus>({
-    error: undefined,
-    snapshotComplete: false,
-  });
+  const statusAtom = atom<SubscribeCollectionStatus>(IDLE_STATUS);
 
   // Captured when the collection starts syncing; cleared when it pauses.
   type WriteApi = Parameters<
@@ -214,7 +224,8 @@ export function createSubscribeCollection<T extends object>(options: {
     const current = store.get(statusAtom);
     if (
       current.error !== next.error ||
-      current.snapshotComplete !== next.snapshotComplete
+      current.snapshotComplete !== next.snapshotComplete ||
+      current.isRefreshing !== next.isRefreshing
     ) {
       store.set(statusAtom, next);
     }
@@ -227,7 +238,11 @@ export function createSubscribeCollection<T extends object>(options: {
     const isEmptyPreload =
       !state.snapshotComplete && state.data.length === 0 && !state.error;
     if (isEmptyPreload) {
-      writeStatus({ error: undefined, snapshotComplete: lastSnapshotComplete });
+      writeStatus({
+        error: undefined,
+        snapshotComplete: lastSnapshotComplete,
+        isRefreshing: lastSnapshotComplete,
+      });
       return;
     }
 
@@ -240,6 +255,7 @@ export function createSubscribeCollection<T extends object>(options: {
     writeStatus({
       error: state.error,
       snapshotComplete: state.snapshotComplete,
+      isRefreshing: Boolean(state.resubscribing),
     });
   };
 
@@ -296,7 +312,12 @@ export function createSubscribeCollection<T extends object>(options: {
     // On a scope change this also flushes the emptied set, so the collection
     // stops serving the previous scope's rows while the new snapshot loads.
     materialize();
-    writeStatus({ error: undefined, snapshotComplete: lastSnapshotComplete });
+    // Cached rows stand in until the live snapshot arrives.
+    writeStatus({
+      error: undefined,
+      snapshotComplete: lastSnapshotComplete,
+      isRefreshing: lastSnapshotComplete,
+    });
   };
 
   const clearPendingPersist = () => {
@@ -312,7 +333,11 @@ export function createSubscribeCollection<T extends object>(options: {
     lastSnapshotComplete = false;
     liveSnapshotApplied = false;
     materialize();
-    writeStatus({ error: undefined, snapshotComplete: false });
+    writeStatus({
+      error: undefined,
+      snapshotComplete: false,
+      isRefreshing: false,
+    });
   };
 
   const suspendPersistence = () => {
@@ -332,3 +357,46 @@ export function createSubscribeCollection<T extends object>(options: {
     suspendPersistence,
   };
 }
+
+export interface ScopedSubscribeCollections<T extends object> {
+  /** The collection for `key` under `scope`, created on first use. */
+  get: (scope: string, key: string) => SubscribeCollection<T>;
+  getKey: (row: T) => string;
+}
+
+/**
+ * In-memory collections, one per tenant scope (`organizationId|regionId`) and
+ * key, such as a cluster id. Keys repeat across environments, so a collection
+ * never serves another scope's rows. Rows stay in memory until the scope
+ * changes, which makes returning to a page instant. They are not persisted,
+ * since an entry per key would have no bound in localStorage.
+ */
+export function createScopedSubscribeCollections<T extends object>(options: {
+  /** Prefix of each collection's id. */
+  id: string;
+  getKey: (row: T) => string;
+}): ScopedSubscribeCollections<T> {
+  const { id, getKey } = options;
+  const byKey = new Map<string, SubscribeCollection<T>>();
+  let currentScope: string | undefined;
+  return {
+    get: (scope, key) => {
+      if (scope !== currentScope) {
+        // A return to an earlier scope gets fresh collections, so the dropped
+        // ones are never served again.
+        byKey.clear();
+        currentScope = scope;
+      }
+      let collection = byKey.get(key);
+      if (!collection) {
+        collection = createSubscribeCollection<T>({
+          id: `${id}|${scope}|${key}`,
+          getKey,
+        });
+        byKey.set(key, collection);
+      }
+      return collection;
+    },
+    getKey,
+  };
+}

console/src/api/materialize/subscribeSession.ts

@@ -185,6 +185,14 @@ export function createCollectionSubscribeSession<
       unsubscribeScope?.();
       keepAlive.unsubscribe();
       session.destroy();
+      // No subscribe backs the held rows any more, so they read as refreshing
+      // until a later session's snapshot lands, as behind a new subscribe's
+      // empty pre-snapshot.
+      target.applySnapshot({
+        data: [],
+        snapshotComplete: false,
+        error: undefined,
+      });
     },
   };
 }