Fix cache memory store unbounded memory usage

ericallam · ericallam · commit 7b1e77b2da6f · 2025-09-29T16:18:12.000+01:00
diff --git a/apps/webapp/app/services/authorizationRateLimitMiddleware.server.ts b/apps/webapp/app/services/authorizationRateLimitMiddleware.server.ts
@@ -64,6 +64,7 @@ type Options = {
   limiterCache?: {
     fresh: number;
     stale: number;
+    maxItems: number;
   };
   log?: {
     requests?: boolean;
@@ -145,7 +146,10 @@ export function authorizationRateLimitMiddleware({
   limiterConfigOverride,
 }: Options) {
   const ctx = new DefaultStatefulContext();
-  const memory = new MemoryStore({ persistentMap: new Map() });
+  const memory = new MemoryStore({
+    persistentMap: new Map(),
+    unstableEvictOnSet: { frequency: 0.001, maxItems: limiterCache?.maxItems ?? 1000 },
+  });
   const redisCacheStore = new RedisCacheStore({
     connection: {
       keyPrefix: `cache:${keyPrefix}:rate-limit-cache:`,
diff --git a/apps/webapp/app/services/platform.v3.server.ts b/apps/webapp/app/services/platform.v3.server.ts
@@ -44,7 +44,13 @@ const client = singleton("billingClient", initializeClient);
 
 function initializePlatformCache() {
   const ctx = new DefaultStatefulContext();
-  const memory = new MemoryStore({ persistentMap: new Map() });
+  const memory = new MemoryStore({
+    persistentMap: new Map(),
+    unstableEvictOnSet: {
+      frequency: 0.01,
+      maxItems: 1000,
+    },
+  });
   const redisCacheStore = new RedisCacheStore({
     connection: {
       keyPrefix: "tr:cache:platform:v3",
diff --git a/apps/webapp/app/services/realtimeClient.server.ts b/apps/webapp/app/services/realtimeClient.server.ts
@@ -83,7 +83,10 @@ export class RealtimeClient {
     this.#registerCommands();
 
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({ persistentMap: new Map() });
+    const memory = new MemoryStore({
+      persistentMap: new Map(),
+      unstableEvictOnSet: { frequency: 0.01, maxItems: 1000 },
+    });
     const redisCacheStore = new RedisCacheStore({
       connection: {
         keyPrefix: "tr:cache:realtime",
diff --git a/apps/webapp/app/services/requestIdempotency.server.ts b/apps/webapp/app/services/requestIdempotency.server.ts
@@ -33,7 +33,13 @@ export class RequestIdempotencyService<TTypes extends string> {
       : "request-idempotency:";
 
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({ persistentMap: new Map() });
+    const memory = new MemoryStore({
+      persistentMap: new Map(),
+      unstableEvictOnSet: {
+        frequency: 0.001,
+        maxItems: 1000,
+      },
+    });
     const redisCacheStore = new RedisCacheStore({
       name: "request-idempotency",
       connection: {
diff --git a/apps/webapp/app/v3/eventRepository/originalRunIdCache.server.ts b/apps/webapp/app/v3/eventRepository/originalRunIdCache.server.ts
@@ -1,7 +1,6 @@
 import {
   createCache,
   DefaultStatefulContext,
-  MemoryStore,
   Namespace,
   RedisCacheStore,
   type UnkeyCache,
@@ -25,7 +24,6 @@ export class OriginalRunIdCache {
   constructor(options: OriginalRunIdCacheOptions) {
     // Initialize cache
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({ persistentMap: new Map() });
     const redisCacheStore = new RedisCacheStore({
       name: "original-run-id-cache",
       connection: {
@@ -37,7 +35,7 @@ export class OriginalRunIdCache {
 
     this.cache = createCache({
       originalRunId: new Namespace<string>(ctx, {
-        stores: [memory, redisCacheStore],
+        stores: [redisCacheStore],
         fresh: ORIGINAL_RUN_ID_FRESH_TTL,
         stale: ORIGINAL_RUN_ID_STALE_TTL,
       }),
diff --git a/apps/webapp/app/v3/marqs/fairDequeuingStrategy.server.ts b/apps/webapp/app/v3/marqs/fairDequeuingStrategy.server.ts
@@ -99,7 +99,13 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
 
   constructor(private options: FairDequeuingStrategyOptions) {
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({ persistentMap: new Map() });
+    const memory = new MemoryStore({
+      persistentMap: new Map(),
+      unstableEvictOnSet: {
+        frequency: 0.01,
+        maxItems: 500,
+      },
+    });
 
     this._cache = createCache({
       concurrencyLimit: new Namespace<number>(ctx, {
diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts
@@ -220,8 +220,8 @@ function convertLogsToCreateableEvents(
           SemanticInternalAttributes.METADATA
         );
 
-        const properties = {
-          ...convertKeyValueItemsToMap(
+        const properties =
+          convertKeyValueItemsToMap(
             truncateAttributes(log.attributes ?? [], spanAttributeValueLengthLimit),
             [],
             undefined,
@@ -233,8 +233,7 @@ function convertLogsToCreateableEvents(
               SemanticInternalAttributes.METRIC_EVENTS,
               SemanticInternalAttributes.TRIGGER,
             ]
-          ),
-        };
+          ) ?? {};
 
         return {
           traceId: binaryToHex(log.traceId),
@@ -257,7 +256,7 @@ function convertLogsToCreateableEvents(
           metadata: logProperties.metadata ?? resourceProperties.metadata ?? {},
           environmentId:
             logProperties.environmentId ?? resourceProperties.environmentId ?? "unknown",
-          environmentType: "DEVELOPMENT" as const,
+          environmentType: "DEVELOPMENT" as const, // We've deprecated this but we need to keep it for backwards compatibility
           organizationId:
             logProperties.organizationId ?? resourceProperties.organizationId ?? "unknown",
           projectId: logProperties.projectId ?? resourceProperties.projectId ?? "unknown",
@@ -304,8 +303,8 @@ function convertSpansToCreateableEvents(
           SemanticInternalAttributes.METADATA
         );
 
-        const properties = {
-          ...convertKeyValueItemsToMap(
+        const properties =
+          convertKeyValueItemsToMap(
             truncateAttributes(span.attributes ?? [], spanAttributeValueLengthLimit),
             [],
             undefined,
@@ -317,8 +316,7 @@ function convertSpansToCreateableEvents(
               SemanticInternalAttributes.METRIC_EVENTS,
               SemanticInternalAttributes.TRIGGER,
             ]
-          ),
-        };
+          ) ?? {};
 
         return {
           traceId: binaryToHex(span.traceId),
diff --git a/apps/webapp/app/v3/services/tracePubSub.server.ts b/apps/webapp/app/v3/services/tracePubSub.server.ts
@@ -1,5 +1,5 @@
 import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
-import { EventEmitter } from "node:stream";
+import { EventEmitter } from "node:events";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
 
diff --git a/apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts b/apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts
@@ -1,4 +1,4 @@
-import { createCache, DefaultStatefulContext, MemoryStore, Namespace } from "@internal/cache";
+import { createCache, createMemoryStore, DefaultStatefulContext, Namespace } from "@internal/cache";
 import {
   CheckpointInput,
   CompleteRunAttemptResult,
@@ -39,7 +39,7 @@ function createAuthenticatedWorkerInstanceCache() {
     authenticatedWorkerInstance: new Namespace<AuthenticatedWorkerInstance>(
       new DefaultStatefulContext(),
       {
-        stores: [new MemoryStore({ persistentMap: new Map() })],
+        stores: [createMemoryStore(1000, 0.001)],
         fresh: 60_000 * 10, // 10 minutes
         stale: 60_000 * 11, // 11 minutes
       }
diff --git a/internal-packages/cache/package.json b/internal-packages/cache/package.json
@@ -6,10 +6,10 @@
   "types": "./src/index.ts",
   "type": "module",
   "dependencies": {
+    "@internal/redis": "workspace:*",
+    "@trigger.dev/core": "workspace:*",
     "@unkey/cache": "^1.5.0",
     "@unkey/error": "^0.2.0",
-    "@trigger.dev/core": "workspace:*",
-    "@internal/redis": "workspace:*",
     "superjson": "^2.2.1"
   },
   "scripts": {
diff --git a/internal-packages/cache/src/index.ts b/internal-packages/cache/src/index.ts
@@ -6,5 +6,5 @@ export {
   type CacheError,
 } from "@unkey/cache";
 export { type Result, Ok, Err } from "@unkey/error";
-export { MemoryStore } from "@unkey/cache/stores";
 export { RedisCacheStore } from "./stores/redis.js";
+export { createMemoryStore, type MemoryStore } from "./stores/memory.js";
diff --git a/internal-packages/cache/src/stores/memory.ts b/internal-packages/cache/src/stores/memory.ts
@@ -0,0 +1,13 @@
+import { MemoryStore } from "@unkey/cache/stores";
+
+export type { MemoryStore };
+
+export function createMemoryStore(maxItems: number, frequency: number = 0.01) {
+  return new MemoryStore({
+    persistentMap: new Map(),
+    unstableEvictOnSet: {
+      frequency,
+      maxItems,
+    },
+  });
+}
diff --git a/internal-packages/run-engine/src/engine/billingCache.ts b/internal-packages/run-engine/src/engine/billingCache.ts
@@ -8,6 +8,7 @@ import {
   type UnkeyCache,
   type CacheError,
   type Result,
+  createMemoryStore,
 } from "@internal/cache";
 import type { RedisOptions } from "@internal/redis";
 import type { Logger } from "@trigger.dev/core/logger";
@@ -41,7 +42,6 @@ export class BillingCache {
 
     // Initialize cache
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({ persistentMap: new Map() });
     const redisCacheStore = new RedisCacheStore({
       name: "billing-cache",
       connection: {
@@ -53,7 +53,7 @@ export class BillingCache {
 
     this.cache = createCache({
       currentPlan: new Namespace<BillingPlan>(ctx, {
-        stores: [memory, redisCacheStore],
+        stores: [createMemoryStore(1000), redisCacheStore],
         fresh: BILLING_FRESH_TTL,
         stale: BILLING_STALE_TTL,
       }),
diff --git a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts
@@ -1,5 +1,6 @@
 import {
   createCache,
+  createMemoryStore,
   DefaultStatefulContext,
   MemoryStore,
   Namespace,
@@ -125,8 +126,7 @@ export class RunAttemptSystem {
     this.delayedRunSystem = options.delayedRunSystem;
 
     const ctx = new DefaultStatefulContext();
-    // TODO: use an LRU cache for memory store
-    const memory = new MemoryStore({ persistentMap: new Map() });
+    const memory = createMemoryStore(5000, 0.001);
     const redisCacheStore = new RedisCacheStore({
       name: "run-attempt-system",
       connection: {
diff --git a/internal-packages/run-engine/src/run-queue/fairQueueSelectionStrategy.ts b/internal-packages/run-engine/src/run-queue/fairQueueSelectionStrategy.ts
@@ -6,6 +6,7 @@ import {
   Namespace,
   type UnkeyCache,
   MemoryStore,
+  createMemoryStore,
 } from "@internal/cache";
 import { randomUUID } from "crypto";
 import seedrandom from "seedrandom";
@@ -106,7 +107,7 @@ export class FairQueueSelectionStrategy implements RunQueueSelectionStrategy {
 
   constructor(private options: FairQueueSelectionStrategyOptions) {
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({ persistentMap: new Map() });
+    const memory = createMemoryStore(1000);
 
     this._cache = createCache({
       concurrencyLimit: new Namespace<number>(ctx, {

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-import { createCache, DefaultStatefulContext, MemoryStore, Namespace } from "@internal/cache";`
	`1`	`+import { createCache, createMemoryStore, DefaultStatefulContext, Namespace } from "@internal/cache";`
`2`	`2`	`import {`
`3`	`3`	`CheckpointInput,`
`4`	`4`	`CompleteRunAttemptResult,`
`@@ -39,7 +39,7 @@ function createAuthenticatedWorkerInstanceCache() {`
`39`	`39`	`authenticatedWorkerInstance: new Namespace<AuthenticatedWorkerInstance>(`
`40`	`40`	`new DefaultStatefulContext(),`
`41`	`41`	`{`
`42`		`- stores: [new MemoryStore({ persistentMap: new Map() })],`
	`42`	`+ stores: [createMemoryStore(1000, 0.001)],`
`43`	`43`	`fresh: 60_000 * 10, // 10 minutes`
`44`	`44`	`stale: 60_000 * 11, // 11 minutes`
`45`	`45`	`}`