diff --git a/worker/src/cache.ts b/worker/src/cache.ts index 6f3fce3..ff8e050 100644 --- a/worker/src/cache.ts +++ b/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 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` 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(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); } diff --git a/worker/test/cache.spec.ts b/worker/test/cache.spec.ts index 77600bf..3e28dcb 100644 --- a/worker/test/cache.spec.ts +++ b/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'); }); });