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
+) || 2097152;
+
+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,24 @@ function createFileSystemCache(): CacheStorage {
     key: string,
     responseArray: Uint8Array,
   ) {
+    if (responseArray.length > CACHE_MAX_ENTRY_SIZE) {
+      // Evict any existing entry so stale data doesn't stay pinned on disk.
+      deleteFile(key).catch(() => {});
+      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 +157,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 +180,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.test.ts

@@ -0,0 +1,68 @@
+import { assertEquals, assertNotEquals } from "@std/assert";
+import { caches } from "./inMemoryCache.ts";
+
+// Use unique cache names per test to avoid cross-test contamination
+// from the shared singleton LRU store.
+let seq = 0;
+const nextCache = () => `inMemoryTest_${seq++}_${Date.now()}`;
+
+const REQ = new Request("https://example.com/item");
+
+Deno.test("inMemoryCache: preserves response body", async () => {
+  const cache = await caches.open(nextCache());
+  await cache.put(REQ, new Response("hello world"));
+  const result = await cache.match(REQ);
+  assertNotEquals(result, undefined);
+  assertEquals(await result!.text(), "hello world");
+});
+
+Deno.test("inMemoryCache: preserves response status", async () => {
+  const cache = await caches.open(nextCache());
+  await cache.put(REQ, new Response("not found", { status: 404 }));
+  const result = await cache.match(REQ);
+  assertEquals(result?.status, 404);
+});
+
+Deno.test("inMemoryCache: preserves non-standard status codes", async () => {
+  const cache = await caches.open(nextCache());
+  await cache.put(REQ, new Response("gone", { status: 410 }));
+  assertEquals((await cache.match(REQ))?.status, 410);
+});
+
+Deno.test("inMemoryCache: preserves response headers", async () => {
+  const cache = await caches.open(nextCache());
+  await cache.put(
+    REQ,
+    new Response("data", {
+      headers: { "content-type": "application/json", "x-custom": "value" },
+    }),
+  );
+  const result = await cache.match(REQ);
+  assertEquals(result?.headers.get("content-type"), "application/json");
+  assertEquals(result?.headers.get("x-custom"), "value");
+});
+
+Deno.test("inMemoryCache: miss returns undefined", async () => {
+  const cache = await caches.open(nextCache());
+  assertEquals(await cache.match(REQ), undefined);
+});
+
+Deno.test("inMemoryCache: delete removes entry", async () => {
+  const cache = await caches.open(nextCache());
+  await cache.put(REQ, new Response("data"));
+  assertNotEquals(await cache.match(REQ), undefined);
+  await cache.delete(REQ);
+  assertEquals(await cache.match(REQ), undefined);
+});
+
+Deno.test("inMemoryCache: different requests are independent", async () => {
+  const cache = await caches.open(nextCache());
+  const req1 = new Request("https://example.com/1");
+  const req2 = new Request("https://example.com/2");
+
+  await cache.put(req1, new Response("one", { status: 200 }));
+  await cache.put(req2, new Response("two", { status: 201 }));
+
+  assertEquals((await cache.match(req1))?.status, 200);
+  assertEquals((await cache.match(req2))?.status, 201);
+});