ACM-44890: fix non-admin SSE OOM on release-2.14

backend/src/lib/compression.ts

@@ -1,23 +1,23 @@
 /* Copyright Contributors to the Open Cluster Management project */
 import type { Readable, Transform } from 'node:stream'
 import { pipeline } from 'node:stream'
+import { promisify } from 'node:util'
+import type { Zlib } from 'node:zlib'
 import {
   createBrotliCompress,
   createBrotliDecompress,
   createDeflate,
   createGunzip,
   createGzip,
   createInflate,
-  inflateRaw,
   deflateRaw,
-  type Zlib,
+  inflateRaw,
 } from 'node:zlib'
-import { logger } from './logger'
-import type { ServerSideEvent, WatchEvent } from './server-side-events'
-import { getEventDict } from '../routes/events'
 import { getAppDict, type ICompressedResource, type ITransformedResource } from '../routes/aggregators/applications'
-import { promisify } from 'node:util'
+import { getEventDict } from '../routes/events'
 import type { IResource } from './../resources/resource'
+import { logger } from './logger'
+import type { ServerSideEvent, WatchEvent } from './server-side-events'
 
 const MAX_RECENTLY_ADDED = 200
 
@@ -26,6 +26,7 @@ type Dictionary = {
   map: Record<string, string>
   add: (key: string) => string
   get: (inx: number) => string
+  has: (key: string) => string
   recentlyAdded: string[]
   snapshotSize: () => number
   drainRecentlyAdded: () => string[]
@@ -48,13 +49,17 @@ export function createDictionary(): Dictionary {
   const get = (inx: number) => {
     return arr[inx]
   }
+  const has = (key: string) => {
+    return map[key]
+  }
   const snapshotSize = () => arr.length
   const drainRecentlyAdded = () => recentlyAdded.splice(0)
   return {
     arr,
     map,
     add,
     get,
+    has,
     recentlyAdded,
     snapshotSize,
     drainRecentlyAdded,
@@ -81,6 +86,7 @@ type UncompressedResourceType = Record<string, any> | Record<string, any[]> | st
 type CompressedResourceType = Record<number, any> | Record<number, any[]> | string | number
 
 const NUMBER_MARKER = '#!%'
+const JSON_MARKER = '#!&'
 
 // Detects ISO 8601 timestamps to avoid permanently indexing unique time values.
 // Covers: "2026-05-27T20:18:12Z" (20), "2026-05-27T20:18:12.000Z" (24), "2026-05-27T20:18:12+05:30" (25)
@@ -95,8 +101,44 @@ export function isTimestamp(s: string): boolean {
   )
 }
 
+export class FifoSet<T> {
+  private readonly values: T[] = []
+  private readonly membership: Set<T> = new Set()
+  private readonly capacity?: number
+
+  constructor(capacity?: number) {
+    this.capacity = capacity
+  }
+
+  has(value: T): boolean {
+    return this.membership.has(value)
+  }
+
+  add(value: T): void {
+    if (!this.membership.has(value)) {
+      this.values.push(value)
+      this.membership.add(value)
+
+      if (this.capacity !== undefined && this.values.length > this.capacity) {
+        const evicted = this.values.shift()
+        if (evicted !== undefined) this.membership.delete(evicted)
+      }
+    }
+  }
+
+  delete(value: T): void {
+    if (this.membership.has(value)) {
+      this.membership.delete(value)
+      const index = this.values.indexOf(value)
+      if (index >= 0) this.values.splice(index, 1)
+    }
+  }
+}
+
+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))
@@ -113,7 +155,6 @@ export async function deflateResource(resource: IResource, dictionary: Dictionar
 function compressResource(resource: UncompressedResourceType, dictionary: Dictionary): CompressedResourceType {
   if (resource) {
     if (Array.isArray(resource)) {
-      // eslint-disable-next-line @typescript-eslint/no-unsafe-return, @typescript-eslint/no-unsafe-call
       return resource.map((item: UncompressedResourceType) => compressResource(item, dictionary))
     } else if (typeof resource === 'object') {
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -134,8 +175,9 @@ function compressResource(resource: UncompressedResourceType, dictionary: Dictio
             res[dictionary.add(key)] = resource[key]
           } else {
             const inx = dictionary.add(key)
-            if (valueInDictionaryKeys.has(key)) {
-              res[inx] = dictionary.add(resource[key] as string)
+            // Guard against non-string values (e.g. nested CRD OpenAPI schema objects) corrupting the shared dictionary.
+            if (valueInDictionaryKeys.has(key) && typeof resource[key] === 'string') {
+              res[inx] = dictionary.add(resource[key])
             } else {
               res[inx] = compressResource(resource[key] as UncompressedResourceType, dictionary)
             }
@@ -144,14 +186,39 @@ function compressResource(resource: UncompressedResourceType, dictionary: Dictio
       }
       return res
     } else if (typeof resource === 'string') {
-      if (resource.length < 32 && !resource.endsWith('==')) {
+      if (
+        (resource.length > 128 && resource.startsWith('{') && !resource.startsWith('{{')) ||
+        resource.startsWith('[')
+      ) {
+        // if the resource is a large json string, compress the inner json
+        try {
+          const innerJson = JSON.parse(resource) as UncompressedResourceType
+          return `${JSON_MARKER}${JSON.stringify(compressResource(innerJson, dictionary))}`
+        } catch (error) {
+          // drop thru
+        }
+      }
+      if (resource.length < 32 && !resource.endsWith('=')) {
         // skip indexing of all timestamps
         if (isTimestamp(resource)) {
           return resource
         }
         // index short strings that aren't a base64
         return dictionary.add(resource)
       }
+      // if already in dictionary, return the index
+      const exists = dictionary.has(resource)
+      if (exists) {
+        return exists
+      }
+      // if the string is not in the dictionary, add it to the bigStrings set
+      if (!bigStrings.has(resource)) {
+        bigStrings.add(resource)
+      } else {
+        // if we've seen this string, add to the dictionary
+        bigStrings.delete(resource)
+        return dictionary.add(resource)
+      }
     } else if (typeof resource === 'number' && Number.isInteger(resource)) {
       // to differentiate between an index and a value that is actually a number
       return `${NUMBER_MARKER}${resource}`
@@ -163,7 +230,7 @@ function compressResource(resource: UncompressedResourceType, dictionary: Dictio
 export async function inflateResource(buffer: Buffer, dictionary: Dictionary): Promise<IResource> {
   let inflated
   try {
-    inflated = (await promisify(inflateRaw)(buffer)).toString()
+    inflated = (await promisify(inflateRaw)(new Uint8Array(buffer))).toString()
   } catch (err: unknown) {
     logger.error({
       msg: 'Error from inflateRaw during inflateResource',
@@ -176,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[]> {
@@ -211,6 +285,8 @@ function decompressResource(resource: CompressedResourceType, dictionary: Dictio
       for (const inx in resource) {
         if (Object.prototype.hasOwnProperty.call(resource, inx)) {
           const key = dictionary.get(Number(inx))
+          // Dictionary corruption would produce a non-string key; skip rather than crashing on key.includes().
+          if (typeof key !== 'string') continue
           if (
             valueAsIsKeys.has(key) ||
             (key === 'message' && inx in resource && !Number.isInteger(Number(resource[inx]))) ||
@@ -230,8 +306,14 @@ function decompressResource(resource: CompressedResourceType, dictionary: Dictio
       return res
     } else if (Number.isInteger(Number(resource))) {
       return dictionary.get(Number(resource))
-    } else if (typeof resource === 'string' && resource.startsWith(NUMBER_MARKER)) {
-      return Number(resource.substring(NUMBER_MARKER.length))
+    } else if (typeof resource === 'string') {
+      if (resource.startsWith(NUMBER_MARKER)) {
+        return Number(resource.substring(NUMBER_MARKER.length))
+      }
+      if (resource.startsWith(JSON_MARKER)) {
+        const innerJson = JSON.parse(resource.substring(JSON_MARKER.length)) as CompressedResourceType
+        return JSON.stringify(decompressResource(innerJson, dictionary))
+      }
     }
   }
   return resource

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

@@ -16,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)
 
@@ -34,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: {
@@ -45,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 {
@@ -124,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))
-          .catch(() => undefined) as Promise<ServerSideEvent<unknown>>
+          .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> {
@@ -242,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'
     )
 
@@ -310,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 Promise.all(values.map((event) => inflateEvent(event)))
+    let parts: ServerSideEvent[] = [...values]
 
     // mock a large environment
     if (process.env.MOCK_CLUSTERS) {
@@ -341,9 +372,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':
@@ -377,9 +408,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
 }