fix(run-engine): fix queue cache memory leak and replace MemoryStore with LRU cache

apps/webapp/app/services/authorizationRateLimitMiddleware.server.ts

@@ -1,5 +1,5 @@
 import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } from "@unkey/cache";
-import { MemoryStore } from "@unkey/cache/stores";
+import { createLRUMemoryStore } from "@internal/cache";
 import { Ratelimit } from "@upstash/ratelimit";
 import { Request as ExpressRequest, Response as ExpressResponse, NextFunction } from "express";
 import { createHash } from "node:crypto";
@@ -157,10 +157,7 @@ export function authorizationRateLimitMiddleware({
   limiterConfigOverride,
 }: Options) {
   const ctx = new DefaultStatefulContext();
-  const memory = new MemoryStore({
-    persistentMap: new Map(),
-    unstableEvictOnSet: { frequency: 0.001, maxItems: limiterCache?.maxItems ?? 1000 },
-  });
+  const memory = createLRUMemoryStore(limiterCache?.maxItems ?? 1000);
   const redisCacheStore = new RedisCacheStore({
     connection: {
       keyPrefix: `cache:${keyPrefix}:rate-limit-cache:`,

apps/webapp/app/services/betterstack/betterstack.server.ts

@@ -1,6 +1,6 @@
 import { type ApiResult, wrapZodFetch } from "@trigger.dev/core/v3/zodfetch";
 import { createCache, DefaultStatefulContext, Namespace } from "@unkey/cache";
-import { MemoryStore } from "@unkey/cache/stores";
+import { createLRUMemoryStore } from "@internal/cache";
 import { z } from "zod";
 import { env } from "~/env.server";
 
@@ -17,7 +17,7 @@ const IncidentSchema = z.object({
 export type Incident = z.infer<typeof IncidentSchema>;
 
 const ctx = new DefaultStatefulContext();
-const memory = new MemoryStore({ persistentMap: new Map() });
+const memory = createLRUMemoryStore(100);
 
 const cache = createCache({
   query: new Namespace<ApiResult<Incident>>(ctx, {

apps/webapp/app/services/platform.v3.server.ts

@@ -15,7 +15,7 @@ import {
   type CurrentPlan,
 } from "@trigger.dev/platform";
 import { createCache, DefaultStatefulContext, Namespace } from "@unkey/cache";
-import { MemoryStore } from "@unkey/cache/stores";
+import { createLRUMemoryStore } from "@internal/cache";
 import { existsSync, readFileSync } from "node:fs";
 import { redirect } from "remix-typedjson";
 import { z } from "zod";
@@ -45,13 +45,7 @@ const client = singleton("billingClient", initializeClient);
 
 function initializePlatformCache() {
   const ctx = new DefaultStatefulContext();
-  const memory = new MemoryStore({
-    persistentMap: new Map(),
-    unstableEvictOnSet: {
-      frequency: 0.01,
-      maxItems: 1000,
-    },
-  });
+  const memory = createLRUMemoryStore(1000);
   const redisCacheStore = new RedisCacheStore({
     connection: {
       keyPrefix: "tr:cache:platform:v3",

apps/webapp/app/services/realtime/v1StreamsGlobal.server.ts

@@ -1,6 +1,6 @@
 import {
   createCache,
-  createMemoryStore,
+  createLRUMemoryStore,
   DefaultStatefulContext,
   Namespace,
   RedisCacheStore,
@@ -97,7 +97,7 @@ function initializeS2RealtimeStreamsCache() {
     useModernCacheKeyBuilder: true,
   });
 
-  const memoryStore = createMemoryStore(5000, 0.001);
+  const memoryStore = createLRUMemoryStore(5000);
 
   return createCache({
     accessToken: new Namespace<string>(ctx, {

apps/webapp/app/services/realtimeClient.server.ts

@@ -8,7 +8,7 @@ import { longPollingFetch } from "~/utils/longPollingFetch";
 import { logger } from "./logger.server";
 import { jumpHash } from "@trigger.dev/core/v3/serverOnly";
 import { Cache, createCache, DefaultStatefulContext, Namespace } from "@unkey/cache";
-import { MemoryStore } from "@unkey/cache/stores";
+import { createLRUMemoryStore } from "@internal/cache";
 import { RedisCacheStore } from "./unkey/redisCacheStore.server";
 import { env } from "~/env.server";
 import { API_VERSIONS, CURRENT_API_VERSION } from "~/api/versions";
@@ -84,10 +84,7 @@ export class RealtimeClient {
     this.#registerCommands();
 
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({
-      persistentMap: new Map(),
-      unstableEvictOnSet: { frequency: 0.01, maxItems: 1000 },
-    });
+    const memory = createLRUMemoryStore(1000);
     const redisCacheStore = new RedisCacheStore({
       connection: {
         keyPrefix: "tr:cache:realtime",

apps/webapp/app/services/requestIdempotency.server.ts

@@ -1,6 +1,6 @@
 import { Logger, LogLevel } from "@trigger.dev/core/logger";
 import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } from "@unkey/cache";
-import { MemoryStore } from "@unkey/cache/stores";
+import { createLRUMemoryStore } from "@internal/cache";
 import { RedisCacheStore } from "./unkey/redisCacheStore.server";
 import { RedisWithClusterOptions } from "~/redis.server";
 import { validate as uuidValidate, version as uuidVersion } from "uuid";
@@ -33,13 +33,7 @@ export class RequestIdempotencyService<TTypes extends string> {
       : "request-idempotency:";
 
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({
-      persistentMap: new Map(),
-      unstableEvictOnSet: {
-        frequency: 0.001,
-        maxItems: 1000,
-      },
-    });
+    const memory = createLRUMemoryStore(1000);
     const redisCacheStore = new RedisCacheStore({
       name: "request-idempotency",
       connection: {

apps/webapp/app/v3/marqs/fairDequeuingStrategy.server.ts

@@ -1,5 +1,5 @@
 import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } from "@unkey/cache";
-import { MemoryStore } from "@unkey/cache/stores";
+import { createLRUMemoryStore } from "@internal/cache";
 import { randomUUID } from "crypto";
 import { Redis } from "ioredis";
 import { EnvQueues, MarQSFairDequeueStrategy, MarQSKeyProducer } from "./types";
@@ -99,13 +99,7 @@ export class FairDequeuingStrategy implements MarQSFairDequeueStrategy {
 
   constructor(private options: FairDequeuingStrategyOptions) {
     const ctx = new DefaultStatefulContext();
-    const memory = new MemoryStore({
-      persistentMap: new Map(),
-      unstableEvictOnSet: {
-        frequency: 0.01,
-        maxItems: 500,
-      },
-    });
+    const memory = createLRUMemoryStore(500);
 
     this._cache = createCache({
       concurrencyLimit: new Namespace<number>(ctx, {

apps/webapp/app/v3/services/worker/workerGroupTokenService.server.ts

@@ -1,4 +1,4 @@
-import { createCache, createMemoryStore, DefaultStatefulContext, Namespace } from "@internal/cache";
+import { createCache, createLRUMemoryStore, DefaultStatefulContext, Namespace } from "@internal/cache";
 import {
   CheckpointInput,
   CompleteRunAttemptResult,
@@ -39,7 +39,7 @@ function createAuthenticatedWorkerInstanceCache() {
     authenticatedWorkerInstance: new Namespace<AuthenticatedWorkerInstance>(
       new DefaultStatefulContext(),
       {
-        stores: [createMemoryStore(1000, 0.001)],
+        stores: [createLRUMemoryStore(1000)],
         fresh: 60_000 * 10, // 10 minutes
         stale: 60_000 * 11, // 11 minutes
       }

internal-packages/cache/package.json

@@ -10,9 +10,12 @@
     "@trigger.dev/core": "workspace:*",
     "@unkey/cache": "^1.5.0",
     "@unkey/error": "^0.2.0",
+    "lru-cache": "^11.2.4",
     "superjson": "^2.2.1"
   },
   "scripts": {
-    "typecheck": "tsc --noEmit"
+    "typecheck": "tsc --noEmit",
+    "test": "vitest",
+    "test:watch": "vitest"
   }
 }

internal-packages/cache/src/index.ts

@@ -8,3 +8,8 @@ export {
 export { type Result, Ok, Err } from "@unkey/error";
 export { RedisCacheStore } from "./stores/redis.js";
 export { createMemoryStore, type MemoryStore } from "./stores/memory.js";
+export {
+  LRUMemoryStore,
+  createLRUMemoryStore,
+  type LRUMemoryStoreConfig,
+} from "./stores/lruMemory.js";