Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 44 additions & 2 deletions blocks/loader.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -336,13 +346,45 @@ 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, {
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Outdated
loader,
status: "json_parse",
});
return result;
}

return await matched.json();
};

Expand Down
9 changes: 9 additions & 0 deletions runtime/caches/common.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
88 changes: 88 additions & 0 deletions runtime/caches/inMemoryCache.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
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");

// The admission filter requires MEMORY_CACHE_MIN_HITS puts before a key is
// stored in RAM (default 2). Tests that expect a hit must use this helper.
async function putTwice(
cache: Cache,
req: RequestInfo | URL,
makeResponse: () => Response,
) {
await cache.put(req, makeResponse());
await cache.put(req, makeResponse());
}

Deno.test("inMemoryCache: preserves response body", async () => {
const cache = await caches.open(nextCache());
await putTwice(cache, 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 putTwice(cache, 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 putTwice(cache, 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 putTwice(
cache,
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: single put does not admit to L1 (admission filter)", async () => {
const cache = await caches.open(nextCache());
await cache.put(REQ, new Response("data"));
// First put should not be stored — key has not earned admission yet.
assertEquals(await cache.match(REQ), undefined);
});

Deno.test("inMemoryCache: delete removes entry", async () => {
const cache = await caches.open(nextCache());
await putTwice(cache, 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 putTwice(cache, req1, () => new Response("one", { status: 200 }));
await putTwice(cache, req2, () => new Response("two", { status: 201 }));

assertEquals((await cache.match(req1))?.status, 200);
assertEquals((await cache.match(req2))?.status, 201);
});
146 changes: 146 additions & 0 deletions runtime/caches/inMemoryCache.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
import { LRUCache } from "npm:lru-cache@10.2.0";
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
import { ValueType } from "../../deps.ts";
import { logger } from "../../observability/otel/config.ts";
import { meter } from "../../observability/otel/metrics.ts";
import {
assertCanBeCached,
assertNoOptions,
baseCache,
withCacheNamespace,
} from "./utils.ts";

const MEMORY_CACHE_MAX_SIZE = parseInt(
Deno.env.get("MEMORY_CACHE_MAX_SIZE") ?? "268435456", // 256 MB
) || 268435456;
const MEMORY_CACHE_MAX_ITEMS = parseInt(
Deno.env.get("MEMORY_CACHE_MAX_ITEMS") ?? "2048",
) || 2048;
const CACHE_MAX_ENTRY_SIZE = parseInt(
Deno.env.get("CACHE_MAX_ENTRY_SIZE") ?? "2097152", // 2 MB
) || 2097152;
// Minimum number of cache writes before a key is admitted to L1 (in-memory).
// Prevents one-hit wonders (bot traffic, rare URLs) from displacing hot keys.
// Default 2: a key must be written twice (i.e. accessed from L2 on separate requests)
// before it earns a spot in RAM.
const MEMORY_CACHE_MIN_HITS = parseInt(
Deno.env.get("MEMORY_CACHE_MIN_HITS") ?? "2",
) || 2;

const l1EvictionCounter = meter.createCounter("l1_cache_eviction", {
description: "number of entries evicted from the L1 in-memory cache",
unit: "1",
valueType: ValueType.DOUBLE,
});

interface CacheEntry {
body: Uint8Array;
headers: [string, string][];
status: number;
}

function createInMemoryCache(): CacheStorage {
let totalEvictions = 0;

const store = new LRUCache<string, CacheEntry>({
max: MEMORY_CACHE_MAX_ITEMS,
maxSize: MEMORY_CACHE_MAX_SIZE,
sizeCalculation: (entry) => entry.body.length,
dispose: (_value, _key, reason) => {
l1EvictionCounter.add(1, { reason });
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Outdated
// Log a warning periodically so operators can see if L1 is under pressure.
// "evict" means the cache was full and had to drop an entry to make room —
// that's the signal to watch. "delete" and "set" are normal lifecycle events.
if (reason === "evict") {
totalEvictions++;
// Log on the 1st eviction and every 100 thereafter to avoid log spam.
if (totalEvictions === 1 || totalEvictions % 100 === 0) {
logger.warn(
`l1_cache: ${totalEvictions} total evictions — L1 is full and dropping entries. ` +
`Consider increasing MEMORY_CACHE_MAX_SIZE (current: ${MEMORY_CACHE_MAX_SIZE}) ` +
`or MEMORY_CACHE_MAX_ITEMS (current: ${MEMORY_CACHE_MAX_ITEMS}).`,
);
}
}
},
});

// Admission filter: tracks how many times each key has been presented for storage.
// A key must be seen MEMORY_CACHE_MIN_HITS times before it's actually stored in RAM.
// The admission LRU is sized generously (4x items) since it holds only counters.
const admissionCounts = new LRUCache<string, number>({
max: MEMORY_CACHE_MAX_ITEMS * 4,
});

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);
admissionCounts.delete(cacheKey);
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;
// Fast path: skip the body read entirely if Content-Length already tells us
// the entry is too large. The loader always sets this header.
const cl = parseInt(response.headers.get("content-length") ?? "0");
if (cl > CACHE_MAX_ENTRY_SIZE) return;
const cacheKey = await requestURLSHA1(request);

// Admission filter: only promote to L1 after MEMORY_CACHE_MIN_HITS writes.
const hits = (admissionCounts.get(cacheKey) ?? 0) + 1;
if (hits < MEMORY_CACHE_MIN_HITS) {
admissionCounts.set(cacheKey, hits);
return;
}
// Key has earned its place — remove from admission tracker and store in RAM.
admissionCounts.delete(cacheKey);

const body = new Uint8Array(await response.arrayBuffer());
if (body.length > CACHE_MAX_ENTRY_SIZE) return;
const headers: [string, string][] = [...response.headers.entries()];
store.set(cacheKey, { body, headers, status: response.status });
Comment thread
coderabbitai[bot] marked this conversation as resolved.
},
});
},
};

return caches;
}

export const caches = createInMemoryCache();
6 changes: 5 additions & 1 deletion runtime/caches/mod.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ import { caches as lruCache } from "./lrucache.ts";

import { caches as fileSystem } from "./fileSystem.ts";

import { caches as inMemoryCache } from "./inMemoryCache.ts";

export const ENABLE_LOADER_CACHE: boolean =
Deno.env.get("ENABLE_LOADER_CACHE") !== "false";
const DEFAULT_CACHE_ENGINE = "CACHE_API";
Expand All @@ -38,7 +40,9 @@ export const cacheImplByEngine: Record<CacheEngine, CacheStorageOption> = {
isAvailable: typeof globalThis.caches !== "undefined",
},
FILE_SYSTEM: {
implementation: headersCache(lruCache(fileSystem)),
implementation: headersCache(
lruCache(createTieredCache(inMemoryCache, fileSystem)),
),
isAvailable: isFileSystemAvailable,
},
REDIS: {
Expand Down
Loading
Loading