fix(network): bound configuration cache and handle nested TLS sockets

Author: waleedlatif1

apps/sim/lib/core/config/appconfig.test.ts

@@ -133,6 +133,48 @@ describe('fetchAppConfigProfile', () => {
     expect(b).toEqual({ x: 1 })
     expect(mockSend.mock.calls.map(([c]) => c.__type)).toEqual(['start', 'get'])
   })
+
+  it('honors the server poll interval and serves warm values during one shared refresh', async () => {
+    vi.useFakeTimers()
+    vi.setSystemTime(100_000)
+    try {
+      mockSend.mockImplementation((command: { __type: string }) =>
+        Promise.resolve(
+          command.__type === 'start'
+            ? { InitialConfigurationToken: 'token' }
+            : {
+                Configuration: encode({ revision: 'first' }),
+                NextPollConfigurationToken: 'next',
+                NextPollIntervalInSeconds: 60,
+              }
+        )
+      )
+      const ids = uniqueIds()
+      const parse = (value: unknown) => value
+      expect(await fetchAppConfigProfile(ids, parse)).toEqual({ revision: 'first' })
+      vi.setSystemTime(130_001)
+      await fetchAppConfigProfile(ids, parse)
+      expect(mockSend).toHaveBeenCalledTimes(2)
+
+      let finish: (value: unknown) => void = () => {}
+      mockSend.mockReturnValueOnce(
+        new Promise((resolve) => {
+          finish = resolve
+        })
+      )
+      vi.setSystemTime(161_000)
+      expect(
+        await Promise.all([fetchAppConfigProfile(ids, parse), fetchAppConfigProfile(ids, parse)])
+      ).toEqual([{ revision: 'first' }, { revision: 'first' }])
+      expect(mockSend).toHaveBeenCalledTimes(3)
+      finish({ Configuration: encode({ revision: 'second' }), NextPollConfigurationToken: 'next' })
+      await vi.waitFor(async () => {
+        expect(await fetchAppConfigProfile(ids, parse)).toEqual({ revision: 'second' })
+      })
+    } finally {
+      vi.useRealTimers()
+    }
+  })
 })
 
 describe('fetchAppConfigSnapshot freshness', () => {
@@ -143,6 +185,28 @@ describe('fetchAppConfigSnapshot freshness', () => {
   })
   afterEach(() => vi.useRealTimers())
 
+  it('evicts old profiles and requires fresh evidence when they are requested again', async () => {
+    mockSend.mockImplementation((command: { __type: string }) =>
+      Promise.resolve(
+        command.__type === 'start'
+          ? { InitialConfigurationToken: 'token' }
+          : { Configuration: encode({ revision: 'valid' }), NextPollConfigurationToken: 'next' }
+      )
+    )
+    const oldest = uniqueIds()
+    await fetchAppConfigSnapshot(oldest, (value) => value)
+    const results = await Promise.all(
+      Array.from({ length: 65 }, () => fetchAppConfigSnapshot(uniqueIds(), (value) => value))
+    )
+    expect(results.every((result) => result.value !== null)).toBe(true)
+    mockSend.mockRejectedValueOnce(new Error('unavailable'))
+    expect(await fetchAppConfigSnapshot(oldest, (value) => value)).toEqual({
+      value: null,
+      validatedAt: null,
+    })
+    expect(mockSend.mock.calls.at(-1)?.[0].__type).toBe('start')
+  })
+
   it('does not turn an empty first response or a cold failure into a valid snapshot', async () => {
     mockSend.mockRejectedValueOnce(new Error('unavailable'))
     expect(await fetchAppConfigSnapshot(uniqueIds(), (value) => value)).toEqual({
@@ -180,16 +244,16 @@ describe('fetchAppConfigSnapshot freshness', () => {
     const first = await fetchAppConfigSnapshot(ids, parse)
     expect(first.validatedAt).toBe(100_000)
     payload = new TextEncoder().encode('invalid json')
-    vi.setSystemTime(130_000)
+    vi.setSystemTime(130_001)
     expect(await fetchAppConfigSnapshot(ids, parse)).toEqual(first)
     payload = new Uint8Array()
-    vi.setSystemTime(160_000)
+    vi.setSystemTime(160_002)
     expect(await fetchAppConfigSnapshot(ids, parse)).toEqual(first)
     payload = encode({ revision: 'replacement' })
-    vi.setSystemTime(190_000)
+    vi.setSystemTime(190_003)
     expect(await fetchAppConfigSnapshot(ids, parse)).toEqual({
       value: { revision: 'replacement' },
-      validatedAt: 190_000,
+      validatedAt: 190_003,
     })
   })
 
@@ -210,12 +274,12 @@ describe('fetchAppConfigSnapshot freshness', () => {
     const parse = (value: unknown) => value
     await fetchAppConfigSnapshot(ids, parse)
     payload = new Uint8Array()
-    vi.setSystemTime(130_000)
+    vi.setSystemTime(130_001)
     const results = await Promise.all([
       fetchAppConfigSnapshot(ids, parse),
       fetchAppConfigSnapshot(ids, parse),
     ])
-    expect(results[0].validatedAt).toBe(130_000)
+    expect(results[0].validatedAt).toBe(130_001)
     expect(results[1]).toEqual(results[0])
     expect(mockSend.mock.calls).toHaveLength(3)
   })

apps/sim/lib/core/config/appconfig.ts

@@ -6,6 +6,7 @@ import {
 } from '@aws-sdk/client-appconfigdata'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
+import { LRUCache } from 'lru-cache'
 import { getAwsCredentialsFromEnv } from '@/lib/core/config/aws'
 import { env } from '@/lib/core/config/env'
 
@@ -22,13 +23,8 @@ export interface AppConfigProfileIdentifiers {
 interface CacheEntry<T> {
   /** Last successfully parsed value, or `null` if the config is empty/unseeded. */
   value: T | null
-  /** True once any poll has completed (success, empty payload, or error). */
-  loaded: boolean
   /** Token for the next `GetLatestConfiguration` poll, rotated on each call. */
   nextToken: string | undefined
-  expiresAt: number
-  /** In-flight poll, shared so concurrent callers don't each hit AppConfig. */
-  inflight: Promise<T | null> | null
   validatedAt: number | null
   remoteMatchesValue: boolean
   strict: boolean
@@ -39,7 +35,31 @@ export interface AppConfigSnapshot<T> {
   readonly validatedAt: number | null
 }
 
-const cache = new Map<string, CacheEntry<unknown>>()
+interface PollContext {
+  ids: AppConfigProfileIdentifiers
+  parse: (json: unknown) => unknown
+  strict: boolean
+}
+
+const cache = new LRUCache<string, CacheEntry<unknown>, PollContext>({
+  max: 64,
+  ttl: DEFAULT_TTL_MS,
+  ttlResolution: 0,
+  ignoreFetchAbort: true,
+  /** Poll intervals and snapshot freshness share the same clock. */
+  perf: { now: () => Date.now() },
+  fetchMethod: async (_key, stale, { context, options }) => {
+    const entry = stale ?? {
+      value: null,
+      nextToken: undefined,
+      validatedAt: null,
+      remoteMatchesValue: false,
+      strict: context.strict,
+    }
+    options.ttl = await poll(context.ids, context.parse, entry)
+    return entry
+  },
+})
 
 let client: AppConfigDataClient | null = null
 
@@ -66,15 +86,14 @@ function cacheKey(ids: AppConfigProfileIdentifiers): string {
  * Run one AppConfig poll for `entry`: starts a session if no token is held, then
  * calls `GetLatestConfiguration`. An empty payload means "unchanged" (or an
  * unseeded profile) and the previous value is kept. Any error is logged and the
- * last good value is retained. Marks the entry `loaded` on any outcome so callers
- * never re-block on the cold path, and honors AppConfig's `NextPollInterval` so we
+ * last good value is retained. Returns AppConfig's `NextPollInterval` so we
  * don't poll faster than the server allows (which would throttle).
  */
 async function poll<T>(
   ids: AppConfigProfileIdentifiers,
   parse: (json: unknown) => T,
   entry: CacheEntry<T>
-): Promise<T | null> {
+): Promise<number> {
   let response: GetLatestConfigurationCommandOutput
   try {
     const dataClient = getClient()
@@ -98,23 +117,16 @@ async function poll<T>(
     )
     entry.nextToken = response.NextPollConfigurationToken ?? entry.nextToken
   } catch (error) {
-    // Network/session failure: drop the token so the next attempt starts a fresh
-    // session (handles expired or invalid tokens). Mark loaded + back off so we
-    // serve the fallback and retry in the background rather than blocking every
-    // request during an outage.
+    /** A failed or expired session retries after backoff without renewing snapshot freshness. */
     entry.nextToken = undefined
-    entry.expiresAt = Date.now() + DEFAULT_TTL_MS
-    entry.loaded = true
     logger.error('AppConfig fetch failed; serving last known value', {
       profile: cacheKey(ids),
       error: getErrorMessage(error),
     })
-    return entry.value
+    return DEFAULT_TTL_MS
   }
 
-  // Parse outside the network try: a decode/parse error must NOT discard the
-  // already-rotated session token — the round trip succeeded, so the next poll
-  // can reuse it instead of opening a new session. Keep the last good value.
+  /** Decode failures retain the rotated session token and last validated value. */
   try {
     if (response.Configuration && response.Configuration.length > 0) {
       entry.remoteMatchesValue = false
@@ -136,9 +148,7 @@ async function poll<T>(
   }
 
   const intervalMs = (response.NextPollIntervalInSeconds ?? 60) * 1000
-  entry.expiresAt = Date.now() + Math.max(DEFAULT_TTL_MS, intervalMs)
-  entry.loaded = true
-  return entry.value
+  return Math.max(DEFAULT_TTL_MS, intervalMs)
 }
 
 /**
@@ -156,36 +166,11 @@ export async function fetchAppConfigProfile<T>(
   ids: AppConfigProfileIdentifiers,
   parse: (json: unknown) => T
 ): Promise<T | null> {
-  const key = cacheKey(ids)
-  const entry = (cache.get(key) as CacheEntry<T> | undefined) ?? {
-    value: null,
-    loaded: false,
-    nextToken: undefined,
-    expiresAt: 0,
-    inflight: null,
-    validatedAt: null,
-    remoteMatchesValue: false,
-    strict: false,
-  }
-  cache.set(key, entry)
-
-  // Cold: never polled — await a single shared poll so concurrent callers don't
-  // each hit AppConfig (and don't race the rotating session token).
-  if (!entry.loaded) {
-    entry.inflight ??= poll(ids, parse, entry).finally(() => {
-      entry.inflight = null
-    })
-    return entry.inflight
-  }
-
-  // Warm but stale: serve cached value, refresh once in the background.
-  if (Date.now() >= entry.expiresAt && !entry.inflight) {
-    entry.inflight = poll(ids, parse, entry).finally(() => {
-      entry.inflight = null
-    })
-  }
-
-  return entry.value
+  const entry = await cache.fetch(cacheKey(ids), {
+    context: { ids, parse, strict: false },
+    allowStale: true,
+  })
+  return (entry?.value ?? null) as T | null
 }
 
 /**
@@ -197,23 +182,12 @@ export async function fetchAppConfigSnapshot<T>(
   ids: AppConfigProfileIdentifiers,
   parse: (json: unknown) => T
 ): Promise<AppConfigSnapshot<T>> {
-  const key = `strict:${cacheKey(ids)}`
-  const entry = (cache.get(key) as CacheEntry<T> | undefined) ?? {
-    value: null,
-    loaded: false,
-    nextToken: undefined,
-    expiresAt: 0,
-    inflight: null,
-    validatedAt: null,
-    remoteMatchesValue: false,
-    strict: true,
-  }
-  cache.set(key, entry)
-  if (!entry.loaded || Date.now() >= entry.expiresAt) {
-    entry.inflight ??= poll(ids, parse, entry).finally(() => {
-      entry.inflight = null
-    })
-    await entry.inflight
-  }
-  return Object.freeze({ value: entry.value, validatedAt: entry.validatedAt })
+  const entry = await cache.fetch(`strict:${cacheKey(ids)}`, {
+    context: { ids, parse, strict: true },
+    allowStale: false,
+  })
+  return Object.freeze({
+    value: (entry?.value ?? null) as T | null,
+    validatedAt: entry?.validatedAt ?? null,
+  })
 }

apps/sim/lib/core/network/gateway.server.ts

@@ -10,16 +10,31 @@ const CONNECT_TIMEOUT_MS = 10_000
 const MAX_TUNNEL_AGE_MS = 300_000
 
 /** The proxy CA never changes trust for the upstream service. */
-async function secureGatewayTunnel(socket: Socket, hostname: string): Promise<Socket> {
+async function secureGatewayTunnel(
+  socket: Socket,
+  hostname: string,
+  port: number
+): Promise<Socket> {
   return new Promise((resolve, reject) => {
     const tls = connectTls({
       socket,
       host: hostname,
+      port,
       servername: isIP(hostname) ? undefined : hostname,
       checkServerIdentity: (_name, certificate) => checkServerIdentity(hostname, certificate),
       rejectUnauthorized: true,
       ALPNProtocols: ['http/1.1'],
     })
+    /** The inner TLS stream has no TCP descriptor; QoS must be applied to the outer socket. */
+    if ('setTypeOfService' in socket && typeof socket.setTypeOfService === 'function') {
+      const setTypeOfService = socket.setTypeOfService.bind(socket)
+      Object.defineProperty(tls, 'setTypeOfService', {
+        value(tos: number) {
+          setTypeOfService(tos)
+          return tls
+        },
+      })
+    }
     const timer = setTimeout(
       () => tls.destroy(new OutboundRoutingError('GATEWAY_UNAVAILABLE')),
       CONNECT_TIMEOUT_MS
@@ -102,7 +117,7 @@ export function createGatewayDispatcher(
         }
         const socket = await openGatewayTunnel(gateway, address, port)
         if (connection.protocol !== 'https:') return socket
-        return secureGatewayTunnel(socket, hostname)
+        return secureGatewayTunnel(socket, hostname, port)
       })().then(
         (socket) => callback(null, socket),
         () => callback(new OutboundRoutingError('GATEWAY_UNAVAILABLE'), null)