LAB-1765: turn cachekit compression on for /api/v2 cached bodies

worker/src/cache.ts

@@ -9,7 +9,9 @@
 // (read-heavy public APIs); the Cache API backend is a per-colo
 // point-of-presence tier, which is the right shape here because the
 // authoritative copy is D1 itself and every entry is re-derivable. No
-// encryption/compression/SaaS tier on this pass (ticket non-goals).
+// encryption/SaaS tier: the data is public and re-derivable, and there is one
+// SDK and one worker, so neither buys anything here. Compression IS on
+// (LAB-1765) — see cacheInstance().
 //
 // Key properties (also the abuse posture):
 // - Canonical keys: alias params collapse onto their canonical name and time
@@ -68,8 +70,10 @@ export const CLOSED_WINDOW_TTL_SECONDS = 86400;
 export const GENERATORS_TTL_SECONDS = 3600;
 
 // Bump to invalidate every existing entry on a key-format, CachedResponse
-// shape, or policy change — get<CachedResponse> is a blind cast, so a shape
-// change without a bump would deserialize stale entries with missing fields.
+// shape, or policy change. isCachedResponse() below is the backstop for a
+// forgotten bump, not a substitute for one: it turns a mismatched entry into a
+// miss, which is correct but costs a D1 refill on every request until the old
+// entries lapse.
 // k2: LAB-1696 changed aggregate values at resolution 3600/86400 (rollup
 // path: full-bucket edge semantics, global-denominator means) — pre-rollup
 // entries must not survive the cutover.
@@ -82,7 +86,23 @@ export const GENERATORS_TTL_SECONDS = 3600;
 // publishes no emission factor for any station — a field missing from an old
 // entry and a published `null` are indistinguishable to the consumer's
 // `== null` check (a numeric 0 is a real published factor and unaffected).
-const KEY_VERSION = 'k4';
+// k5: LAB-1765 turned compression on, which changes the STORED BYTES, not just
+// the shape: entries are now a ByteStorage envelope (LZ4 + xxHash3-64) instead
+// of bare MessagePack. (Originally numbered k4 on this branch; renumbered on
+// merge because master's LAB-1702 bump took k4 first — its entries are live
+// and uncompressed, so the compressed reader must not share their version.)
+// The envelope reader cannot decode a plain k4 entry, and
+// createCache.minimal sets degradation:false, so the get THROWS — survivable
+// (handleApiCached catches, logs and serves from D1, and the refill overwrites
+// the entry) but it would fire once per live key across the whole deploy. A
+// version bump buys the same refill at the same cost without the error storm.
+// A REVERT needs a bump too, and for a worse reason: the envelope is itself
+// valid positional MessagePack, so a compression-off reader DECODES a k5 entry
+// successfully and hands the envelope tuple back as the value — not a miss
+// (cachekit-ts finding, LAB-1388; the tolerant read path that fixes it is
+// merged upstream but unreleased as of 0.1.5). That is what isCachedResponse()
+// catches: without it, the tuple's absent `body` was served as an empty 200.
+const KEY_VERSION = 'k5';
 
 /**
  * Seconds until this entry must expire so it never outlives the data:
@@ -102,6 +122,10 @@ interface CacheEntry {
   ttl: number;
 }
 
+// Exported so tests bind to the SAME caps as production — a test-local copy
+// silently stops testing the real config the moment either number moves.
+export const SERIALIZER_LIMITS = { maxEncodedSize: 64 * 1024 * 1024, maxDecodedSize: 64 * 1024 * 1024 };
+
 // One cache per isolate, per the SDK's own guidance — per-request creation
 // leaks wasm allocations on hot isolates. Lazy so module init stays
 // side-effect-free. Exported for tests (they must exercise THIS configured
@@ -111,9 +135,42 @@ let cache: WorkersCache | null = null;
 export function cacheInstance(): WorkersCache {
   cache ??= createCache.minimal({
     backend: workersCacheAPI(),
-    // Ticket non-goal, and the values are served-as-is JSON — skip the wasm
-    // ByteStorage envelope (cachekit defaults compression ON).
-    compression: false,
+    // LZ4 + xxHash3-64 ByteStorage envelope, via the wasm32 cachekit-core
+    // build. Bodies here are columnar JSON (repeated key names, long runs of
+    // `null` gaps, comma-separated floats) — the shape LZ4 eats.
+    //
+    // Decompression is INLINE in every cache hit, so it is user-visible TTFB,
+    // not a background cost. Measured end to end through worker.fetch at the
+    // dashboard's default range (LAB-1765, miniflare workerd, hours=24, 10
+    // hits each, compression off -> on). The dashboard fetches exactly three
+    // things (public/app.js), and all three are ~20 KiB:
+    //
+    //   /values/aggregate?group_by=fuel    21 KiB   hit 0.6 -> 1.1 ms
+    //   /dispatch                          20 KiB   hit 0.5 -> 0.5 ms
+    //   /intensity                         17 KiB   hit 0.4 -> 0.5 ms
+    //   /values  (API drill-down — NOT     501 KiB  hit 1.6 -> 3.8 ms
+    //             fetched by the dashboard)
+    //
+    // So a page load pays ~+0.6 ms across all three fetches. The entries where
+    // compression earns anything (0.5-1.6 MB /values responses) are the same
+    // ones that pay the few ms, and the same ones whose miss costs a D1
+    // aggregation over up to 300000 rows — cost and benefit land together.
+    // Stored size falls 2.7x (966 -> 350 KiB at 24 h, 1653 -> 606 KiB at the
+    // 300000-row ceiling), which is the whole point: Cache API storage is free
+    // to us, but colo LRU evicts big objects first, so smaller entries survive.
+    //
+    // Not free, just cheap where users actually are. Revert (with a
+    // KEY_VERSION bump, see above) if wrangler tail shows hit-path CPU near
+    // the limit — these are miniflare numbers on a fast desktop core, so
+    // expect edge hardware to be slower.
+    // The wasm module is a static import in cachekit's Workers runtime, so it
+    // is in the bundle whether or not this is on — verified byte-identical at
+    // 373.44 KiB / 113.30 KiB gz either way. Off was paying for it unused.
+    // Written explicitly rather than left to the default ON: cachekit is
+    // adding a per-backend `compressionDefault` and CacheAPIBackend will
+    // advertise FALSE, so relying on the default would silently un-compress
+    // this cache on a future upgrade.
+    compression: true,
     // cachekit's L1 would repopulate on an L2 hit with ITS default TTL
     // (cache-core get(): ttlSeconds ?? defaultTtl), not the entry's remaining
     // lifetime — an isolate could serve a boundary-TTL entry past the
@@ -125,7 +182,12 @@ export function cacheInstance(): WorkersCache {
     // 1 MiB encode default would throw ValueTooLargeError on exactly the
     // heaviest queries — caught and logged, but silently never cached.
     // 64 MiB covers realistic maxima; anything larger falls through uncached.
-    serializer: { maxEncodedSize: 64 * 1024 * 1024, maxDecodedSize: 64 * 1024 * 1024 },
+    // Compression does NOT relax this: both caps sit on the UNCOMPRESSED side
+    // of the envelope (encode caps the msgpack before pack(), decode caps the
+    // plaintext after unpack()), so the body ceiling is unchanged and only the
+    // stored bytes shrink. Neither cap bounds decompression itself, which is
+    // fine here — every envelope in this cache was written by this worker.
+    serializer: SERIALIZER_LIMITS,
   });
   return cache;
 }
@@ -136,6 +198,29 @@ interface CachedResponse {
   expires: number;
 }
 
+/**
+ * `get<CachedResponse>` is a blind cast, and the ways it can hand back
+ * something else are not hypothetical: a stored-format change without a
+ * KEY_VERSION bump, or a compression flip against a store holding the other
+ * format (the ByteStorage envelope is valid positional MessagePack, so a
+ * plain reader DECODES it and returns the 4-tuple). Unchecked, `body`
+ * undefined becomes `new Response(undefined)` — an empty 200 served with
+ * `x-cache: HIT` and cached client-side for the whole TTL, with nothing
+ * logged. Everything else in this layer fails closed; this is the one path
+ * that failed open, so it gets an explicit shape check and a miss on
+ * anything unrecognised.
+ */
+function isCachedResponse(value: unknown): value is CachedResponse {
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    'body' in value &&
+    typeof value.body === 'string' &&
+    'expires' in value &&
+    typeof value.expires === 'number'
+  );
+}
+
 /**
  * Response headers rebuilt on both paths: the handler's own CORS + JSON
  * headers, Cache-Control carrying the REMAINING lifetime (so a client
@@ -292,7 +377,15 @@ export async function handleApiCached(request: Request, env: Env): Promise<Respo
 
   let hit: CachedResponse | null = null;
   try {
-    hit = await cacheInstance().get<CachedResponse>(entry.key);
+    const stored = await cacheInstance().get(entry.key);
+    if (stored !== null && !isCachedResponse(stored)) {
+      // A decodable entry that is not ours: loud, because the only ways to get
+      // here are a missed KEY_VERSION bump or a stored-format mismatch, and
+      // both are silent everywhere else.
+      console.error(`cache entry ignored, unexpected shape (${entry.key})`);
+    } else {
+      hit = stored;
+    }
   } catch (err) {
     console.error(`cache get failed (${entry.key}):`, err);
   }

worker/test/cache.spec.ts

@@ -1,3 +1,4 @@
+import { ByteStorage, createCache, MessagePackSerializer, workersCacheAPI } from '@cachekit-io/cachekit/workers';
 import { env } from 'cloudflare:test';
 import { beforeEach, describe, expect, it } from 'vitest';
 import {
@@ -8,6 +9,7 @@ import {
   GENERATORS_TTL_SECONDS,
   INGEST_GRACE_SECONDS,
   ROOFTOP_PUBLICATION_GRACE_SECONDS,
+  SERIALIZER_LIMITS,
 } from '../src/cache';
 import { upsertValues } from '../src/ingest';
 import worker from '../src/index';
@@ -18,6 +20,60 @@ const T0 = 1784901600;
 // A boundary-aligned "now" for deterministic policy tests.
 const NOW = T0 + 7 * 86400;
 
+
+/**
+ * A /api/v2/values body at production scale and production SHAPE: the columnar
+ * payload handleValues returns (one shared `timestamps` axis, one aligned
+ * `values` array per generator, nulls for gaps), always 600 DUIDs — so
+ * `buckets` alone sets how close the body sits to the 300000-row MAX_LIMIT
+ * ceiling the handler clips at.
+ *
+ * Values come from an LCG so the compression numbers are not flattered by an
+ * artificial period, and Math.imul is load-bearing for that: plain
+ * `seed * 1103515245` overflows 2^53, destroys the low bits, collapses the
+ * period to ~10k and repeats whole series verbatim, which inflated the
+ * measured ratio by ~10% until the expert panel caught it.
+ *
+ * Shape on top of that is dispatch-like: ~25% of units sit at exactly 0 (real
+ * NEM — most registered DUIDs are offline at any moment), the rest hold their
+ * setpoint across several intervals before moving, which is how dispatch
+ * actually behaves. Still a model, so treat the measured ratio as indicative
+ * of the shape, not as a promise about a specific query.
+ */
+function productionBody(buckets: number): string {
+  const timestamps = Array.from({ length: buckets }, (_, i) => T0 + i * 300);
+  let seed = 0x2545f491;
+  const rand = () => ((seed = (Math.imul(seed, 1103515245) + 12345) & 0x7fffffff) / 0x7fffffff);
+  const series = Array.from({ length: 600 }, (_, s) => {
+    const offline = rand() < 0.25;
+    const values: (number | null)[] = [];
+    let held = Math.round(rand() * 42000); // centi-MW, so values carry 2 decimals
+    for (let i = 0; i < buckets; i++) {
+      if (rand() < 0.02) values.push(null); // ingest gap
+      else if (offline) values.push(0);
+      else {
+        if (rand() < 0.3) held = Math.max(0, held + Math.round((rand() - 0.5) * 6000));
+        values.push(held / 100);
+      }
+    }
+    return {
+      id: s + 1,
+      duid: `${['BW', 'ER', 'LD', 'MP', 'YW'][s % 5]}${String(s).padStart(3, '0')}1`,
+      name: `Generator ${s} Power Station Unit ${(s % 4) + 1}`,
+      fuel: ['Wind', 'Solar', 'Black Coal', 'Natural Gas', 'Water'][s % 5],
+      values,
+    };
+  });
+  return JSON.stringify({
+    start: T0,
+    end: timestamps[buckets - 1],
+    resolution: 300,
+    truncated: false,
+    timestamps,
+    series,
+  });
+}
+
 function entryFor(query: string, now = NOW, host = 'nem-api.test', path = '/api/v2/values') {
   return buildCacheEntry(new URL(`https://${host}${path}${query}`), now);
 }
@@ -273,13 +329,122 @@ describe('handleApiCached — integration', () => {
     expect(hit.headers.get('access-control-allow-origin')).toBe('*');
   });
 
-  it('round-trips multi-MB bodies (default 300000-row responses far exceed cachekit’s 1 MiB encode default)', async () => {
-    // Exercises the REAL configured instance: on cachekit defaults this set
-    // throws ValueTooLargeError and the heaviest queries silently never cache.
+  it('round-trips a ceiling-sized production-shaped body through the configured instance', async () => {
+    // Two things at once: cachekit's 1 MiB encode default would throw
+    // ValueTooLargeError here (heaviest queries silently never cached), and
+    // the LAB-1765 ByteStorage envelope has to survive a real wasm
+    // pack/unpack under vitest-pool-workers. A length check would pass on a
+    // corrupted payload, so compare the whole body — but as a boolean, since
+    // toEqual on a mismatch would try to diff two multi-MB strings and wedge
+    // the run that was supposed to tell us what broke.
     const key = `test:big:${host}`;
-    const body = 'x'.repeat(2 * 1024 * 1024);
+    const body = productionBody(500); // 300000 rows, the MAX_LIMIT ceiling
+    expect(body.length).toBeGreaterThan(1024 * 1024);
     await cacheInstance().set(key, { body, expires: 1 }, { ttl: 60 });
     const hit = await cacheInstance().get<{ body: string; expires: number }>(key);
-    expect(hit?.body.length).toBe(body.length);
+    expect(hit?.body === body).toBe(true);
+    expect(hit?.expires).toBe(1);
+  });
+});
+
+/**
+ * LAB-1765: compression is ON, which changes the STORED BYTES. These pin the
+ * two things that decision rests on — that the envelope actually pays for
+ * itself on this workload, and that a config/entry mismatch in either
+ * direction can never serve a wrong body.
+ */
+describe('compression — ByteStorage envelope', () => {
+  let host: string;
+
+  /** A cache configured exactly like production except compression OFF. */
+  function plainCache() {
+    return createCache.minimal({
+      backend: workersCacheAPI(),
+      compression: false,
+      l1: { enabled: false },
+      serializer: SERIALIZER_LIMITS,
+    });
+  }
+
+  beforeEach(() => {
+    host = `t-${crypto.randomUUID()}.test`;
+  });
+
+  it('shrinks a production-shaped body, and the measurement is the decision record', async () => {
+    // Two shapes: 288 buckets is the DEFAULT query (no params = 24 h, 172800
+    // rows, what the dashboard actually asks for), 500 is 300000 rows exactly
+    // — the MAX_LIMIT ceiling, so the widest body the handler can emit.
+    //
+    // Sizes only. Latency was measured on the flip with the same bodies and
+    // recorded in src/cache.ts and on LAB-1765, then the timing harness was
+    // dropped: miniflare has no production baseline to regress against, so it
+    // would have been 20 multi-MB round trips per CI run asserting nothing.
+    // Re-measure by deploying, not by trusting a number from this file.
+    for (const buckets of [288, 500]) {
+      const value = { body: productionBody(buckets), expires: T0 + 300 };
+      // The exact pipeline cache-core runs — serializer.encode() then
+      // byteStorage.pack() — so these ARE the bytes each config hands the Cache
+      // API, not a proxy for them. This measures the CODEC, not the shipped
+      // config: what pins `compression: true` in src/cache.ts is the pair of
+      // mismatch tests below, which go red on a revert.
+      const plain = new MessagePackSerializer(SERIALIZER_LIMITS).encode(value);
+      const packed = new ByteStorage().pack(plain);
+      const ratio = plain.length / packed.length;
+      console.log(
+        `[LAB-1765] ${buckets} buckets x 600 series: ${(value.body.length / 1024).toFixed(0)} KiB JSON, ` +
+          `stored ${(plain.length / 1024).toFixed(0)} -> ${(packed.length / 1024).toFixed(0)} KiB (${ratio.toFixed(2)}x)`,
+      );
+      // Asserted per shape, not after the loop: hoisting it would leave the
+      // 288 case — the common one — with no assertion at all. The floor sits
+      // just under the measured 2.5x/2.7x, so a codec that stopped working
+      // fails instead of passing on a technicality.
+      expect(ratio).toBeGreaterThan(2.2);
+    }
+  });
+
+  it('survives a pre-flip plain entry — the error storm the KEY_VERSION bump avoids', async () => {
+    // What a plain (pre-k5) entry does to the k5 reader, end to end. unpack cannot read
+    // bare MessagePack and createCache.minimal disables degradation, so the
+    // get THROWS rather than returning null — handleApiCached's own catch is
+    // what keeps the endpoint up, and the refill overwrites the entry. The
+    // `expires` here is far future, so a stale body could only appear if the
+    // entry were actually readable.
+    const url = new URL(`https://${host}/api/v2/generators?duid=BAPS`);
+    const key = buildCacheEntry(url, Math.floor(Date.now() / 1000))!.key;
+    await plainCache().set(key, { body: '{"stale":true}', expires: 2 ** 31 }, { ttl: 60 });
+
+    await expect(cacheInstance().get(key)).rejects.toThrow();
+
+    const res = await worker.fetch(new Request(url.toString()), env);
+    expect(res.status).toBe(200);
+    expect(res.headers.get('x-cache')).toBe('MISS');
+    expect(await res.text()).not.toContain('stale');
+  });
+
+  it('never serves a decodable-but-foreign entry as a body', async () => {
+    // The other direction, and the reason a revert is not a one-line config
+    // change: the ByteStorage envelope is itself valid positional MessagePack,
+    // so a compression-off reader decodes an enveloped entry SUCCESSFULLY and
+    // returns the 4-tuple [compressed, checksum, size, 'msgpack'] as the
+    // value. No throw, nothing for degradation to catch, and `hit.body`
+    // undefined used to become an empty 200 stamped `x-cache: HIT` and cached
+    // client-side for the full TTL.
+    //
+    // Asserted against isCachedResponse rather than against the cachekit bug:
+    // any decodable non-CachedResponse must become a MISS, whatever produced
+    // it (a missed KEY_VERSION bump, a revert, a future shape change). Pinning
+    // the vendor misread instead would have gone red on the upstream fix
+    // (LAB-1388, merged but unreleased) without the guard ever regressing.
+    const url = new URL(`https://${host}/api/v2/generators?duid=BAPS`);
+    const key = buildCacheEntry(url, Math.floor(Date.now() / 1000))!.key;
+    await cacheInstance().set(key, ['not', 'a', 'CachedResponse'], { ttl: 60 });
+
+    const res = await worker.fetch(new Request(url.toString()), env);
+    expect(res.status).toBe(200);
+    expect(res.headers.get('x-cache')).toBe('MISS');
+    // A real refilled body, not the empty 200 the missing guard used to serve.
+    const generators: Array<{ duid: string }> = JSON.parse(await res.text());
+    expect(generators.length).toBeGreaterThan(0);
+    expect(generators[0].duid).toBe('BAPS');
   });
 });