ACM-44888: fix non-admin SSE OOM on release-2.16

backend/src/lib/compression.ts

@@ -138,7 +138,7 @@ export class FifoSet<T> {
 const bigStrings: FifoSet<string> = new FifoSet(200)
 
 export async function deflateResource(resource: IResource, dictionary: Dictionary): Promise<Buffer> {
-  const res = compressResource(resource as UncompressedResourceType, dictionary)
+  const res = compressResource(resource, dictionary)
   let buffer
   try {
     buffer = await promisify(deflateRaw)(JSON.stringify(res))
@@ -243,11 +243,18 @@ export async function inflateResource(buffer: Buffer, dictionary: Dictionary): P
 }
 
 export async function inflateEvent(event: ServerSideEvent): Promise<ServerSideEvent> {
-  const { id, data } = event
-  const { type, object } = data as WatchEvent
+  const { id, name, namespace, data } = event
+  if (!data || typeof data !== 'object') return event
+  const watchEvent = data as WatchEvent & { meta?: unknown }
+  const { type, object } = watchEvent
   return !object
     ? event
-    : { id, data: { type, object: Buffer.isBuffer(object) ? await inflateResource(object, getEventDict()) : object } }
+    : {
+        id,
+        name,
+        namespace,
+        data: { type, object: Buffer.isBuffer(object) ? await inflateResource(object, getEventDict()) : object },
+      }
 }
 
 export async function inflateApps(apps: ICompressedResource[]): Promise<ITransformedResource[]> {

backend/src/lib/server-side-events.ts

@@ -4,7 +4,6 @@ import { constants } from 'node:http2'
 import type { Transform } from 'node:stream'
 import { clearInterval } from 'node:timers'
 import type { Zlib } from 'node:zlib'
-import { batchPromiseAll } from './batch-promise-all'
 import { getEncodeStream, inflateEvent } from './compression'
 import { setCookie } from './cookies'
 import { logger } from './logger'
@@ -17,7 +16,7 @@ import { sizeOf } from '../routes/aggregators/utils'
 
 // If a client hasn't finished receiving a broadcast in PURGE_CLIENT_TIMEOUT
 // assume the browser has been refreshed or closed
-const PURGE_CLIENT_TIMEOUT = 4 * 60 * 60 * 1000
+const PURGE_CLIENT_TIMEOUT = 30 * 60 * 1000
 
 const instanceID = randomString(8)
 
@@ -35,6 +34,15 @@ export interface ServerSideEvent<DataT = unknown> {
   namespace?: string
   data?: DataT
 }
+
+/** Lightweight resource identity for RBAC filtering without inflating compressed objects. */
+export interface EventResourceMeta {
+  kind: string
+  apiVersion: string
+  name?: string
+  namespace?: string
+}
+
 export interface WatchEvent {
   type: 'ADDED' | 'DELETED' | 'MODIFIED' | 'EOP'
   object: {
@@ -46,6 +54,24 @@ export interface WatchEvent {
       resourceVersion: string
     }
   }
+  meta?: EventResourceMeta
+}
+
+/** Resolve kind/apiVersion/name/namespace from meta or an already-inflated object. */
+export function getEventResourceMeta(event: ServerSideEvent): EventResourceMeta | undefined {
+  const data = event.data as (WatchEvent & { type?: string }) | undefined
+  if (!data || typeof data !== 'object') return undefined
+  if (data.meta?.kind) return data.meta
+  const object = data.object as WatchEvent['object'] | Buffer | undefined
+  if (object && !Buffer.isBuffer(object) && typeof object === 'object' && object.kind) {
+    return {
+      kind: object.kind,
+      apiVersion: object.apiVersion,
+      name: object.metadata?.name,
+      namespace: object.metadata?.namespace,
+    }
+  }
+  return undefined
 }
 
 export interface ServerSideEventClient {
@@ -125,22 +151,26 @@ export class ServerSideEvents {
     }
   }
 
-  private static async sendEvent(clientID: string, event: ServerSideEvent): Promise<void> {
+  private static sendEvent(clientID: string, event: ServerSideEvent): Promise<void> {
     const client = this.clients[clientID]
-    if (!client) return
-    if (client.events && !client.events[event.name]) return
-    if (client.namespaces && !client.namespaces[event.namespace]) return
-    event = await inflateEvent(event)
+    if (!client) return Promise.resolve()
+    if (client.events && !client.events[event.name]) return Promise.resolve()
+    if (client.namespaces && !client.namespaces[event.namespace]) return Promise.resolve()
+    // Filter before inflate so denied events never materialize full resource JSON in memory.
     if (this.eventFilter) {
       client.eventQueue.push(
         this.eventFilter(client.token, event)
-          .then((shouldSendEvent) => (shouldSendEvent ? event : undefined))
+          .then((shouldSendEvent) => {
+            if (!shouldSendEvent) return undefined
+            return inflateEvent(event)
+          })
           .catch((): undefined => undefined)
       )
     } else {
-      client.eventQueue.push(Promise.resolve(event))
+      client.eventQueue.push(inflateEvent(event))
     }
     void this.processClient(clientID)
+    return Promise.resolve()
   }
 
   private static async processClient(clientID: string): Promise<void> {
@@ -243,8 +273,8 @@ export class ServerSideEvents {
     res: Http2ServerResponse
   ): Promise<ServerSideEventClient> {
     const [writableStream, compressionStream, encoding] = getEncodeStream(
-      res as unknown as NodeJS.WritableStream,
-      req.headers[HTTP2_HEADER_ACCEPT_ENCODING] as string,
+      res,
+      req.headers[HTTP2_HEADER_ACCEPT_ENCODING],
       process.env.DISABLE_STREAM_COMPRESSION === 'true'
     )
 
@@ -311,10 +341,10 @@ export class ServerSideEvents {
 
     // SORT EVENTS INTO SMALLER PACKETS
     // SO THAT BROWSER PAGE LOADS QUICKER
-    // uncompress and split events into packets
+    // Classify using meta / inflated object identity — do not inflate the whole cache up front.
     const values = Object.values(this.events)
     const compressed = sizeOf(values)
-    let parts = await batchPromiseAll(values, (event) => inflateEvent(event))
+    let parts: ServerSideEvent[] = [...values]
 
     // mock a large environment
     if (process.env.MOCK_CLUSTERS) {
@@ -343,9 +373,9 @@ export class ServerSideEvents {
     const other: ServerSideEvent<unknown>[] = []
     const remainder: ServerSideEvent<unknown>[] = []
     parts.forEach((event) => {
-      const data = event.data as WatchEvent
+      const meta = getEventResourceMeta(event)
       // see frontend/src/components/LoadPluginData.tsx for what pages are fast loaded
-      switch (data.object.kind) {
+      switch (meta?.kind) {
         case 'ManagedCluster':
         case 'HostedCluster':
         case 'ClusterDeployment':
@@ -384,9 +414,9 @@ export class ServerSideEvents {
     // sort events alphabetically so that browser list fills from top to bottom
     const compareFn =
       (propName: 'name' | 'namespace') => (a: ServerSideEvent<unknown>, b: ServerSideEvent<unknown>) => {
-        const adata = a.data as WatchEvent
-        const bdata = b.data as WatchEvent
-        return adata.object.metadata[propName].localeCompare(bdata.object.metadata[propName])
+        const aVal = getEventResourceMeta(a)?.[propName] ?? ''
+        const bVal = getEventResourceMeta(b)?.[propName] ?? ''
+        return aVal.localeCompare(bVal)
       }
     clusters.sort(compareFn('name'))
     infos.sort(compareFn('namespace'))

backend/src/resources/watch-options.ts

@@ -7,4 +7,10 @@ export interface IWatchOptions {
   // poll the resource list instead of watching it
   // process the items in its own cache so not to overload event cache
   isPolled?: boolean
+  /**
+   * True when the Kubernetes resource is cluster-scoped.
+   * Used by SSE RBAC to decide whether SelfSubjectRulesReview should probe `default`
+   * (cluster-scoped) or the resource namespace (namespaced).
+   */
+  clusterScoped?: boolean
 }