feat(cache): in-memory tier, size limits, sharding, observability

blocks/loader.ts

@@ -146,6 +146,16 @@ const stats = {
     unit: "ms",
     valueType: ValueType.DOUBLE,
   }),
+  cacheEntrySize: meter.createHistogram("loader_cache_entry_size", {
+    description: "size of cached loader responses in bytes",
+    unit: "bytes",
+    valueType: ValueType.DOUBLE,
+  }),
+  bgRevalidation: meter.createHistogram("loader_bg_revalidation", {
+    description: "duration of background stale-while-revalidate calls",
+    unit: "ms",
+    valueType: ValueType.DOUBLE,
+  }),
 };
 
 let maybeCache: Cache | undefined;
@@ -155,6 +165,9 @@ caches?.open("loader")
   .catch(() => maybeCache = undefined);
 
 const MAX_AGE_S = parseInt(Deno.env.get("CACHE_MAX_AGE_S") ?? "60"); // 60 seconds
+const CACHE_MAX_ENTRY_SIZE = parseInt(
+  Deno.env.get("CACHE_MAX_ENTRY_SIZE") ?? "2097152", // 2 MB
+) || 2097152;
 
 // Reuse TextEncoder instance to avoid repeated instantiation
 const textEncoder = new TextEncoder();
@@ -248,7 +261,14 @@ const wrapLoader = (
           !shouldNotCache && ctx.vary?.push(cacheKeyValue);
 
           status = "bypass";
-          stats.cache.add(1, { status, loader });
+          const bypassReason = isCacheNoStore
+            ? "no-store"
+            : isCacheNoCache
+            ? "no-cache"
+            : isCacheKeyNull
+            ? "null-key"
+            : "disabled";
+          stats.cache.add(1, { status, loader, reason: bypassReason });
 
           RequestContext?.signal?.throwIfAborted();
           return await handler(props, req, ctx);
@@ -297,6 +317,19 @@ const wrapLoader = (
           // Serialize and encode once on the main thread.
           const jsonStringEncoded = textEncoder.encode(JSON.stringify(json));
 
+          // Skip caching oversized entries to protect disk and memory.
+          // Also evict any existing stale entry so it doesn't stay pinned forever.
+          if (jsonStringEncoded.length > CACHE_MAX_ENTRY_SIZE) {
+            cache.delete(request).catch((error) =>
+              logger.error(`loader error ${error}`)
+            );
+            return json;
+          }
+
+          if (OTEL_ENABLE_EXTRA_METRICS) {
+            stats.cacheEntrySize.record(jsonStringEncoded.length, { loader });
+          }
+
           const expires = new Date(Date.now() + (cacheMaxAge * 1e3))
             .toUTCString();
           const headerPairs: [string, string][] = [
@@ -305,7 +338,7 @@ const wrapLoader = (
             ["Content-Length", "" + jsonStringEncoded.length],
           ];
 
-          // Cache write goes through the full chain (LRU → filesystem)
+          // Cache write goes through the full chain (LRU → in-memory → filesystem)
           // so the LRU registers the key for fast match lookups.
           // The filesystem layer offloads the actual I/O to a worker thread
           // when DECO_CACHE_WRITE_WORKER=true.
@@ -336,13 +369,44 @@ const wrapLoader = (
             status = "stale";
             stats.cache.add(1, { status, loader });
 
-            bgFlights.do(request.url, callHandlerAndCache)
-              .catch((error) => logger.error(`loader error ${error}`));
+            // Timer lives inside the singleFlight fn so it records exactly once
+            // per revalidation, not once per concurrent waiter on the same key.
+            bgFlights.do(request.url, async () => {
+              const bgStart = performance.now();
+              try {
+                return await callHandlerAndCache();
+              } finally {
+                if (OTEL_ENABLE_EXTRA_METRICS) {
+                  stats.bgRevalidation.record(
+                    performance.now() - bgStart,
+                    { loader },
+                  );
+                }
+              }
+            }).catch((error) => logger.error(`loader error ${error}`));
           } else {
             status = "hit";
             stats.cache.add(1, { status, loader });
           }
 
+          if (OTEL_ENABLE_EXTRA_METRICS) {
+            const cl = parseInt(
+              matched.headers.get("Content-Length") ?? "0",
+            );
+            if (cl > 0) {
+              stats.cacheEntrySize.record(cl, { loader, status });
+            }
+          }
+
+          if (OTEL_ENABLE_EXTRA_METRICS) {
+            const parseStart = performance.now();
+            const result = await matched.json();
+            stats.latency.record(performance.now() - parseStart, {
+              loader,
+              status: "json_parse",
+            });
+            return result;
+          }
           return await matched.json();
         };
 

runtime/caches/cacheWriteWorker.ts

@@ -1,6 +1,10 @@
 // Worker thread for cache write operations.
 // Offloads SHA1 hashing, buffer combining, and FS writes from the main event loop.
 
+const CACHE_MAX_ENTRY_SIZE = parseInt(
+  Deno.env.get("CACHE_MAX_ENTRY_SIZE") ?? "2097152", // 2 MB
+) || 2097152;
+
 const textEncoder = new TextEncoder();
 
 const initializedDirs = new Set<string>();
@@ -57,6 +61,12 @@ function generateCombinedBuffer(
   return buf;
 }
 
+function shardedPath(cacheDir: string, key: string): string {
+  const l1 = key.substring(0, 2);
+  const l2 = key.substring(2, 4);
+  return `${cacheDir}/${l1}/${l2}/${key}`;
+}
+
 // --- Message handler ---
 
 export interface CacheWriteMessage {
@@ -85,8 +95,12 @@ self.onmessage = async (e: MessageEvent<CacheWriteMessage>) => {
     // Combine into single buffer
     const buffer = generateCombinedBuffer(body, headersBytes);
 
-    // Write to filesystem
-    const filePath = `${cacheDir}/${cacheKey}`;
+    if (buffer.length > CACHE_MAX_ENTRY_SIZE) return;
+
+    // Write to filesystem (with hex sharding for directory distribution)
+    const filePath = shardedPath(cacheDir, cacheKey);
+    const dir = filePath.substring(0, filePath.lastIndexOf("/"));
+    ensureCacheDir(dir);
     await Deno.writeFile(filePath, buffer);
   } catch (err) {
     console.error("[cache-write-worker] error:", err);

runtime/caches/common.ts

@@ -46,6 +46,15 @@ export const withInstrumentation = (
             const result = getCacheStatus(isMatch);
 
             span.setAttribute("cache_status", result);
+            if (isMatch) {
+              const cl = isMatch.headers.get("Content-Length");
+              if (cl) span.setAttribute("content_length", parseInt(cl));
+              const tier = isMatch.headers.get("X-Cache-Tier");
+              if (tier) {
+                span.setAttribute("cache_tier", parseInt(tier));
+                isMatch.headers.delete("X-Cache-Tier");
+              }
+            }
             cacheHit.add(1, {
               result,
               engine,

runtime/caches/fileSystem.ts

@@ -11,6 +11,18 @@ import {
 const FILE_SYSTEM_CACHE_DIRECTORY =
   Deno.env.get("FILE_SYSTEM_CACHE_DIRECTORY") ?? "/tmp/deco_cache";
 
+const CACHE_MAX_ENTRY_SIZE = parseInt(
+  Deno.env.get("CACHE_MAX_ENTRY_SIZE") ?? "2097152", // 2 MB
+);
+
+const initializedShardDirs = new Set<string>();
+
+function shardedPath(cacheDir: string, key: string): string {
+  const l1 = key.substring(0, 2);
+  const l2 = key.substring(2, 4);
+  return `${cacheDir}/${l1}/${l2}/${key}`;
+}
+
 // Reuse TextEncoder instance to avoid repeated instantiation
 const textEncoder = new TextEncoder();
 
@@ -106,7 +118,7 @@ function createFileSystemCache(): CacheStorage {
       if (
         FILE_SYSTEM_CACHE_DIRECTORY && !existsSync(FILE_SYSTEM_CACHE_DIRECTORY)
       ) {
-        await Deno.mkdirSync(FILE_SYSTEM_CACHE_DIRECTORY, { recursive: true });
+        await Deno.mkdir(FILE_SYSTEM_CACHE_DIRECTORY, { recursive: true });
       }
       isCacheInitialized = true;
     } catch (err) {
@@ -118,11 +130,20 @@ function createFileSystemCache(): CacheStorage {
     key: string,
     responseArray: Uint8Array,
   ) {
+    if (responseArray.length > CACHE_MAX_ENTRY_SIZE) return;
     if (!isCacheInitialized) {
       await assertCacheDirectory();
     }
-    const filePath = `${FILE_SYSTEM_CACHE_DIRECTORY}/${key}`;
-
+    const filePath = shardedPath(FILE_SYSTEM_CACHE_DIRECTORY, key);
+    const dir = filePath.substring(0, filePath.lastIndexOf("/"));
+    if (!initializedShardDirs.has(dir)) {
+      try {
+        await Deno.mkdir(dir, { recursive: true });
+        initializedShardDirs.add(dir);
+      } catch {
+        // transient failure — don't mark initialized so next write retries mkdir
+      }
+    }
     await Deno.writeFile(filePath, responseArray);
     return;
   }
@@ -132,8 +153,12 @@ function createFileSystemCache(): CacheStorage {
       await assertCacheDirectory();
     }
     try {
-      const filePath = `${FILE_SYSTEM_CACHE_DIRECTORY}/${key}`;
+      const filePath = shardedPath(FILE_SYSTEM_CACHE_DIRECTORY, key);
       const fileContent = await Deno.readFile(filePath);
+      if (fileContent.length > CACHE_MAX_ENTRY_SIZE) {
+        Deno.remove(filePath).catch(() => {});
+        return null;
+      }
       return fileContent;
     } catch (_err) {
       const err = _err as { code?: string };
@@ -151,7 +176,7 @@ function createFileSystemCache(): CacheStorage {
       await assertCacheDirectory();
     }
     try {
-      const filePath = `${FILE_SYSTEM_CACHE_DIRECTORY}/${key}`;
+      const filePath = shardedPath(FILE_SYSTEM_CACHE_DIRECTORY, key);
       await Deno.remove(filePath);
       return true;
     } catch (err) {

runtime/caches/inMemoryCache.ts

@@ -0,0 +1,85 @@
+import { LRUCache } from "npm:lru-cache@10.2.0";
+import {
+  assertCanBeCached,
+  assertNoOptions,
+  baseCache,
+  withCacheNamespace,
+} from "./utils.ts";
+
+const MEMORY_CACHE_MAX_SIZE = parseInt(
+  Deno.env.get("MEMORY_CACHE_MAX_SIZE") ?? "268435456", // 256 MB
+);
+const MEMORY_CACHE_MAX_ITEMS = parseInt(
+  Deno.env.get("MEMORY_CACHE_MAX_ITEMS") ?? "2048",
+);
+
+interface CacheEntry {
+  body: Uint8Array;
+  headers: [string, string][];
+  status: number;
+}
+
+function createInMemoryCache(): CacheStorage {
+  const store = new LRUCache<string, CacheEntry>({
+    max: MEMORY_CACHE_MAX_ITEMS,
+    maxSize: MEMORY_CACHE_MAX_SIZE,
+    sizeCalculation: (entry) => entry.body.length,
+  });
+
+  const caches: CacheStorage = {
+    delete: () => {
+      throw new Error("Not Implemented");
+    },
+    has: () => {
+      throw new Error("Not Implemented");
+    },
+    keys: () => {
+      throw new Error("Not Implemented");
+    },
+    match: () => {
+      throw new Error("Not Implemented");
+    },
+    open: (cacheName: string): Promise<Cache> => {
+      const requestURLSHA1 = withCacheNamespace(cacheName);
+      return Promise.resolve({
+        ...baseCache,
+        delete: async (
+          request: RequestInfo | URL,
+          _options?: CacheQueryOptions,
+        ): Promise<boolean> => {
+          const cacheKey = await requestURLSHA1(request);
+          return store.delete(cacheKey);
+        },
+        match: async (
+          request: RequestInfo | URL,
+          options?: CacheQueryOptions,
+        ): Promise<Response | undefined> => {
+          assertNoOptions(options);
+          const cacheKey = await requestURLSHA1(request);
+          const entry = store.get(cacheKey);
+          if (!entry) return undefined;
+          return new Response(entry.body as unknown as BodyInit, {
+            headers: new Headers(entry.headers),
+            status: entry.status,
+          });
+        },
+        put: async (
+          request: RequestInfo | URL,
+          response: Response,
+        ): Promise<void> => {
+          const req = new Request(request);
+          assertCanBeCached(req, response);
+          if (!response.body) return;
+          const cacheKey = await requestURLSHA1(request);
+          const body = new Uint8Array(await response.arrayBuffer());
+          const headers: [string, string][] = [...response.headers.entries()];
+          store.set(cacheKey, { body, headers, status: response.status });
+        },
+      });
+    },
+  };
+
+  return caches;
+}
+
+export const caches = createInMemoryCache();