diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index d53370ba..bf6c7142 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -22,6 +22,8 @@ repos: entry: pnpm exec eslint --fix --max-warnings 0 language: system files: ^packages/cachekit/src/.*\.(ts|tsx)$ + # eslint.config.js ignores declaration files, and an ignored file passed by name warns. + exclude: \.d\.ts$ pass_filenames: true - id: prettier diff --git a/packages/cachekit/README.md b/packages/cachekit/README.md index e827cccd..e9cc726c 100644 --- a/packages/cachekit/README.md +++ b/packages/cachekit/README.md @@ -870,9 +870,72 @@ The logger may be `async`. If it throws or its promise rejects, the SDK reports that failure, with the original message, to `console.error`; it never reaches the cache operation and never surfaces as an unhandled rejection. -With the CachekitIO backend, the `X-CacheKit-L1-*` telemetry headers are wired -automatically from the cache's live L1/L2 hit and miss counters; pass your own -`metricsProvider` in the backend config to override. Those counters, and the +### Telemetry (CachekitIO only) + +Only the CachekitIO backend sends telemetry, and only to the API it is +configured with; Redis, Memcached, File, Workers KV, the Cache API and an +L1-only cache send nothing anywhere. While telemetry is on, the default during +the closed beta, each CachekitIO request carries these optional headers: + +- the session headers: `X-CacheKit-Session-ID` and `-Session-Start`, and the + cache's hit and miss counters `X-CacheKit-L1-Hits`, `-L2-Hits`, `-Misses` + and `-L1-Hit-Rate`, the share of all counted operations L1 served, + `L1-Hits / (L1-Hits + L2-Hits + Misses)`; +- unreleased: the L1 hit-latency report, `X-CacheKit-L1-Latency-Buckets`, + `-Sum-Ns` and `-One-In`. On Node.js the cache times one call in 1024, + chosen at random before its outcome is known, with `process.hrtime.bigint()`; + each timed call that L1 serves adds its whole duration to a histogram of + sixteen buckets. A report rides the backend's next request, never its health + check, and is never sent twice, not even when Node's `fetch` resends a + request on its own after a `421`. A report is sent only through an `Agent` + exactly as `new Agent()` builds it, which Node's own default dispatcher is: + with any other global dispatcher (`setGlobalDispatcher`), such as a + `RetryAgent`, a `MockAgent`, a proxy, an `Agent` composed with + interceptors, or an `Agent` created with any option set, the SDK cannot + rule out a resend, so the report waits and no latency headers are sent. + A report can be lost: + - when the global dispatcher, or the one an npm undici `fetch` of another + major than Node's reads, is anything but an `Agent` built with defaults; + - when the `Agent` has to queue the request: on Node.js 26, whose undici + probes an `https:` API for HTTP/2, requests made while a new connection + is being set up queue, so reports go missing more often around a cold + start or a reconnect. + + The SDK cannot see resend code in a `fetch` wrapper you install: one that + retries with a new init must not resend a request that carries the + `X-CacheKit-L1-Latency-*` headers. A `fetch` wrapper that spreads the SDK's + options over its own `dispatcher`, such as a proxy, has it overridden on a + request that carries a report, which then goes straight to the configured + API: set a proxy with `setGlobalDispatcher` instead, or turn telemetry off. + Cloudflare Workers, Bun and Deno send no + latency report: Workers' clock does not advance during an in-memory hit, and + the protocol names no clock for the other two. + +The counters are wired automatically when the backend is given as a config +(`backend: { apiKey }` or `createCache.io`); a `cachekitio()` instance passed +as `backend` gets them only from its own `metricsProvider`, and the Workers +entry wires none. The latency report needs no wiring: it works in both forms. + +Turn telemetry off (unreleased) with `telemetry: false` in the CachekitIO +backend config or `createCache.io` options, or with the environment variable +`CACHEKIT_TELEMETRY` set to `false`, `0`, `off` or `no` (case-insensitive; +`true`, `1`, `on` and `yes` mean on, and any other value means off, with one +warning). On Workers the variable is read from `process.env` and from the +Worker's bindings, through `import { env } from 'cloudflare:workers'`. A +variable the SDK cannot read counts as off, with one warning. Telemetry is +therefore off wherever the variable cannot be read, including under the +`disallow_importable_env` compatibility flag, which hides the bindings, on +a workerd that exposes no `Cloudflare.compatibilityFlags` or no importable +`env`, and on Deno without `--allow-env`. Off from either control wins over on from the other, and +both are read once, when the backend is constructed. With telemetry off a +request carries none of the headers above, only `X-CacheKit-L1-Status`, which +the API requires, and the `User-Agent` below. + +```typescript +createCache({ backend: { apiKey: process.env.CACHEKIT_API_KEY!, telemetry: false } }); +``` + +Pass your own `metricsProvider` in the backend config to override the counters. Those counters, and the Prometheus hit and miss counters, count one outcome per read: a plain `get()` or `exists()` counts a hit by the layer that served it, or a miss. A `wrap()` call also counts one outcome: an L1 or L2 hit by the layer that served its value diff --git a/packages/cachekit/bench/ir-seconds.json b/packages/cachekit/bench/ir-seconds.json index 51e929fe..c09028ef 100644 --- a/packages/cachekit/bench/ir-seconds.json +++ b/packages/cachekit/bench/ir-seconds.json @@ -26,5 +26,7 @@ "wasm-encrypted-64kb": 92, "secure-set-get": 73, "wrap-l1-hit": 22, - "wrap-l1-hit-int": 20 + "wrap-l1-hit-int": 20, + "wrap-l1-hit-cachekitio": 22, + "wrap-l1-hit-int-cachekitio": 20 } diff --git a/packages/cachekit/bench/ir-workload.mjs b/packages/cachekit/bench/ir-workload.mjs index 2aa77416..c247c787 100644 --- a/packages/cachekit/bench/ir-workload.mjs +++ b/packages/cachekit/bench/ir-workload.mjs @@ -32,6 +32,14 @@ async function wasm() { } const MASTER_KEY = 'a'.repeat(64); +// Never sent anywhere: the CachekitIO workloads stub fetch. +const FAKE_KEY = 'ck_test_fake-not-a-secret'; // pragma: allowlist secret + +/** The CachekitIO workloads' API: every read misses, every write succeeds. */ +function stubCachekitIO() { + globalThis.fetch = async (_url, init) => + new globalThis.Response(null, { status: init?.method === 'GET' ? 404 : 200 }); +} const KEY = 'bench:getUser:' + 'a'.repeat(64); const ARGS = [42, 'user-profile', { region: 'au', tier: 3 }]; // 1,055 B of msgpack. Keep fixtures under the serializer's 10,000-element @@ -410,6 +418,43 @@ export const WORKLOADS = { 50, 20 ), + // wrap-l1-hit on a CachekitIO backend, configured as `{ apiKey }`, so its + // telemetry is on and one call in L1_LATENCY_ONE_IN is timed into the + // backend's latency report. It sets no option a build before the report + // lacks, so the gate measures it on the base too. fetch is stubbed: the + // warm-up's miss and write never leave the process, and the measured hits + // make no request. + 'wrap-l1-hit-cachekitio': { + n: 10_000, + setup: async () => { + stubCachekitIO(); + const { createCache } = await load('index.js'); + const cache = createCache({ backend: { apiKey: FAKE_KEY }, metrics: false }); + const getUser = cache.wrap(async (id, kind, opts) => ({ id, kind, opts }), { + namespace: 'bench', + ttl: 3600, + }); + return async () => { + sink += (await getUser(...ARGS)).id; + }; + }, + }, + // wrap-l1-hit-int on the same CachekitIO backend: the memoised hit, where + // the timing draw is the largest share of a hit + 'wrap-l1-hit-int-cachekitio': { + n: 20_000, + setup: async () => { + stubCachekitIO(); + const { createCache } = await load('index.js'); + const cache = createCache({ backend: { apiKey: FAKE_KEY }, metrics: false }); + const getThing = cache.wrap(async (id) => ({ id }), { namespace: 'bench', ttl: 3600 }); + for (let i = 0; i < 64; i++) await getThing(i); + let i = 0; + return async () => { + sink += (await getThing(i++ & 63)).id; + }; + }, + }, // the README benchmark's hit (bench/runtime/probe.mjs): one integer argument, // cycling over 64 values that are all in L1 'wrap-l1-hit-int': { diff --git a/packages/cachekit/scripts/check-workers-bundle.mjs b/packages/cachekit/scripts/check-workers-bundle.mjs index c00f13ff..c3b6ab68 100644 --- a/packages/cachekit/scripts/check-workers-bundle.mjs +++ b/packages/cachekit/scripts/check-workers-bundle.mjs @@ -49,6 +49,8 @@ try { conditions: ['workerd'], loader: { '.wasm': 'binary' }, plugins: [guardPlugin], + // workerd's built-in modules, as wrangler leaves them. + external: ['cloudflare:*'], write: false, logLevel: 'silent', }); diff --git a/packages/cachekit/src/backends/cachekitio-factory.ts b/packages/cachekit/src/backends/cachekitio-factory.ts index f22d9e2b..57628553 100644 --- a/packages/cachekit/src/backends/cachekitio-factory.ts +++ b/packages/cachekit/src/backends/cachekitio-factory.ts @@ -2,6 +2,7 @@ import type { Backend, CachekitIOBackendConfig, LockableBackend, TTLBackend } fr import { CachekitIOCore } from './cachekitio.js'; import { LockableCachekitIO } from './cachekitio-lockable.js'; import { TTLCachekitIO } from './cachekitio-ttl.js'; +import type { L1LatencyReport } from './telemetry.js'; export function cachekitio(config: CachekitIOBackendConfig): Backend { return new CachekitIOCore(config); @@ -35,6 +36,10 @@ class CachekitIO get transformsKeys(): boolean | undefined { return this.lockable.transformsKeys; } + /** @internal Forwarded like keyPrefix — see Backend.l1Latency. */ + get l1Latency(): L1LatencyReport | undefined { + return this.lockable.l1Latency; + } get(key: string) { return this.lockable.get(key); } diff --git a/packages/cachekit/src/backends/cachekitio-lockable.ts b/packages/cachekit/src/backends/cachekitio-lockable.ts index 83c4a15e..e173c423 100644 --- a/packages/cachekit/src/backends/cachekitio-lockable.ts +++ b/packages/cachekit/src/backends/cachekitio-lockable.ts @@ -2,6 +2,7 @@ import type { Backend, LockableBackend } from './types.js'; import { CachekitIOCore, encodeKey } from './cachekitio.js'; import { BackendError, TimeoutError } from '../errors.js'; import { classifyHttpError, classifyNetworkError } from './error-classifier.js'; +import type { L1LatencyReport } from './telemetry.js'; /** * Lock capability token travels in this request header, never the query string @@ -24,6 +25,10 @@ export class LockableCachekitIO get transformsKeys(): boolean | undefined { return this.inner.transformsKeys; } + /** @internal Forwarded like keyPrefix — see Backend.l1Latency. */ + get l1Latency(): L1LatencyReport | undefined { + return this.inner.l1Latency; + } get(key: string) { return this.inner.get(key); diff --git a/packages/cachekit/src/backends/cachekitio-ttl.ts b/packages/cachekit/src/backends/cachekitio-ttl.ts index bd7a3004..6410b1e8 100644 --- a/packages/cachekit/src/backends/cachekitio-ttl.ts +++ b/packages/cachekit/src/backends/cachekitio-ttl.ts @@ -2,6 +2,7 @@ import type { Backend, TTLBackend } from './types.js'; import { CachekitIOCore, validateTtl, encodeKey } from './cachekitio.js'; import { BackendError, TimeoutError } from '../errors.js'; import { classifyHttpError, classifyNetworkError } from './error-classifier.js'; +import type { L1LatencyReport } from './telemetry.js'; // Required<…getWithTtl | setWithStaleTtl>: tsc, not only tests, enforces the forwards. export class TTLCachekitIO @@ -16,6 +17,10 @@ export class TTLCachekitIO get transformsKeys(): boolean | undefined { return this.inner.transformsKeys; } + /** @internal Forwarded like keyPrefix — see Backend.l1Latency. */ + get l1Latency(): L1LatencyReport | undefined { + return this.inner.l1Latency; + } get(key: string) { return this.inner.get(key); diff --git a/packages/cachekit/src/backends/cachekitio.test.ts b/packages/cachekit/src/backends/cachekitio.test.ts index e285e35b..119d9c47 100644 --- a/packages/cachekit/src/backends/cachekitio.test.ts +++ b/packages/cachekit/src/backends/cachekitio.test.ts @@ -443,9 +443,9 @@ describe('CachekitIO Backend', () => { await cache.close(); }); - // The SaaS telemetry headers are auto-wired from the cache's + // The X-CacheKit-L1-* counter headers are auto-wired from the cache's // live hit/miss counters — no user-supplied metricsProvider needed. - it('auto-wires X-CacheKit-L1-* telemetry headers from live cache counters', async () => { + it('auto-wires the X-CacheKit-L1-* counter headers from live cache counters', async () => { const { createCache } = await import('../cache.js'); const cache = createCache({ diff --git a/packages/cachekit/src/backends/cachekitio.ts b/packages/cachekit/src/backends/cachekitio.ts index eeb85f46..77f67676 100644 --- a/packages/cachekit/src/backends/cachekitio.ts +++ b/packages/cachekit/src/backends/cachekitio.ts @@ -3,6 +3,8 @@ import { BackendError, ConfigurationError, TimeoutError } from '../errors.js'; import { DEFAULT_TTL_SECONDS } from '../constants.js'; import { getSessionHeaders } from './session.js'; import { buildMetricsHeaders } from './metrics-headers.js'; +import { L1_LATENCY_ONE_IN, L1LatencyReport, resolveTelemetry } from './telemetry.js'; +import { oneAttemptDispatcher } from './one-attempt-dispatcher.js'; import { classifyHttpError, classifyNetworkError } from './error-classifier.js'; import { validateCachekitUrl } from './url-validator.js'; import { USER_AGENT } from './user-agent.js'; @@ -163,6 +165,14 @@ export class CachekitIOCore implements Backend { private readonly defaultTtl: number; private readonly timeout: number; private readonly metricsProvider?: () => import('./types.js').L1Metrics | null; + /** The telemetry switch, resolved once at construction (API-95). */ + private readonly telemetry: boolean; + /** + * This backend's one L1 hit-latency accumulator (API-83 to API-86); absent + * while telemetry is off, so nothing is timed into it. See Backend.l1Latency. + * @internal + */ + readonly l1Latency?: L1LatencyReport; private closed = false; constructor(config: CachekitIOBackendConfig) { @@ -175,6 +185,8 @@ export class CachekitIOCore implements Backend { this.defaultTtl = validateTtl(config.defaultTtl ?? DEFAULT_TTL_SECONDS); this.timeout = config.timeout ?? DEFAULT_TIMEOUT_MS; this.metricsProvider = config.metricsProvider; + this.telemetry = resolveTelemetry(config.telemetry); + if (this.telemetry) this.l1Latency = new L1LatencyReport(L1_LATENCY_ONE_IN); } async get(key: string): Promise { @@ -300,7 +312,10 @@ export class CachekitIOCore implements Backend { async health(): Promise<{ healthy: boolean; latencyMs: number; version?: string }> { const start = performance.now(); try { - const response = await this.request('GET', `${this.apiUrl}/v1/cache/health`); + // No latency report: the server does not record health checks (API-84). + const response = await this.request('GET', `${this.apiUrl}/v1/cache/health`, { + latencyReport: false, + }); const latencyMs = performance.now() - start; if (response.status === 429) return { healthy: false, latencyMs }; if (!response.ok) return { healthy: false, latencyMs }; @@ -356,23 +371,34 @@ export class CachekitIOCore implements Backend { return `${this.apiUrl}/v1/cache/${encodeKey(key)}`; } + /** + * One attempt. With telemetry off it carries no optional metrics header + * (API-96). A latency report is taken here, on every call, so each attempt + * the cache's retries make takes its own: the one it takes is consumed + * whatever the outcome, and never re-sent (API-86), not even by the resend + * fetch makes on its own after a 421 (see takeLatencyReport). The headers + * go only to this backend's configured API, and a redirect is never + * followed (API-77). + */ private async request( method: string, url: string, - opts?: { body?: Uint8Array; headers?: Record } + opts?: { body?: Uint8Array; headers?: Record; latencyReport?: false } ): Promise { + const report = opts?.latencyReport === false ? undefined : this.takeLatencyReport(); const headers: Record = { 'User-Agent': USER_AGENT, Authorization: `Bearer ${this.apiKey}`, - ...getSessionHeaders(), - ...buildMetricsHeaders(this.metricsProvider), + ...(this.telemetry && getSessionHeaders()), + ...buildMetricsHeaders(this.metricsProvider, this.telemetry), + ...report?.headers, ...opts?.headers, }; if (opts?.body) { headers['Content-Type'] = opts.headers?.['Content-Type'] ?? 'application/octet-stream'; } - return fetch(url, { + const init: RequestInit = { method, headers, body: opts?.body, @@ -382,7 +408,27 @@ export class CachekitIOCore implements Backend { // Not 'error': Node raises that as a network failure, which is retried. redirect: 'manual', signal: AbortSignal.timeout(this.timeout), - }); + }; + // Typed loosely: the dispatcher implements the one method of undici's + // Dispatcher that fetch calls, and the package does not depend on undici. + if (report) (init as { dispatcher?: object }).dispatcher = report.dispatcher; + return fetch(url, init); + } + + /** + * This backend's latency report for one request, with the fetch dispatcher + * that strips it from any resend of that request (one-attempt-dispatcher.ts). + * Undefined, leaving the report in place, while there is nothing to report + * (API-85) or no such dispatcher: on any runtime but Node, which alone + * times L1 hits, and on Node while its global dispatcher is anything but + * undici's own Agent, which might resend the request. + */ + private takeLatencyReport(): { headers: Record; dispatcher: object } | undefined { + const report = this.l1Latency; + if (report === undefined || report.empty) return undefined; + const dispatcher = oneAttemptDispatcher(); + if (dispatcher === undefined) return undefined; + return { headers: report.take()!, dispatcher }; } private async httpError(operation: string, response: Response): Promise { diff --git a/packages/cachekit/src/backends/metrics-headers.test.ts b/packages/cachekit/src/backends/metrics-headers.test.ts index 36368872..0b144bf6 100644 --- a/packages/cachekit/src/backends/metrics-headers.test.ts +++ b/packages/cachekit/src/backends/metrics-headers.test.ts @@ -3,29 +3,35 @@ import { buildMetricsHeaders } from './metrics-headers.js'; describe('Metrics Headers', () => { it('returns disabled when no provider given', () => { - const h = buildMetricsHeaders(); + const h = buildMetricsHeaders(undefined, true); expect(h).toEqual({ 'X-CacheKit-L1-Status': 'disabled' }); }); it('returns disabled when provider returns null', () => { - const h = buildMetricsHeaders(() => null); + const h = buildMetricsHeaders(() => null, true); expect(h).toEqual({ 'X-CacheKit-L1-Status': 'disabled' }); }); it('returns disabled when provider throws', () => { const h = buildMetricsHeaders(() => { throw new Error('boom'); - }); + }, true); expect(h).toEqual({ 'X-CacheKit-L1-Status': 'disabled' }); }); it('returns disabled when l1Enabled is false', () => { - const h = buildMetricsHeaders(() => ({ l1Hits: 5, l2Hits: 3, misses: 2, l1Enabled: false })); + const h = buildMetricsHeaders( + () => ({ l1Hits: 5, l2Hits: 3, misses: 2, l1Enabled: false }), + true + ); expect(h).toEqual({ 'X-CacheKit-L1-Status': 'disabled' }); }); it('calculates correct hit rate', () => { - const h = buildMetricsHeaders(() => ({ l1Hits: 3, l2Hits: 2, misses: 5, l1Enabled: true })); + const h = buildMetricsHeaders( + () => ({ l1Hits: 3, l2Hits: 2, misses: 5, l1Enabled: true }), + true + ); expect(h['X-CacheKit-L1-Status']).toBe('miss'); expect(h['X-CacheKit-L1-Hits']).toBe('3'); expect(h['X-CacheKit-L2-Hits']).toBe('2'); @@ -33,8 +39,31 @@ describe('Metrics Headers', () => { expect(h['X-CacheKit-L1-Hit-Rate']).toBe('0.300'); }); - it('handles zero-division with 0.000', () => { - const h = buildMetricsHeaders(() => ({ l1Hits: 0, l2Hits: 0, misses: 0, l1Enabled: true })); - expect(h['X-CacheKit-L1-Hit-Rate']).toBe('0.000'); + it('computes L1-Hit-Rate as L1-Hits / (L1-Hits + L2-Hits + Misses), never the share of hits (API-78)', () => { + const rate = (l1Hits: number, l2Hits: number, misses: number) => + buildMetricsHeaders(() => ({ l1Hits, l2Hits, misses, l1Enabled: true }), true)[ + 'X-CacheKit-L1-Hit-Rate' + ]; + // 2 / (2 + 1 + 1), not 2 / (2 + 1). + expect(rate(2, 1, 1)).toBe('0.500'); + expect(rate(1, 0, 2)).toBe('0.333'); + expect(rate(2, 0, 1)).toBe('0.667'); + expect(rate(5, 0, 0)).toBe('1.000'); + expect(rate(0, 0, 0)).toBe('0.000'); + for (const [l1, l2, m] of [ + [1, 2, 3], + [999, 0, 1], + [1, 999_999, 0], + ]) { + expect(rate(l1, l2, m)).toMatch(/^[01]\.[0-9]{3}$/); + } + }); + + it('sends only L1-Status while telemetry is off (API-96)', () => { + const h = buildMetricsHeaders( + () => ({ l1Hits: 3, l2Hits: 2, misses: 5, l1Enabled: true }), + false + ); + expect(h).toEqual({ 'X-CacheKit-L1-Status': 'miss' }); }); }); diff --git a/packages/cachekit/src/backends/metrics-headers.ts b/packages/cachekit/src/backends/metrics-headers.ts index b61e263d..35e5a5bd 100644 --- a/packages/cachekit/src/backends/metrics-headers.ts +++ b/packages/cachekit/src/backends/metrics-headers.ts @@ -1,6 +1,14 @@ import type { L1Metrics } from './types.js'; -export function buildMetricsHeaders(provider?: () => L1Metrics | null): Record { +/** + * `X-CacheKit-L1-Status`, plus the session counters and hit rate while + * `telemetry` is on. With it off only L1-Status goes, which API-1 requires on + * every `ck_sdk_` request (API-96). + */ +export function buildMetricsHeaders( + provider: (() => L1Metrics | null) | undefined, + telemetry: boolean +): Record { const disabled = { 'X-CacheKit-L1-Status': 'disabled' }; if (!provider) return disabled; let stats: L1Metrics | null; @@ -10,6 +18,8 @@ export function buildMetricsHeaders(provider?: () => L1Metrics | null): Record 0 ? stats.l1Hits / total : 0; return { diff --git a/packages/cachekit/src/backends/one-attempt-dispatcher.test.ts b/packages/cachekit/src/backends/one-attempt-dispatcher.test.ts new file mode 100644 index 00000000..97fbef66 --- /dev/null +++ b/packages/cachekit/src/backends/one-attempt-dispatcher.test.ts @@ -0,0 +1,185 @@ +/** + * The dispatcher that keeps a latency report to one attempt (protocol + * spec/saas-api.md, API-86). telemetry.test.ts checks it on the wire, with + * this Node's own fetch and genuine undici dispatchers; this pins, for each + * undici Node bundles, which global slot it reads, which Agents it accepts, + * and what each attempt is handed. + */ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { oneAttemptDispatcher } from './one-attempt-dispatcher.js'; +import { npmUndici } from '../../test/fixtures/npm-undici.js'; + +const LEGACY = Symbol.for('undici.globalDispatcher.1'); +const CURRENT = Symbol.for('undici.globalDispatcher.2'); +let saved: unknown[] = []; + +/** Put dispatchers in undici's two global slots, as undici 6 to 8 fill them. */ +function install(legacy: unknown, current: unknown): void { + Reflect.set(globalThis, LEGACY, legacy); + Reflect.set(globalThis, CURRENT, current); +} + +const HEADERS = { + Authorization: 'Bearer x', + 'X-CacheKit-L1-Status': 'miss', + 'X-CacheKit-L1-Latency-Buckets': '0,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0', + 'X-CacheKit-L1-Latency-Sum-Ns': '300', + 'X-CacheKit-L1-Latency-One-In': '1024', +}; +const WITHOUT_REPORT = { Authorization: 'Bearer x', 'X-CacheKit-L1-Status': 'miss' }; +const UNDICI_8 = { version: '8.12.2', handler: { onRequestStart() {} }, slot: 'current' }; +const UNDICI_7 = { version: '7.30.1', handler: { onConnect() {} }, slot: 'legacy' }; + +/** + * What each dispatch was handed: the headers as read then, the object itself, + * and whether the attempt was marked idempotent. + */ +type Log = { name: string; read: unknown; held: unknown; idempotent?: boolean }[]; + +// Where undici 6 to 8 keep an Agent's options and factory. +const OPTIONS = Symbol('options'); +const FACTORY = Symbol('factory'); + +/** Shaped like undici's own Agent as `new Agent()` builds it. */ +class Agent { + [OPTIONS] = { maxOrigins: Infinity, connect: undefined, tls: undefined }; + [FACTORY] = function defaultFactory(): void {}; + constructor( + private readonly name: string, + private readonly log: Log + ) {} + dispatch(options: { headers?: unknown; idempotent?: boolean }): boolean { + this.log.push({ + name: this.name, + read: { ...(options.headers as object) }, + held: options.headers, + idempotent: options.idempotent, + }); + return true; + } +} + +class RetryAgent extends Agent {} + +beforeEach(() => { + void globalThis.Headers; // loads Node's undici, which fills its global slots + saved = [Reflect.get(globalThis, LEGACY), Reflect.get(globalThis, CURRENT)]; +}); + +afterEach(() => install(saved[0], saved[1])); + +describe('oneAttemptDispatcher', () => { + it.each([ + ['undici 8 (onRequestStart)', UNDICI_8], + ['undici 6 and 7 (onConnect)', UNDICI_7], + ])( + 'under %s, hands the report to the slot its fetch reads, on the first attempt only', + (_, { version, handler, slot }) => { + const log: Log = []; + install(new Agent('legacy', log), new Agent('current', log)); + const dispatcher = oneAttemptDispatcher(version)!; + dispatcher.dispatch({ headers: HEADERS }, handler); + dispatcher.dispatch({ headers: HEADERS }, handler); + // The first attempt is not idempotent, so no client could pipeline it. + expect(log.map(({ name, read, idempotent }) => [name, read, idempotent])).toEqual([ + [slot, HEADERS, false], + [slot, WITHOUT_REPORT, undefined], + ]); + } + ); + + it('takes the report out of the headers it handed on, once dispatch returns', () => { + const log: Log = []; + install(new Agent('legacy', log), new Agent('current', log)); + oneAttemptDispatcher(UNDICI_8.version)!.dispatch({ headers: HEADERS }, UNDICI_8.handler); + // A layer that kept the options to resend them, as undici's retry handler + // does, reads them again without the report. + expect(log[0].read).toEqual(HEADERS); + expect(log[0].held).toEqual(WITHOUT_REPORT); + expect(HEADERS['X-CacheKit-L1-Latency-Sum-Ns']).toBe('300'); + }); + + it.each([UNDICI_8, UNDICI_7])( + 'makes none, and the report waits, unless the slot fetch reads holds undici’s Agent ($version)', + ({ version }) => { + const log: Log = []; + install(new RetryAgent('legacy', log), new RetryAgent('current', log)); + expect(oneAttemptDispatcher(version)).toBeUndefined(); + } + ); + + it('under undici 8, never falls back to the first slot, which holds the legacy adapter', () => { + install(new Agent('legacy', []), undefined); + expect(oneAttemptDispatcher(UNDICI_8.version)).toBeUndefined(); + }); + + it.each([ + ['undici’s Agent: with the report', Agent, HEADERS], + ['anything else: without it', RetryAgent, WITHOUT_REPORT], + ])("sends another fetch's handler to the slot that fetch reads, holding %s", (_, Kind, sent) => { + // npm undici's fetch, of another major than Node's, on a global slot of its own. + const log: Log = []; + install(new Kind('legacy', log), new Agent('current', log)); + oneAttemptDispatcher(UNDICI_8.version)!.dispatch({ headers: HEADERS }, UNDICI_7.handler); + expect(log.map(({ name, read }) => [name, read])).toEqual([['legacy', sent]]); + }); + + describe('only an Agent built with defaults gets a report', () => { + it("accepts this Node's own default dispatcher", () => { + expect(oneAttemptDispatcher()).toBeDefined(); + }); + + it.each([UNDICI_8, UNDICI_7])('accepts a genuine `new Agent()` ($version)', ({ version }) => { + const agent = new npmUndici.Agent(); + install(agent, agent); + expect(oneAttemptDispatcher(version)).toBeDefined(); + }); + + it.each([ + ['allowH2', { allowH2: true }], + ['connect.allowH2', { connect: { allowH2: true } }], + ['a connect function', { connect: () => undefined }], + [ + 'a factory', + { factory: (origin: string, options: object) => new npmUndici.Pool(origin, options) }, + ], + ['pipelining', { pipelining: 2 }], + ['a CA only', { connect: { ca: 'x' } }], + ])('refuses a genuine Agent created with %s, on either slot', (_, options) => { + const agent = new npmUndici.Agent(options); + install(agent, agent); + expect(oneAttemptDispatcher(UNDICI_8.version)).toBeUndefined(); + expect(oneAttemptDispatcher(UNDICI_7.version)).toBeUndefined(); + }); + + it.each([ + ['its options', OPTIONS], + ['its factory', FACTORY], + ])('refuses an Agent whose %s it cannot read', (_, field) => { + const agent = new Agent('current', []); + delete (agent as unknown as Record)[field]; + install(agent, agent); + expect(oneAttemptDispatcher(UNDICI_8.version)).toBeUndefined(); + }); + }); + + it('refuses headers it cannot read, rather than send a report it cannot remove', () => { + const log: Log = []; + install(new Agent('legacy', log), new Agent('current', log)); + const raw = Object.entries(HEADERS).flat(); + expect(() => + oneAttemptDispatcher(UNDICI_8.version)!.dispatch({ headers: raw }, UNDICI_8.handler) + ).toThrow(/cannot read the headers/); + expect(log).toHaveLength(0); + }); + + it('treats a slot holding something other than a dispatcher as empty', () => { + install({ dispatch: 'not a function' }, 42); + expect(oneAttemptDispatcher(UNDICI_8.version)).toBeUndefined(); + expect(oneAttemptDispatcher(UNDICI_7.version)).toBeUndefined(); + }); + + it('makes none off Node, where there is no undici', () => { + expect(oneAttemptDispatcher(null)).toBeUndefined(); + }); +}); diff --git a/packages/cachekit/src/backends/one-attempt-dispatcher.ts b/packages/cachekit/src/backends/one-attempt-dispatcher.ts new file mode 100644 index 00000000..cef7d3e4 --- /dev/null +++ b/packages/cachekit/src/backends/one-attempt-dispatcher.ts @@ -0,0 +1,152 @@ +/** + * A latency report rides one attempt (protocol spec/saas-api.md, API-86), and + * on Node a request can be resent below the code that built it: by fetch + * itself after a 421 (the Fetch standard's own retry), by a retrying global + * dispatcher such as undici's RetryAgent, by a retry layer inside an Agent, by + * an HTTP/1.1 client that pipelines and loses its connection, or by undici + * 6's HTTP/2 client, which writes every stream still in flight again when any + * one ends. + * + * So a report is sent only through an Agent exactly as `new Agent()` builds + * it, which Node's own default is: every option left at its default and the + * default factory. Such an Agent neither retries nor pipelines; over HTTP/2, + * which only undici 8 turns on by default, it replays only a stream the + * server refused or, after GOAWAY, never processed, which the spec allows. + * Any other dispatcher, and an Agent with any option set, leaves the report + * waiting: a report may be lost, never sent twice. + * + * The report rides only the first attempt fetch dispatches, marked not + * idempotent so that no client could pipeline it, and in a copy of the + * headers from which it is removed as soon as dispatch returns. Any later + * attempt gets the request without it. What this cannot see, such as a fetch + * wrapper that retries, is in the README's Telemetry section. + * + * Node keeps undici's global dispatcher under the symbols undici registers it + * with. The fetch of undici 6 and 7 reads the first and passes handlers of the + * older API (onConnect); undici 8's reads the second and passes onRequestStart + * handlers, and its first holds an adapter for the older API. Elsewhere there + * is no undici, and nothing here is used: only Node times L1 hits. + */ + +/** The part of undici's Dispatcher that fetch uses. */ +interface Dispatcher { + dispatch(options: { headers?: unknown; idempotent?: boolean }, handler: object): boolean; +} + +const LEGACY = Symbol.for('undici.globalDispatcher.1'); +const CURRENT = Symbol.for('undici.globalDispatcher.2'); +const REPORT_HEADER = /^x-cachekit-l1-latency-/i; + +function isDispatcher(value: unknown): value is Dispatcher { + return ( + typeof value === 'object' && + value !== null && + typeof Reflect.get(value, 'dispatch') === 'function' + ); +} + +/** The dispatcher in one of undici's global slots, if what is there is one. */ +function slot(key: symbol): Dispatcher | undefined { + const value: unknown = Reflect.get(globalThis, key); + return isDispatcher(value) ? value : undefined; +} + +/** The slot for a handler: onRequestStart handlers go to the second only. */ +function slotFor(handler: object): symbol { + return 'onRequestStart' in handler ? CURRENT : LEGACY; +} + +/** An own field of `agent` under a symbol described `name`, as undici 6 to 8 keep them. */ +function privateField(agent: object, name: string): unknown { + const key = Object.getOwnPropertySymbols(agent).find((symbol) => symbol.description === name); + return key === undefined ? undefined : Reflect.get(agent, key); +} + +/** + * Exactly as `new Agent()` builds it. The class is undici's Agent, and its + * dispatch is its prototype's: not a RetryAgent or a MockAgent, nor the proxy + * that compose() returns in undici 7 and 8 (undici 6's is a ComposedDispatcher). + * Its options, under the symbol described 'options', are all undefined but + * `maxOrigins: Infinity` (undici 7 and 8), and its factory, under 'factory', + * is undici's `defaultFactory`. Any option set fails this, an HTTP/2 or + * pipelining one among them, as does undici moving either field. + */ +function deliversOnce(dispatcher: Dispatcher): boolean { + const prototype: unknown = Object.getPrototypeOf(dispatcher); + const options = privateField(dispatcher, 'options'); + const factory = privateField(dispatcher, 'factory'); + return ( + dispatcher.constructor?.name === 'Agent' && + typeof prototype === 'object' && + prototype !== null && + Reflect.get(dispatcher, 'dispatch') === Reflect.get(prototype, 'dispatch') && + typeof factory === 'function' && + factory.name === 'defaultFactory' && + typeof options === 'object' && + options !== null && + Object.entries(options).every( + ([name, value]) => value === undefined || (name === 'maxOrigins' && value === Infinity) + ) + ); +} + +/** A copy of the headers fetch dispatches: the plain object undici 6 to 8 build. */ +function copyOf(headers: unknown): Record { + if (typeof headers !== 'object' || headers === null || Array.isArray(headers)) { + throw new Error('cachekit: cannot read the headers of a request carrying a latency report'); + } + return { ...headers }; +} + +function dropReport(headers: Record): Record { + for (const name of Object.keys(headers)) { + if (REPORT_HEADER.test(name)) delete headers[name]; + } + return headers; +} + +function nodeUndiciVersion(): unknown { + return typeof process !== 'undefined' ? process.versions?.undici : undefined; +} + +/** + * A fetch `dispatcher` for one request that carries a latency report, or + * undefined, and the report waits, unless the global dispatcher Node's fetch + * reads is an Agent built with defaults. `undiciVersion` is the undici Node + * bundles, which decides the slot its fetch reads; tests pass one. + */ +export function oneAttemptDispatcher( + undiciVersion: unknown = nodeUndiciVersion() +): Dispatcher | undefined { + if (typeof undiciVersion !== 'string') return undefined; + const key = Number.parseInt(undiciVersion, 10) >= 8 ? CURRENT : LEGACY; + // Node loads undici on the first use of fetch or one of its classes, and + // undici fills its slots as it loads. + if (slot(key) === undefined) void globalThis.Headers; + const target = slot(key); + if (target === undefined || !deliversOnce(target)) return undefined; + let dispatched = false; + return { + dispatch(options, handler) { + // A handler of the other API comes from a fetch other than Node's own, + // such as npm undici's: it goes to the slot that fetch reads. + const handlerSlot = slotFor(handler); + const own = handlerSlot === key; + const to = own ? target : slot(handlerSlot); + if (to === undefined) throw new Error('cachekit: no global dispatcher for this request'); + const headers = copyOf(options.headers); + const first = !dispatched; + dispatched = true; + if (!first || !(own || deliversOnce(to))) { + return to.dispatch({ ...options, headers: dropReport(headers) }, handler); + } + try { + return to.dispatch({ ...options, headers, idempotent: false }, handler); + } finally { + // The attempt has read them by now, or, if a full pool queued it, + // will go without the report. + dropReport(headers); + } + }, + }; +} diff --git a/packages/cachekit/src/backends/telemetry.test.ts b/packages/cachekit/src/backends/telemetry.test.ts new file mode 100644 index 00000000..eab0dde6 --- /dev/null +++ b/packages/cachekit/src/backends/telemetry.test.ts @@ -0,0 +1,580 @@ +/** + * The telemetry switch and the L1 hit-latency report on the CachekitIO + * backend (protocol spec/saas-api.md § Optional Metrics Headers). Each test + * names the requirement it pins. + */ +import { once } from 'node:events'; +import { createServer } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + cachekitio, + cachekitioFull, + cachekitioWithLocking, + cachekitioWithTTL, +} from './cachekitio-factory.js'; +import { CachekitIOCore } from './cachekitio.js'; +import { + L1_LATENCY_ONE_IN, + L1LatencyReport, + _resetTelemetryWarningForTesting, + resolveTelemetry, + setTelemetryBindingsReader, +} from './telemetry.js'; +import { ConfigurationError } from '../errors.js'; +import { setLogger } from '../logger.js'; +import { npmUndici, type Closable } from '../../test/fixtures/npm-undici.js'; + +const API_URL = 'https://api.test.cachekit.io'; +const API_KEY = 'ck_test_fake-not-a-secret'; // pragma: allowlist secret +const CONFIG = { apiKey: API_KEY, apiUrl: API_URL, allowCustomHost: true }; + +const SESSION_AND_COUNTERS = [ + 'X-CacheKit-Session-ID', + 'X-CacheKit-Session-Start', + 'X-CacheKit-L1-Hits', + 'X-CacheKit-L2-Hits', + 'X-CacheKit-Misses', + 'X-CacheKit-L1-Hit-Rate', +]; +const LATENCY = [ + 'X-CacheKit-L1-Latency-Buckets', + 'X-CacheKit-L1-Latency-Sum-Ns', + 'X-CacheKit-L1-Latency-One-In', +]; +const OPTIONAL_METRICS = [...SESSION_AND_COUNTERS, ...LATENCY]; + +const COUNTERS = { l1Hits: 3, l2Hits: 1, misses: 1, l1Enabled: true }; + +let fetchSpy: ReturnType; +const nodeFetch = globalThis.fetch; + +const undici = npmUndici; +const RETRY_ONCE = { maxRetries: 1, minTimeout: 1, maxTimeout: 1 }; +const DISPATCHER_SLOTS = [ + Symbol.for('undici.globalDispatcher.1'), + Symbol.for('undici.globalDispatcher.2'), +]; + +/** Run `body`, then put undici's global dispatcher slots back as they were. */ +async function keepingGlobalDispatcher(body: () => Promise): Promise { + void globalThis.Headers; // loads Node's undici, which sets its global dispatcher + const saved: unknown[] = DISPATCHER_SLOTS.map((key) => Reflect.get(globalThis, key)); + try { + await body(); + } finally { + DISPATCHER_SLOTS.forEach((key, i) => Reflect.set(globalThis, key, saved[i])); + } +} + +/** + * Answer `statuses` in turn from a real HTTP server, with `dispatcher` as + * undici's global dispatcher if given, and run `body` with Node's own fetch + * routed there. Only the destination changes: the backend's options, its + * dispatcher included, reach fetch unchanged. Returns what each attempt on the + * wire carried. + */ +async function onTheWire( + statuses: number[], + body: () => Promise, + dispatcher?: Closable +): Promise<{ sumNs?: string; authorization?: string }[]> { + const seen: { sumNs?: string; authorization?: string }[] = []; + const server = createServer((request, response) => { + seen.push({ + sumNs: request.headers['x-cachekit-l1-latency-sum-ns'] as string | undefined, + authorization: request.headers.authorization, + }); + response.writeHead(statuses[seen.length - 1] ?? 404).end(); + }); + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const { port } = server.address() as AddressInfo; + vi.stubGlobal('fetch', (url: string, init: RequestInit) => + nodeFetch(`http://127.0.0.1:${port}${new URL(url).pathname}`, init) + ); + try { + await keepingGlobalDispatcher(async () => { + if (dispatcher) undici.setGlobalDispatcher(dispatcher); + await body(); + }); + } finally { + await dispatcher?.close(); + server.closeAllConnections(); + server.close(); + } + return seen; +} + +function sentHeaders(call = -1): Record { + const calls = fetchSpy.mock.calls; + const [, init] = calls[call < 0 ? calls.length + call : call] as [string, RequestInit]; + return init.headers as Record; +} + +function latencyOf(headers: Record) { + return LATENCY.map((name) => headers[name]); +} + +beforeEach(() => { + fetchSpy = vi.fn(async () => new Response(null, { status: 404 })); + vi.stubGlobal('fetch', fetchSpy); + vi.stubEnv('CACHEKIT_TELEMETRY', undefined); + _resetTelemetryWarningForTesting(); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.unstubAllEnvs(); + vi.restoreAllMocks(); + setTelemetryBindingsReader(undefined); + setLogger(null); +}); + +describe('L1LatencyReport encoding (API-87)', () => { + it("encodes the spec's worked example: 90, 300, 300 and 40,000 ns at N = 1", () => { + const report = new L1LatencyReport(1); + for (const ns of [90, 300, 300, 40_000]) report.record(ns); + expect(report.take()).toEqual({ + 'X-CacheKit-L1-Latency-Buckets': '1,0,2,0,0,0,0,0,0,1,0,0,0,0,0,0', + 'X-CacheKit-L1-Latency-Sum-Ns': '40690', + 'X-CacheKit-L1-Latency-One-In': '1', + }); + }); + + it('puts each bound in its own bucket and the next nanosecond in the next', () => { + // Bucket i (from 1) holds up to 2^(i+6) ns; bucket 16 up to 60 s. + const bounds = Array.from({ length: 15 }, (_, i) => 2 ** (i + 7)); + for (const [i, bound] of bounds.entries()) { + const report = new L1LatencyReport(1); + report.record(bound); + report.record(bound + 1); + const buckets = report.take()!['X-CacheKit-L1-Latency-Buckets'].split(',').map(Number); + const expected = new Array(16).fill(0); + expected[i] = 1; + expected[i + 1] = 1; + expect(buckets, `bound ${bound}`).toEqual(expected); + } + }); + + it('counts 0 ns in bucket 1, and 60 s in bucket 16', () => { + const report = new L1LatencyReport(1); + report.record(0); + report.record(60_000_000_000); + expect(report.take()!['X-CacheKit-L1-Latency-Buckets']).toBe('1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1'); + }); + + it('discards a duration over 60 s, counting it nowhere', () => { + const report = new L1LatencyReport(1); + report.record(60_000_000_001); + expect(report.take()).toBeUndefined(); + report.record(7); + expect(report.take()!['X-CacheKit-L1-Latency-Sum-Ns']).toBe('7'); + }); + + it('keeps every value within the grammar: no hit takes the sum past 2^53 - 1', () => { + const report = new L1LatencyReport(1_000_000); + for (let i = 0; i < 200_000; i++) report.record(59_999_999_999); + const headers = report.take()!; + for (const value of [ + ...headers['X-CacheKit-L1-Latency-Buckets'].split(','), + headers['X-CacheKit-L1-Latency-Sum-Ns'], + headers['X-CacheKit-L1-Latency-One-In'], + ]) { + expect(value).toMatch(/^(0|[1-9][0-9]{0,15})$/); + expect(Number(value)).toBeLessThanOrEqual(Number.MAX_SAFE_INTEGER); + } + expect(headers['X-CacheKit-L1-Latency-Buckets'].split(',')).toHaveLength(16); + }); +}); + +describe('L1LatencyReport take (API-85)', () => { + it('takes every value and resets in one step', () => { + const report = new L1LatencyReport(4); + report.record(500); + expect(report.take()!['X-CacheKit-L1-Latency-Sum-Ns']).toBe('500'); + expect(report.take()).toBeUndefined(); + }); + + it('attaches nothing while the sum is 0, and keeps the 0 ns hits for the next report', () => { + const report = new L1LatencyReport(1); + report.record(0); + expect(report.take()).toBeUndefined(); + report.record(3); + expect(report.take()).toEqual({ + 'X-CacheKit-L1-Latency-Buckets': '2,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0', + 'X-CacheKit-L1-Latency-Sum-Ns': '3', + 'X-CacheKit-L1-Latency-One-In': '1', + }); + }); +}); + +describe('telemetry switch: CACHEKIT_TELEMETRY parsing (API-95)', () => { + it.each([ + ['false', false], + ['0', false], + ['off', false], + ['no', false], + [' OFF ', false], + ['No', false], + ['true', true], + ['1', true], + ['on', true], + ['yes', true], + [' Yes\t', true], + ['TRUE', true], + ['', true], + [' ', true], + ])('reads %j as %s', (value, on) => { + vi.stubEnv('CACHEKIT_TELEMETRY', value); + expect(resolveTelemetry(undefined)).toBe(on); + }); + + it('reads an unknown value as off and warns once per process', () => { + const logged: string[] = []; + setLogger((message) => logged.push(message)); + vi.stubEnv('CACHEKIT_TELEMETRY', 'disabled'); + expect(resolveTelemetry(undefined)).toBe(false); + expect(resolveTelemetry(true)).toBe(false); + expect(logged).toHaveLength(1); + expect(logged[0]).toContain('CACHEKIT_TELEMETRY'); + }); + + it('reads a process environment it cannot read as off, warning once (Deno without --allow-env)', () => { + const logged: string[] = []; + setLogger((message) => logged.push(message)); + const env = process.env; + process.env = new Proxy( + {}, + { + get() { + throw new Error('Requires env access to "CACHEKIT_TELEMETRY"'); + }, + } + ); + try { + expect(resolveTelemetry(undefined)).toBe(false); + expect(resolveTelemetry(true)).toBe(false); + } finally { + process.env = env; + } + expect(logged).toEqual([ + '[cachekit] CACHEKIT_TELEMETRY could not be read ' + + '(Requires env access to "CACHEKIT_TELEMETRY"): telemetry is off', + ]); + }); + + it('reads bindings it cannot see as off, warning once (disallow_importable_env)', async () => { + const logged: string[] = []; + setLogger((message) => logged.push(message)); + setTelemetryBindingsReader(() => { + throw new Error("disallow_importable_env hides the Worker's bindings"); + }); + const backend = cachekitio({ ...CONFIG, telemetry: true, metricsProvider: () => COUNTERS }); + expect(resolveTelemetry(undefined)).toBe(false); + await backend.get('ns:key'); + for (const name of OPTIONAL_METRICS) expect(sentHeaders()[name], name).toBeUndefined(); + expect(logged).toHaveLength(1); + expect(logged[0]).toContain('disallow_importable_env'); + }); + + it('refuses a telemetry setting that is not a boolean', () => { + expect(() => resolveTelemetry('false')).toThrow(ConfigurationError); + expect(() => new CachekitIOCore({ ...CONFIG, telemetry: 0 as unknown as boolean })).toThrow( + ConfigurationError + ); + }); +}); + +describe('telemetry switch: the truth table (API-94, API-95)', () => { + const env = { unset: undefined, on: 'on', off: 'off' } as const; + const setting = { unset: undefined, on: true, off: false } as const; + const table: [keyof typeof setting, keyof typeof env, boolean][] = [ + ['unset', 'unset', true], + ['unset', 'on', true], + ['unset', 'off', false], + ['on', 'unset', true], + ['on', 'on', true], + ['on', 'off', false], + ['off', 'unset', false], + ['off', 'on', false], + ['off', 'off', false], + ]; + + it.each(table)('setting %s, environment %s: on is %s', async (s, e, on) => { + vi.stubEnv('CACHEKIT_TELEMETRY', env[e]); + const backend = cachekitio({ + ...CONFIG, + telemetry: setting[s], + metricsProvider: () => COUNTERS, + }); + await backend.get('ns:key'); + const headers = sentHeaders(); + for (const name of SESSION_AND_COUNTERS) { + expect(headers[name] !== undefined, name).toBe(on); + } + expect(backend.l1Latency !== undefined).toBe(on); + }); + + it.each(table)('on Workers bindings: setting %s, binding %s: on is %s', async (s, e, on) => { + setTelemetryBindingsReader(() => env[e]); + const backend = cachekitio({ + ...CONFIG, + telemetry: setting[s], + metricsProvider: () => COUNTERS, + }); + await backend.get('ns:key'); + expect(sentHeaders()['X-CacheKit-Session-ID'] !== undefined).toBe(on); + }); + + it('takes off from either the process environment or the bindings', () => { + vi.stubEnv('CACHEKIT_TELEMETRY', 'on'); + setTelemetryBindingsReader(() => 'off'); + expect(resolveTelemetry(true)).toBe(false); + vi.stubEnv('CACHEKIT_TELEMETRY', 'off'); + setTelemetryBindingsReader(() => 'on'); + expect(resolveTelemetry(true)).toBe(false); + }); + + it('reads a non-string binding by its string form: a JSON var `false` is off', () => { + setTelemetryBindingsReader(() => false); + expect(resolveTelemetry(undefined)).toBe(false); + }); + + it('reads both controls once, at construction', async () => { + const backend = cachekitio({ ...CONFIG, metricsProvider: () => COUNTERS }); + vi.stubEnv('CACHEKIT_TELEMETRY', 'off'); + await backend.get('ns:key'); + expect(sentHeaders()['X-CacheKit-Session-ID']).toBeDefined(); + + const off = cachekitio({ ...CONFIG, metricsProvider: () => COUNTERS }); + vi.stubEnv('CACHEKIT_TELEMETRY', 'on'); + await off.get('ns:key'); + expect(sentHeaders()['X-CacheKit-Session-ID']).toBeUndefined(); + }); +}); + +describe('telemetry off (API-96)', () => { + it.each([ + ['the setting', { telemetry: false }, undefined], + ['CACHEKIT_TELEMETRY', {}, 'off'], + ])('off through %s: none of the nine optional headers, on any request', async (_, extra, e) => { + vi.stubEnv('CACHEKIT_TELEMETRY', e); + const backend = cachekitioFull({ ...CONFIG, ...extra, metricsProvider: () => COUNTERS }); + fetchSpy.mockImplementation(async () => new Response('{}', { status: 200 })); + await backend.get('ns:key'); + await backend.set('ns:key', new Uint8Array([1]), 60); + await backend.exists('ns:key'); + await backend.delete('ns:key'); + await backend.getTTL('ns:key'); + await backend.acquireLock('ns:key'); + await new CachekitIOCore({ ...CONFIG, ...extra, metricsProvider: () => COUNTERS }).health(); + expect(fetchSpy).toHaveBeenCalledTimes(7); + for (let i = 0; i < 7; i++) { + const headers = sentHeaders(i); + for (const name of OPTIONAL_METRICS) expect(headers[name], name).toBeUndefined(); + // Kept: API-1 requires L1-Status on every ck_sdk_ request. + expect(headers['X-CacheKit-L1-Status']).toBe('miss'); + expect(headers['User-Agent']).toMatch(/cachekit/i); + } + }); + + it('keeps L1-Status disabled for a cache without L1', async () => { + const backend = cachekitio({ + ...CONFIG, + telemetry: false, + metricsProvider: () => ({ ...COUNTERS, l1Enabled: false }), + }); + await backend.get('ns:key'); + expect(sentHeaders()['X-CacheKit-L1-Status']).toBe('disabled'); + }); +}); + +describe('latency report on the wire (API-84 to API-86)', () => { + it('rides the next request, and only that one', async () => { + const backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + await backend.get('ns:a'); + expect(latencyOf(sentHeaders())).toEqual([ + '0,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0', + '300', + String(L1_LATENCY_ONE_IN), + ]); + await backend.get('ns:b'); + expect(latencyOf(sentHeaders())).toEqual([undefined, undefined, undefined]); + }); + + it('never rides GET /v1/cache/health, and stays for the next request (API-84)', async () => { + const backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + fetchSpy.mockResolvedValueOnce(new Response('{"status":"ok"}', { status: 200 })); + await backend.health(); + expect(fetchSpy.mock.calls[0][0]).toBe(`${API_URL}/v1/cache/health`); + expect(latencyOf(sentHeaders())).toEqual([undefined, undefined, undefined]); + await backend.get('ns:a'); + expect(sentHeaders()['X-CacheKit-L1-Latency-Sum-Ns']).toBe('300'); + }); + + it('is consumed by an attempt that may have reached the server, whatever the outcome (API-86)', async () => { + const backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + fetchSpy.mockRejectedValueOnce(new TypeError('fetch failed: socket hang up')); + await expect(backend.get('ns:a')).rejects.toThrow(); + expect(sentHeaders()['X-CacheKit-L1-Latency-Sum-Ns']).toBe('300'); + // The retry, or any later request, carries none of it. + await backend.get('ns:a'); + expect(latencyOf(sentHeaders())).toEqual([undefined, undefined, undefined]); + }); + + describe('rides one attempt on the wire, whatever resends the request', () => { + const auth = `Bearer ${API_KEY}`; + let backend: CachekitIOCore; + const get = async () => { + backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + expect(await backend.get('ns:a')).toBeNull(); + }; + + it("fetch's own resend after a 421 carries none of it", async () => { + // A 421 makes Node's fetch resend the request on a new connection, without + // returning to the backend (the Fetch standard's misdirected-request retry). + expect(await onTheWire([421, 404], get)).toEqual([ + { sumNs: '300', authorization: auth }, + { sumNs: undefined, authorization: auth }, + ]); + expect(backend.l1Latency!.take()).toBeUndefined(); + }); + + it('rides the one attempt of an uncontended request through an Agent built with defaults', async () => { + expect(await onTheWire([404], get, new undici.Agent())).toEqual([ + { sumNs: '300', authorization: auth }, + ]); + expect(backend.l1Latency!.take()).toBeUndefined(); + }); + + // A retrying dispatcher answers a 503 by retrying; the others get a 404 at once. + it.each([ + ['a RetryAgent', [503, 404], () => new undici.RetryAgent(new undici.Agent(), RETRY_ONCE)], + [ + 'an Agent composed with retry', + [503, 404], + () => new undici.Agent().compose(undici.interceptors.retry(RETRY_ONCE)), + ], + [ + 'an Agent whose factory retries', + [503, 404], + () => + new undici.Agent({ + factory: (origin: string, options: object) => + new undici.Pool(origin, options).compose(undici.interceptors.retry(RETRY_ONCE)), + }), + ], + ['an Agent that pipelines', [404], () => new undici.Agent({ pipelining: 2 })], + ])('waits while the global dispatcher is %s, which may resend', async (_, statuses, make) => { + const seen = await onTheWire(statuses, get, make()); + expect(seen).toHaveLength(statuses.length); + for (const attempt of seen) + expect(attempt).toEqual({ sumNs: undefined, authorization: auth }); + // Not taken: it rides a later request, through undici's Agent. + expect(backend.l1Latency!.take()?.['X-CacheKit-L1-Latency-Sum-Ns']).toBe('300'); + }); + + it('leaves a MockAgent alone: a body-matching intercept still sees the bytes of a set', async () => { + vi.stubGlobal('fetch', nodeFetch); + const mock = new undici.MockAgent(); + mock.disableNetConnect(); + const bodies: unknown[] = []; + mock + .get(API_URL) + .intercept({ + path: (path: string) => path.startsWith('/v1/cache/'), + method: 'PUT', + body: (body: unknown) => bodies.push(body) > 0, + }) + .reply(200, '{}'); + try { + await keepingGlobalDispatcher(async () => { + undici.setGlobalDispatcher(mock); + backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + await backend.set('ns:a', new TextEncoder().encode('payload'), 60); + }); + } finally { + await mock.close(); + } + // The bytes themselves, not a stream: fetch read the mock's isMockActive. + const decoded = bodies.map((body) => + body instanceof Uint8Array ? new TextDecoder().decode(body) : String(body) + ); + expect(new Set(decoded)).toEqual(new Set(['payload'])); + // Not undici's Agent: the report is not taken, and waits. + expect(backend.l1Latency!.take()?.['X-CacheKit-L1-Latency-Sum-Ns']).toBe('300'); + }); + + it('a fetch wrapper that retries with the same init carries none of it', async () => { + const seen = await onTheWire([503, 404], async () => { + const routed = globalThis.fetch; + vi.stubGlobal('fetch', async (url: string, init: RequestInit) => { + const response = await routed(url, init); + return response.status === 503 ? routed(url, init) : response; + }); + await get(); + }); + expect(seen).toEqual([ + { sumNs: '300', authorization: auth }, + { sumNs: undefined, authorization: auth }, + ]); + }); + }); + + it('stays in place while no dispatcher can keep it to one attempt', async () => { + await keepingGlobalDispatcher(async () => { + for (const key of DISPATCHER_SLOTS) Reflect.set(globalThis, key, undefined); + const backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + await backend.get('ns:a'); + expect(latencyOf(sentHeaders())).toEqual([undefined, undefined, undefined]); + expect(backend.l1Latency!.take()?.['X-CacheKit-L1-Latency-Sum-Ns']).toBe('300'); + }); + }); + + it('is consumed by a 5xx too: nothing is added back', async () => { + const backend = new CachekitIOCore(CONFIG); + backend.l1Latency!.record(300); + fetchSpy.mockResolvedValueOnce(new Response('boom', { status: 503 })); + await expect(backend.get('ns:a')).rejects.toThrow(); + await backend.get('ns:a'); + expect(latencyOf(sentHeaders())).toEqual([undefined, undefined, undefined]); + }); + + it("goes only to the backend's configured API, and never follows a redirect (API-77)", async () => { + const backend = cachekitioFull(CONFIG); + backend.l1Latency!.record(300); + fetchSpy.mockImplementation(async () => new Response('{}', { status: 200 })); + await backend.get('ns:a'); + await backend.acquireLock('ns:a'); + for (const [url, init] of fetchSpy.mock.calls as [string, RequestInit][]) { + expect(url.startsWith(`${API_URL}/v1/cache/`)).toBe(true); + expect(init.redirect).toBe('manual'); + } + }); + + it("keeps one accumulator per backend: no backend's hits ride another's requests (§ Reports)", async () => { + // API-83 (a forked child starts empty) needs no test here: Node has no + // fork(), and a child process (child_process, cluster) loads the module + // afresh, so its backends start empty by construction. + const a = new CachekitIOCore(CONFIG); + const b = new CachekitIOCore(CONFIG); + a.l1Latency!.record(300); + expect(b.l1Latency!.take()).toBeUndefined(); + await b.get('ns:x'); + expect(sentHeaders()['X-CacheKit-L1-Latency-Sum-Ns']).toBeUndefined(); + }); + + it('is forwarded by every CachekitIO wrapper (Backend.l1Latency)', () => { + for (const factory of [cachekitio, cachekitioWithLocking, cachekitioWithTTL, cachekitioFull]) { + expect(factory(CONFIG).l1Latency).toBeInstanceOf(L1LatencyReport); + expect(factory({ ...CONFIG, telemetry: false }).l1Latency).toBeUndefined(); + } + }); +}); diff --git a/packages/cachekit/src/backends/telemetry.ts b/packages/cachekit/src/backends/telemetry.ts new file mode 100644 index 00000000..282d5e23 --- /dev/null +++ b/packages/cachekit/src/backends/telemetry.ts @@ -0,0 +1,161 @@ +import { ConfigurationError } from '../errors.js'; +import { logError } from '../logger.js'; + +/** + * The telemetry switch and the L1 hit-latency report of a CachekitIO backend + * (protocol spec/saas-api.md § Optional Metrics Headers). + */ + +type Control = 'on' | 'off' | 'unset'; + +let bindingsReader: (() => unknown) | undefined; +let warnedUnknownValue = false; +let warnedUnreadable = false; + +/** + * Package-internal: where the Workers entrypoint reads `CACHEKIT_TELEMETRY` + * from the Worker's bindings. Unset on Node, which has only `process.env`. + * The reader throws when it cannot see the bindings, which reads as off. + */ +export function setTelemetryBindingsReader(reader: (() => unknown) | undefined): void { + bindingsReader = reader; +} + +/** + * One `CACHEKIT_TELEMETRY` value by the spec's own rule (API-95): trimmed and + * case-insensitive; unset or empty is unset, and a value that is neither on + * nor off reads as off, with one warning per process. + */ +function parseControl(raw: unknown): Control { + if (raw === undefined || raw === null) return 'unset'; + const value = String(raw).trim().toLowerCase(); + if (value === '') return 'unset'; + if (value === 'true' || value === '1' || value === 'on' || value === 'yes') return 'on'; + if (value === 'false' || value === '0' || value === 'off' || value === 'no') return 'off'; + if (!warnedUnknownValue) { + warnedUnknownValue = true; + logError( + `[cachekit] CACHEKIT_TELEMETRY=${JSON.stringify(String(raw))} is not one of ` + + 'true/false/1/0/on/off/yes/no: telemetry is off' + ); + } + return 'off'; +} + +/** + * One `CACHEKIT_TELEMETRY` source, parsed, or off when it cannot be read: + * the user may have set it to off there, so an unreadable source is off, as + * an unknown value is (API-95), with one warning per process. Deno without + * `--allow-env` throws on a `process.env` read; Workers under + * `disallow_importable_env` hides the bindings. + */ +function readControl(read: () => unknown): Control { + try { + return parseControl(read()); + } catch (error) { + if (!warnedUnreadable) { + warnedUnreadable = true; + const reason = error instanceof Error ? error.message : String(error); + logError(`[cachekit] CACHEKIT_TELEMETRY could not be read (${reason}): telemetry is off`); + } + return 'off'; + } +} + +function readProcessEnv(): unknown { + return typeof process !== 'undefined' ? process.env?.CACHEKIT_TELEMETRY : undefined; +} + +/** + * Resolve a CachekitIO backend's telemetry switch (API-94, API-95). Called + * once, at construction. `CACHEKIT_TELEMETRY` is read from `process.env` + * wherever there is one and, on Workers, from the bindings too. The spec's + * truth table reduces to: off from any control wins, and otherwise + * telemetry is on, the closed-beta default. + */ +export function resolveTelemetry(setting: unknown): boolean { + if (setting !== undefined && typeof setting !== 'boolean') { + // A JavaScript caller's 'false' is truthy: refuse it rather than read it as on. + throw new ConfigurationError( + `telemetry must be true, false or undefined, got ${JSON.stringify(setting)}` + ); + } + // Read both before deciding, so an unknown value warns whatever the setting says. + const fromProcess = readControl(readProcessEnv); + const fromBindings = bindingsReader === undefined ? 'unset' : readControl(bindingsReader); + return setting !== false && fromProcess !== 'off' && fromBindings !== 'off'; +} + +/** Package-internal: forget the one-time warnings. Tests only. */ +export function _resetTelemetryWarningForTesting(): void { + warnedUnknownValue = false; + warnedUnreadable = false; +} + +/** + * The sampling denominator N (`X-CacheKit-L1-Latency-One-In`): each call is + * timed with probability 1/N (API-81). A timed hit costs about 5,000 + * instructions more than an untimed one (two clock reads and the bucket), + * against about 4,700 for a whole memoised wrap() hit on Node 24, measured. + * At 1 in 1024 that is 5 per hit; the draw itself costs about 80. + */ +export const L1_LATENCY_ONE_IN = 1024; + +/** Bucket 16's upper bound: 60 s. A longer duration is discarded (API-87). */ +const MAX_DURATION_NS = 60_000_000_000; +/** Bucket 15's upper bound, 2^21 ns: anything above it is bucket 16. */ +const BUCKET_15_MAX_NS = 2_097_152; + +/** + * One CachekitIO backend's L1 hit-latency accumulator (§ L1 hit latency). + * The cache records the durations of the calls it timed; the backend takes + * them, as one report, onto its next request. + */ +export class L1LatencyReport { + private readonly buckets = new Array(16).fill(0); + private sumNs = 0; + + constructor(readonly oneIn: number) {} + + /** + * Add one timed L1 hit, in whole nanoseconds. Over 60 s is discarded + * (API-87), as is a hit that would take the sum past 2^53 - 1, the + * largest value the wire grammar allows. + */ + record(durationNs: number): void { + if (durationNs > MAX_DURATION_NS || this.sumNs + durationNs > Number.MAX_SAFE_INTEGER) return; + // Bucket i (from 0) holds (2^(i+6), 2^(i+7)] ns; bucket 0 holds 0..128. + // 32 - clz32(d - 1) is ceil(log2 d) for 1 <= d <= 2^32, and d is at most + // 2^21 by then. + const bucket = + durationNs <= 128 + ? 0 + : durationNs > BUCKET_15_MAX_NS + ? 15 + : 32 - Math.clz32(durationNs - 1) - 7; + this.buckets[bucket]++; + this.sumNs += durationNs; + } + + /** True while the sum is 0: there is no report to attach (API-85). */ + get empty(): boolean { + return this.sumNs < 1; + } + + /** + * Take the report as its three headers and reset the accumulator, or + * return undefined while the sum is 0, leaving it in place (API-85). One + * synchronous step, so no other request can carry the same hits. + */ + take(): Record | undefined { + if (this.empty) return undefined; + const headers = { + 'X-CacheKit-L1-Latency-Buckets': this.buckets.join(','), + 'X-CacheKit-L1-Latency-Sum-Ns': String(this.sumNs), + 'X-CacheKit-L1-Latency-One-In': String(this.oneIn), + }; + this.buckets.fill(0); + this.sumNs = 0; + return headers; + } +} diff --git a/packages/cachekit/src/backends/types.ts b/packages/cachekit/src/backends/types.ts index f412c940..79d3801f 100644 --- a/packages/cachekit/src/backends/types.ts +++ b/packages/cachekit/src/backends/types.ts @@ -1,3 +1,5 @@ +import type { L1LatencyReport } from './telemetry.js'; + /** * Backend interface for pluggable cache storage. * @@ -197,6 +199,14 @@ export interface Backend { * Like `keyPrefix`, the value MUST be constant from construction onward. */ readonly compressionDefault?: boolean; + + /** + * The CachekitIO backend's L1 hit-latency report, which the cache records + * the L1 hits it times into; present only while that backend's telemetry + * is on. Delegating wrappers forward it like `keyPrefix`. + * @internal + */ + readonly l1Latency?: L1LatencyReport; } /** Result of {@link Backend.getWithTtl}: the bytes plus remaining lifetime. */ @@ -361,6 +371,16 @@ export interface CachekitIOBackendConfig { allowCustomHost?: boolean; /** Provider function for L1 cache metrics (used in request headers) */ metricsProvider?: () => L1Metrics | null; + /** + * The telemetry switch (protocol spec/saas-api.md § Telemetry switch). Left + * `undefined`, the `CACHEKIT_TELEMETRY` environment variable decides, and + * telemetry is on when that is unset too. `false`, or `CACHEKIT_TELEMETRY` + * set to `false`, `0`, `off` or `no`, turns it off: requests then carry no + * session, counter, hit-rate or L1-latency header, only + * `X-CacheKit-L1-Status` and the `User-Agent`. Off wins over on from either + * control. Both are read once, when the backend is constructed. + */ + telemetry?: boolean; } export interface L1Metrics { diff --git a/packages/cachekit/src/cache-core.ts b/packages/cachekit/src/cache-core.ts index 3475ccd1..f28387e9 100644 --- a/packages/cachekit/src/cache-core.ts +++ b/packages/cachekit/src/cache-core.ts @@ -9,6 +9,7 @@ import type { InvalidationConfig, } from './types/cache.js'; import type { Backend, L1Metrics, LockableBackend } from './backends/types.js'; +import type { L1LatencyReport } from './backends/telemetry.js'; import type { InvalidationEvent } from './l1/types.js'; import type { MetricsCollector, MetricsConfig } from './metrics/prometheus.js'; import { checkDefaultLabels } from './metrics/labels.js'; @@ -265,15 +266,15 @@ export interface CacheRuntime { * Resolve a backend config union to a Backend instance. `stampede` carries * the cold-miss protection config so a runtime can select a lock-capable * variant (Node picks cachekitioWithLocking for apiKey configs when - * distributedLock is on); `l1Telemetry` lazily reads the cache's live - * hit/miss counters so a SaaS backend can auto-wire its telemetry headers. + * distributedLock is on); `readHitCounters` lazily reads the cache's live + * hit/miss counters so a CachekitIO backend can auto-wire its counter headers. * Runtimes without these conveniences may ignore both — users can still * pass a fully-configured Backend instance directly. */ resolveBackend( config: CacheOptions['backend'], stampede?: StampedeConfig, - l1Telemetry?: () => L1Metrics + readHitCounters?: () => L1Metrics ): Backend; /** * Create the Prometheus metrics collector. Absent on platforms without @@ -310,6 +311,14 @@ export interface CacheRuntime { * request that set it ends. */ setUnrefTimeout?: UnrefTimeout; + /** + * A monotonic nanosecond clock that advances during synchronous execution + * and resolves 100 ns or finer, which the protocol requires before an SDK + * reports L1 hit latency (spec/saas-api.md § L1 hit latency, API-82). Node + * supplies `process.hrtime.bigint`. Absent on Workers, whose clock moves + * only on I/O: there no L1 hit is timed and no latency header is sent. + */ + l1HitClock?: () => bigint; } /** @@ -351,10 +360,30 @@ export class CacheImpl implements SecureCache { private readonly defaultTtl: number; private readonly invalidationChannel: InvalidationChannelLike | null = null; private readonly metrics: MetricsCollector; - // Live hit/miss counters. Feed both the Prometheus collector and the SaaS - // X-CacheKit-L1-* telemetry headers (auto-wired via resolveBackend's - // l1Telemetry hook below). - private readonly telemetry = { l1Hits: 0, l2Hits: 0, misses: 0 }; + // Live hit/miss counters, kept whether telemetry is on or off. They feed + // the Prometheus collector, and the X-CacheKit-L1-* counter headers of a + // CachekitIO backend whose telemetry is on (auto-wired via resolveBackend's + // readHitCounters hook below). + private readonly hitCounters = { l1Hits: 0, l2Hits: 0, misses: 0 }; + /** + * The backend's L1 hit-latency report, when its telemetry is on and the + * runtime has a qualifying clock; else null, and no call draws or reads a + * clock. See startL1Timer. + */ + private readonly l1Latency: L1LatencyReport | null; + /** + * A call is timed when its draw, the top 30 bits of the generator's 32, is + * below this: 2^30 / l1Latency.oneIn, so with probability 1/oneIn (API-81). + * Thirty bits keep the draw a small integer, which a hit never boxes. + */ + private readonly l1LatencyThreshold: number; + /** + * The xorshift32 state behind each call's draw, seeded from Math.random + * and never 0. Drawing with Math.random instead cost a memoised L1 hit on + * Node 24 about 70 instructions more, measured. + */ + private l1LatencyRng: number; + private readonly l1HitClock: (() => bigint) | undefined; private readonly swrRequiresWaitUntil: boolean; private readonly setUnrefTimeout: UnrefTimeout | undefined; /** @@ -459,12 +488,18 @@ export class CacheImpl implements SecureCache { const metricsOption = options.metrics ?? false; if (typeof metricsOption === 'object') checkDefaultLabels(metricsOption.defaultLabels); - // Initialize backend. The telemetry getter reads `this` lazily (per + // Initialize backend. The hit-counter reader reads `this` lazily (per // request), so constructor field order is safe. this.backend = runtime.resolveBackend(options.backend, options.stampede, () => ({ - ...this.telemetry, + ...this.hitCounters, l1Enabled: this.l1 !== null, })); + // Read once: a backend's report is fixed for its lifetime (API-95). + const report = this.l1 && runtime.l1HitClock ? this.backend.l1Latency : undefined; + this.l1Latency = report ?? null; + this.l1LatencyThreshold = report ? 2 ** 30 / report.oneIn : 0; + this.l1LatencyRng = (Math.random() * 2 ** 32) | 0 || 1; + this.l1HitClock = runtime.l1HitClock; // Lock capability. Duck-typed like cachekit-py's hasattr check: // user-supplied Backend instances aren't required to declare the @@ -558,13 +593,39 @@ export class CacheImpl implements SecureCache { // fire-and-forget `void` keeps them off the hot path. private recordHit(layer: 'l1' | 'l2'): void { - if (layer === 'l1') this.telemetry.l1Hits++; - else this.telemetry.l2Hits++; + if (layer === 'l1') this.hitCounters.l1Hits++; + else this.hitCounters.l2Hits++; this.metrics.countHit(layer); } + /** + * The first step of a call that can count an L1 hit, when l1Latency is set: + * draw whether to time it, before its outcome is known, independently for + * every call (API-81), and if so read the clock. Returns the reading, or + * undefined for a call not timed. Joined calls, exists() and reads inside + * wrap() are never timed (API-80): their callers never pass a reading on. + */ + private startL1Timer(): bigint | undefined { + let x = this.l1LatencyRng; + x ^= x << 13; + x ^= x >>> 17; + x ^= x << 5; + this.l1LatencyRng = x; + return x >>> 2 < this.l1LatencyThreshold ? this.l1HitClock!() : undefined; + } + + /** + * The last step of a timed call that L1 served: add its duration to the + * backend's latency report. Called in the synchronous step that produces + * the value, after its recordHit, or for a read inside wrap() just before + * the one wrapRead makes (API-79). A timed call nobody stops is discarded. + */ + private stopL1Timer(timedFrom: bigint): void { + this.l1Latency!.record(Number(this.l1HitClock!() - timedFrom)); + } + private recordMiss(): void { - this.telemetry.misses++; + this.hitCounters.misses++; this.metrics.countMiss(); } @@ -858,7 +919,8 @@ export class CacheImpl implements SecureCache { // Not async: an async function that returns getEntry's promise settles two // microtask turns after it, which a plaintext L1 hit would pay every time. get(key: string): Promise { - return this.getEntry(key, false); + const timedFrom = this.l1Latency === null ? undefined : this.startL1Timer(); + return this.getEntry(key, false, undefined, undefined, false, true, timedFrom); } /** @@ -1081,7 +1143,8 @@ export class CacheImpl implements SecureCache { * same way, and a stale read never re-populates L1. `readFreshness` asks * for those headers even with L1 off, so a stale-grace wrap sees the label. * `useL1` false (a wrap() call with skipL1) reads as if L1 were off: L1 is - * neither read nor re-populated. + * neither read nor re-populated. `timedFrom` is the start reading of a call + * drawn for timing (see startL1Timer): an L1 hit stops its clock here. */ private async getEntry( key: string, @@ -1089,7 +1152,8 @@ export class CacheImpl implements SecureCache { ttlSeconds?: number, probe?: L2ReadProbe, readFreshness = false, - useL1 = true + useL1 = true, + timedFrom?: bigint ): Promise { this.ensureNotClosed(); const l1 = useL1 ? this.l1 : null; @@ -1106,6 +1170,9 @@ export class CacheImpl implements SecureCache { if (decoded !== null) { if (probe) probe.servedBy = 'l1'; else this.recordHit('l1'); + // Stopped here even for a wrap() read, whose hit wrapRead counts once + // this returns: this is the step that produces the value. + if (timedFrom !== undefined) this.stopL1Timer(timedFrom); return decoded.value; } } @@ -1805,7 +1872,12 @@ export class CacheImpl implements SecureCache { ); if (autoKeys.memoising) this.trackKeyMemo(autoKeys); return async (...args: TArgs): Promise => { - // First, and before keying in particular: keying puts the arguments in the + // The draw comes first, before the call's outcome is known (API-81), and + // the clock starts at the call's first step (§ L1 hit latency). + // The null check is inline: on Node 22 a call here cost a memoised hit + // with no report about 1% more, measured. + const timedFrom = this.l1Latency === null ? undefined : this.startL1Timer(); + // Then, and before keying in particular: keying puts the arguments in the // key memo, which close() has emptied, and only the timer of a new window // would empty it again. The flag is read here because calling // ensureNotClosed() on every call cost an L1 hit about 100 instructions, @@ -1851,6 +1923,7 @@ export class CacheImpl implements SecureCache { const held = slot?.entry; if (held !== undefined && this.l1!.hit(held, now)) { this.recordHit('l1'); + if (timedFrom !== undefined) this.stopL1Timer(timedFrom); return held.value as TResult; } @@ -1869,6 +1942,7 @@ export class CacheImpl implements SecureCache { : { value: l1Value as TResult }; if (decoded !== null) { this.recordHit('l1'); + if (timedFrom !== undefined) this.stopL1Timer(timedFrom); return decoded.value; } } @@ -1926,7 +2000,10 @@ export class CacheImpl implements SecureCache { // own, discarded with the entry: one Map lookup, against the key // generation that call has already paid for. if (autoKeys.memoising) slot!.entry = this.l1.entry(cacheKey); + // After scheduleRefresh: a stale serve's timed interval includes + // the synchronous start of its revalidation. this.recordHit('l1'); + if (timedFrom !== undefined) this.stopL1Timer(timedFrom); return decoded.value; } } @@ -1941,7 +2018,8 @@ export class CacheImpl implements SecureCache { if (existing) { // Served in process without a backend request, as cachekit-py // counts a joined caller: one L1 hit. A flight that rejects counts - // nothing for its joiners. + // nothing for its joiners. Never timed: its duration is the + // flight's work (API-80). const joined = (await existing) as TResult; this.recordHit('l1'); return joined; @@ -1952,7 +2030,8 @@ export class CacheImpl implements SecureCache { () => fn(...args), options, staleTtl, - waitUntil + waitUntil, + timedFrom ); this.inflight.set(cacheKey, flight); try { @@ -1969,7 +2048,9 @@ export class CacheImpl implements SecureCache { * lock when the backend supports it and stampede.distributedLock is on. * `waitUntil` is the flight leader's: the herd shares one lock, so its * release rides the request that took it, as does a revalidation of a - * server-stale hit (`staleTtl` > 0, see wrapRead). + * server-stale hit (`staleTtl` > 0, see wrapRead). So is `timedFrom`, the + * leader's start reading when its call was drawn for timing: a read that + * L1 serves is the call's L1 hit, and is timed as one. */ private async resolveMiss( cacheKey: string, @@ -1977,7 +2058,8 @@ export class CacheImpl implements SecureCache { compute: () => Promise, options: WrapOptionsBase, staleTtl: number, - waitUntil?: WaitUntil + waitUntil?: WaitUntil, + timedFrom?: bigint ): Promise { // Workers cancel unregistered background work at response return, so // without a bound waitUntil there is no revalidation: fail safe to the @@ -1993,7 +2075,7 @@ export class CacheImpl implements SecureCache { } : null; const read: L2ReadProbe = { cleanMiss: false, servedBy: null, stale: false }; - const cached = await this.wrapRead(cacheKey, interop, options, grace, read); + const cached = await this.wrapRead(cacheKey, interop, options, grace, read, timedFrom); if (cached !== null) { return cached; } @@ -2006,7 +2088,8 @@ export class CacheImpl implements SecureCache { options, grace, read.cleanMiss, - waitUntil + waitUntil, + timedFrom ); if (locked !== LOCK_FALLTHROUGH) { return locked; @@ -2061,13 +2144,21 @@ export class CacheImpl implements SecureCache { options: WrapOptionsBase, grace: StaleGrace | null, cleanMiss: boolean, - waitUntil?: WaitUntil + waitUntil?: WaitUntil, + timedFrom?: bigint ): Promise { const lockable = this.lockable!; const { lockTimeoutMs, lockWaitMs, lockPollMs } = this.stampede; const deadline = Date.now() + lockWaitMs; const recheck = async (): Promise => { - const filled = await this.wrapRead(cacheKey, interop, options, grace); + const filled = await this.wrapRead( + cacheKey, + interop, + options, + grace, + undefined, + timedFrom + ); return filled !== null ? filled : LOCK_FALLTHROUGH; }; @@ -2097,7 +2188,9 @@ export class CacheImpl implements SecureCache { cacheKey, interop, options, - grace && { ...grace, revalidate: null } + grace && { ...grace, revalidate: null }, + undefined, + timedFrom ); if (filled !== null) { return filled; @@ -2152,7 +2245,8 @@ export class CacheImpl implements SecureCache { interop: boolean, options: WrapOptionsBase, grace: StaleGrace | null, - probe: L2ReadProbe = { cleanMiss: false, servedBy: null, stale: false } + probe: L2ReadProbe = { cleanMiss: false, servedBy: null, stale: false }, + timedFrom?: bigint ): Promise { const value = await this.getEntry( cacheKey, @@ -2160,7 +2254,8 @@ export class CacheImpl implements SecureCache { options.ttl, probe, grace !== null, - !options.skipL1 + !options.skipL1, + timedFrom ); if (value === null) return null; if (probe.stale && grace) { diff --git a/packages/cachekit/src/cache.l1-latency.test.ts b/packages/cachekit/src/cache.l1-latency.test.ts new file mode 100644 index 00000000..d79ac59e --- /dev/null +++ b/packages/cachekit/src/cache.l1-latency.test.ts @@ -0,0 +1,321 @@ +/** + * L1 hit latency as the cache times it on Node (protocol spec/saas-api.md + * § L1 hit latency): which calls are timed, over what interval, and how they + * reach the CachekitIO backend's next request. The clock is faked; each test + * names the requirement it pins. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createCache } from './index.js'; +import { cachekitio } from './backends/cachekitio-factory.js'; +import { L1LatencyReport } from './backends/telemetry.js'; +import { generateKey } from './serialization/key-generator.js'; +import type { SecureCache } from './types/cache.js'; +import { InMemoryBackend } from '../test/fixtures/metrics.js'; + +const API_URL = 'https://api.test.cachekit.io'; +const API_KEY = 'ck_test_fake-not-a-secret'; // pragma: allowlist secret +const BACKEND = { apiKey: API_KEY, apiUrl: API_URL, allowCustomHost: true }; +const OPTIONS = { namespace: 'svc:getThing', ttl: 60 }; + +/** Each read of the fake clock returns `now`, then moves it on by STEP ns. */ +const STEP = 250n; +let now: bigint; +let clockReads: number; +let fetchSpy: ReturnType; +let caches: SecureCache[]; + +/** A fake of the cache API: GET serves what PUT stored, a lock is always contested. */ +function fakeServer() { + const store = new Map(); + return vi.fn(async (url: string, init: RequestInit) => { + const key = decodeURIComponent(new URL(url).pathname.slice('/v1/cache/'.length)); + if (key.endsWith('/lock')) return new Response(null, { status: 409 }); + switch (init.method) { + case 'GET': { + const value = store.get(key); + return value ? new Response(value.slice()) : new Response(null, { status: 404 }); + } + case 'PUT': + store.set(key, init.body as Uint8Array); + return new Response('{}'); + default: + return new Response('{}'); + } + }); +} + +function sent(call = -1): Record { + const calls = fetchSpy.mock.calls; + const [, init] = calls[call < 0 ? calls.length + call : call] as [string, RequestInit]; + return init.headers as Record; +} + +/** + * The report the requests since the last call carried, ending with a + * cache.delete so that one request at least follows the hits. A background + * refresh's write can be the one that carries it. At most one may. + */ +let scanned = 0; +async function nextReport(cache: SecureCache) { + await cache.delete('svc:probe'); + const reports = (fetchSpy.mock.calls as [string, RequestInit][]) + .slice(scanned) + .map(([, init]) => init.headers as Record) + .filter((headers) => headers['X-CacheKit-L1-Latency-Sum-Ns'] !== undefined); + scanned = fetchSpy.mock.calls.length; + expect(reports.length).toBeLessThanOrEqual(1); + const headers = reports[0] ?? {}; + return { + buckets: headers['X-CacheKit-L1-Latency-Buckets']?.split(',').map(Number), + sum: headers['X-CacheKit-L1-Latency-Sum-Ns'], + oneIn: headers['X-CacheKit-L1-Latency-One-In'], + }; +} + +/** Bucket counts with `count` hits in bucket `index` (from 0). */ +function only(index: number, count: number): number[] { + const buckets = new Array(16).fill(0); + buckets[index] = count; + return buckets; +} + +/** + * Time every call of `cache`: tests the timing, not the sampling. The field + * must exist, or the write would do nothing and a test that expects no + * timing would pass by chance. + */ +function timeEveryCall(cache: SecureCache): SecureCache { + const sampling = cache as unknown as { l1LatencyThreshold: number }; + expect(sampling.l1LatencyThreshold, 'CacheImpl.l1LatencyThreshold').toBeTypeOf('number'); + sampling.l1LatencyThreshold = 2 ** 30; + return cache; +} + +function track(cache: SecureCache): SecureCache { + caches.push(cache); + return cache; +} + +beforeEach(() => { + now = 1_000n; + clockReads = 0; + scanned = 0; + caches = []; + vi.spyOn(process.hrtime, 'bigint').mockImplementation(() => { + clockReads++; + const reading = now; + now += STEP; + return reading; + }); + fetchSpy = fakeServer(); + vi.stubGlobal('fetch', fetchSpy); + vi.stubEnv('CACHEKIT_TELEMETRY', undefined); +}); + +afterEach(async () => { + for (const cache of caches) await cache.close(); + vi.useRealTimers(); + vi.unstubAllGlobals(); + vi.unstubAllEnvs(); + vi.restoreAllMocks(); +}); + +describe('what is timed (API-79)', () => { + it('times a memoised wrap() hit and an SWR L1 read, each once, over the whole call', async () => { + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + const fn = cache.wrap(async (id: number) => ({ id }), OPTIONS); + await fn(7); // miss: drawn and started, then discarded + await fn(7); // the L1 read by key, which hands the entry to the memo + await fn(7); // the memoised hit + // Two hits, each one STEP from its start to its stop. + expect(await nextReport(cache)).toEqual({ + buckets: only(1, 2), + sum: String(2n * STEP), + oneIn: '1024', + }); + }); + + it('times an object-argument wrap() hit, which no memo keeps', async () => { + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + const fn = cache.wrap(async (q: { id: number }) => q, OPTIONS); + await fn({ id: 1 }); + await fn({ id: 1 }); + expect((await nextReport(cache)).buckets).toEqual(only(1, 1)); + }); + + it('times a get() that L1 serves', async () => { + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + await cache.set('svc:k', { v: 1 }); + expect(await cache.get('svc:k')).toEqual({ v: 1 }); + expect((await nextReport(cache)).buckets).toEqual(only(1, 1)); + }); + + it('times hits on a cachekitio() instance passed as the backend', async () => { + const backend = cachekitio(BACKEND); + // One in 1: every call is timed, and the report says so. + (backend as { l1Latency?: L1LatencyReport }).l1Latency = new L1LatencyReport(1); + const cache = track(createCache({ backend, metrics: false })); + await cache.set('svc:k', 1); + await cache.get('svc:k'); + await cache.get('svc:k'); + expect(await nextReport(cache)).toEqual({ + buckets: only(1, 2), + sum: String(2n * STEP), + oneIn: '1', + }); + }); + + it("includes a stale serve's synchronous start of its revalidation", async () => { + vi.useFakeTimers({ toFake: ['Date'] }); + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + // The wrapped function's synchronous part takes 1 ms on the fake clock. + const fn = cache.wrap(async (id: number) => { + now += 1_000_000n; + return { id }; + }, OPTIONS); + await fn(7); + // Past the stale point (half the ttl, plus at most 10% jitter), not past the ttl. + vi.setSystemTime(Date.now() + 40_000); + await fn(7); + // The fn ran twice: the miss, and the revalidation the stale serve started. + expect(now).toBe(1_000n + 2_000_000n + 3n * STEP); + const report = await nextReport(cache); + expect(report.sum).toBe(String(1_000_000n + STEP)); + expect(report.buckets).toEqual(only(13, 1)); // 524,289 to 1,048,576 ns + }); + + it('times a wrap() call whose cold-path read L1 serves', async () => { + const cache = track( + timeEveryCall( + createCache({ + backend: BACKEND, + metrics: false, + stampede: { distributedLock: true, lockWaitMs: 40, lockPollMs: 10 }, + }) + ) + ); + const compute = vi.fn(async (id: number) => ({ id, computed: true })); + const fn = cache.wrap(compute, OPTIONS); + // L1 misses, L2 misses, the lock is contested: the call waits, then reads again. + const call = fn(7); + await new Promise((resolve) => setTimeout(resolve, 15)); + // Filled meanwhile, in process: the call's re-check finds it in L1. + await cache.set(generateKey(OPTIONS.namespace, [7]), { id: 7, computed: false }); + expect(await call).toEqual({ id: 7, computed: false }); + expect(compute).not.toHaveBeenCalled(); + const report = await nextReport(cache); + expect(report.buckets).toEqual(only(1, 1)); + expect(report.sum).toBe(String(STEP)); + }); + + it('draws before the outcome is known, and discards a timed miss or L2 hit (API-81)', async () => { + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + await cache.get('svc:absent'); // miss + const other = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + await other.set('svc:k', 1); + await cache.get('svc:k'); // an L2 hit in `cache`, which backfills its L1 + expect(clockReads).toBe(2); // two starts, no stop + expect((await nextReport(cache)).sum).toBeUndefined(); + }); +}); + +describe('what is never timed (API-80)', () => { + it('a wrap() call that joined another call for the same key', async () => { + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + let finish!: () => void; + const gate = new Promise((resolve) => (finish = resolve)); + const fn = cache.wrap(async (id: number) => { + await gate; + return { id }; + }, OPTIONS); + const leader = fn(7); + const joined = fn(7); + finish(); + await Promise.all([leader, joined]); + expect((await nextReport(cache)).sum).toBeUndefined(); + // The joined call counted its one L1 hit, and nothing was timed. + expect(sent()['X-CacheKit-L1-Hits']).toBe('1'); + }); + + it('an exists() check, which draws no clock at all', async () => { + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + await cache.set('svc:k', 1); + expect(await cache.exists('svc:k')).toBe(true); + expect(clockReads).toBe(0); + expect((await nextReport(cache)).sum).toBeUndefined(); + }); +}); + +describe('sampling (API-81)', () => { + it('times about 1 in N calls, by a draw independent of the key, never every Nth', async () => { + const backend = cachekitio(BACKEND); + (backend as { l1Latency?: L1LatencyReport }).l1Latency = new L1LatencyReport(4); + const cache = track(createCache({ backend, metrics: false })); + const fn = cache.wrap(async (id: number) => ({ id }), OPTIONS); + await fn(1); + await fn(2); + const timed = { 1: 0, 2: 0 }; + const gaps = new Set(); + let last = 0; + const calls = 40_000; + for (let i = 1; i <= calls; i++) { + const key = (i % 2) + 1; + const before = clockReads; + await fn(key); + if (clockReads - before === 2) { + timed[key as 1 | 2]++; + gaps.add(i - last); + last = i; + } + } + // Each key alternates with the other on every call: a counter would time + // only one of them. Binomial(20,000, 1/4): mean 5,000, sd 61. + expect(timed[1]).toBeGreaterThan(4_600); + expect(timed[1]).toBeLessThan(5_400); + expect(timed[2]).toBeGreaterThan(4_600); + expect(timed[2]).toBeLessThan(5_400); + expect(gaps.size).toBeGreaterThan(10); + }); +}); + +describe('only the CachekitIO backend reports (API-77, API-82)', () => { + it('a cache on any other backend never draws or reads the clock', async () => { + const cache = track( + timeEveryCall(createCache({ backend: new InMemoryBackend(), metrics: false })) + ); + const fn = cache.wrap(async (id: number) => ({ id }), OPTIONS); + for (let i = 0; i < 3; i++) await fn(7); + await cache.get(generateKey(OPTIONS.namespace, [7])); + expect(clockReads).toBe(0); + }); + + it('with telemetry off, no call is timed and no latency header is sent', async () => { + vi.stubEnv('CACHEKIT_TELEMETRY', 'off'); + const cache = track(timeEveryCall(createCache({ backend: BACKEND, metrics: false }))); + const fn = cache.wrap(async (id: number) => ({ id }), OPTIONS); + for (let i = 0; i < 3; i++) await fn(7); + expect(clockReads).toBe(0); + const headers = sent(); + expect(headers['X-CacheKit-L1-Latency-Buckets']).toBeUndefined(); + expect(headers['X-CacheKit-Session-ID']).toBeUndefined(); + }); + + it('createCache.io takes the telemetry setting', async () => { + const cache = track(createCache.io({ apiKey: API_KEY, telemetry: false, metrics: false })); + await cache.get('svc:k'); + expect(sent()['X-CacheKit-Session-ID']).toBeUndefined(); + expect(sent()['X-CacheKit-L1-Status']).toBe('miss'); + }); + + it('times with the Node clock by default, one call in 1024', async () => { + const cache = track(createCache({ backend: BACKEND, metrics: false })); + const fn = cache.wrap(async (id: number) => ({ id }), OPTIONS); + for (let i = 0; i < 20_000; i++) await fn(7); + // Binomial(20,000, 1/1024): mean 19.5, sd 4.4. Each timed hit reads twice. + expect(clockReads).toBeGreaterThan(2 * 3); + expect(clockReads).toBeLessThan(2 * 45); + const report = await nextReport(cache); + expect(report.oneIn).toBe('1024'); + expect(Number(report.sum)).toBe(Number(STEP) * report.buckets!.reduce((a, b) => a + b, 0)); + }); +}); diff --git a/packages/cachekit/src/cache.single-flight.test.ts b/packages/cachekit/src/cache.single-flight.test.ts index 13b5e5b5..31d49529 100644 --- a/packages/cachekit/src/cache.single-flight.test.ts +++ b/packages/cachekit/src/cache.single-flight.test.ts @@ -114,8 +114,8 @@ class WaitedGrantBackend extends CountingLockableBackend { /** The cache's live hit/miss counters: what its telemetry headers carry. */ const outcomes = (cache: SecureCache): { l1Hits: number; l2Hits: number; misses: number } => { const { l1Hits, l2Hits, misses } = ( - cache as unknown as { telemetry: { l1Hits: number; l2Hits: number; misses: number } } - ).telemetry; + cache as unknown as { hitCounters: { l1Hits: number; l2Hits: number; misses: number } } + ).hitCounters; return { l1Hits, l2Hits, misses }; }; diff --git a/packages/cachekit/src/cache.stale-grace.test.ts b/packages/cachekit/src/cache.stale-grace.test.ts index e1917847..152a9bf4 100644 --- a/packages/cachekit/src/cache.stale-grace.test.ts +++ b/packages/cachekit/src/cache.stale-grace.test.ts @@ -132,8 +132,8 @@ const revalidating = (cache: SecureCache) => /** The cache's live hit/miss counters: what its telemetry headers carry. */ const outcomes = (cache: SecureCache) => { const { l1Hits, l2Hits, misses } = ( - cache as unknown as { telemetry: { l1Hits: number; l2Hits: number; misses: number } } - ).telemetry; + cache as unknown as { hitCounters: { l1Hits: number; l2Hits: number; misses: number } } + ).hitCounters; return { l1Hits, l2Hits, misses }; }; diff --git a/packages/cachekit/src/cache.ts b/packages/cachekit/src/cache.ts index af3efb39..a5116b02 100644 --- a/packages/cachekit/src/cache.ts +++ b/packages/cachekit/src/cache.ts @@ -22,6 +22,18 @@ import timers from 'node:timers'; // a missing export fails the whole entry. Without it, keys stay in JS. setKeyHashBinding((core as { blake2b256Hex?: Blake2b256Hex }).blake2b256Hex); +/** + * The protocol names `process.hrtime.bigint()` as the qualifying L1-latency + * clock on Node.js (spec/saas-api.md, API-82). Bun and Deno implement it too, + * and load this entry, but a clock qualifies only on a runtime the docs name, + * so they time no L1 hit. + */ +const onNodeJs = + typeof process !== 'undefined' && + typeof process.versions?.node === 'string' && + process.versions.bun === undefined && + process.versions.deno === undefined; + /** * Node runtime: Redis + CachekitIO backends, NAPI bindings for the * ByteStorage envelope and encryption, Redis Pub/Sub invalidation. @@ -32,25 +44,25 @@ const nodeRuntime: CacheRuntime = { resolveBackend( config: CacheOptions['backend'], stampede?: StampedeConfig, - l1Telemetry?: () => L1Metrics + readHitCounters?: () => L1Metrics ) { if ('get' in config) { return config; } if ('apiKey' in config) { const ioConfig = config as CachekitIOBackendConfig; - // Auto-wire the SaaS telemetry headers from the cache's live hit/miss + // Auto-wire the counter headers from the cache's live hit/miss // counters; an explicit user-supplied provider still wins. - const withTelemetry: CachekitIOBackendConfig = { + const withCounters: CachekitIOBackendConfig = { ...ioConfig, - metricsProvider: ioConfig.metricsProvider ?? l1Telemetry, + metricsProvider: ioConfig.metricsProvider ?? readHitCounters, }; // The plain SaaS factory has no lock capability; the distributedLock // opt-in selects the lockable wrapper so config-based users aren't // forced to construct the backend by hand. return stampede?.distributedLock - ? cachekitioWithLocking(withTelemetry) - : cachekitio(withTelemetry); + ? cachekitioWithLocking(withCounters) + : cachekitio(withCounters); } return redis(config as RedisBackendConfig); }, @@ -67,6 +79,7 @@ const nodeRuntime: CacheRuntime = { timer.unref(); return () => timers.clearTimeout(timer); }, + l1HitClock: onNodeJs ? () => process.hrtime.bigint() : undefined, }; /** diff --git a/packages/cachekit/src/cache.write-behind.test.ts b/packages/cachekit/src/cache.write-behind.test.ts index 56e6f34c..d6cbc439 100644 --- a/packages/cachekit/src/cache.write-behind.test.ts +++ b/packages/cachekit/src/cache.write-behind.test.ts @@ -104,7 +104,8 @@ function spyRecordOperation(cache: SecureCache) { } const outcomes = (cache: SecureCache): { l1Hits: number; l2Hits: number; misses: number } => - (cache as unknown as { telemetry: { l1Hits: number; l2Hits: number; misses: number } }).telemetry; + (cache as unknown as { hitCounters: { l1Hits: number; l2Hits: number; misses: number } }) + .hitCounters; describe('write-behind', () => { const caches: SecureCache[] = []; diff --git a/packages/cachekit/src/intents-core.ts b/packages/cachekit/src/intents-core.ts index 3b76d2d2..874ac0ef 100644 --- a/packages/cachekit/src/intents-core.ts +++ b/packages/cachekit/src/intents-core.ts @@ -148,6 +148,11 @@ export interface IOOptions extends BaseIntentOptions { apiUrl?: string; /** Per-attempt request timeout in milliseconds (default: 5000) */ timeout?: number; + /** + * The telemetry switch: `false` turns off the optional metrics headers, + * as `CACHEKIT_TELEMETRY=false` does. See CachekitIOBackendConfig.telemetry. + */ + telemetry?: boolean; /** Optional encryption config for zero-knowledge mode */ encryption?: EncryptionConfig; /** @@ -334,6 +339,7 @@ export function buildIntents( apiKey: apiKeyOption, apiUrl, timeout, + telemetry, ttl, l1, encryption, @@ -359,6 +365,7 @@ export function buildIntents( apiKey, apiUrl, timeout, + telemetry, }, defaultTtl: ttl ?? 3600, l1: withFullL1Defaults(l1), diff --git a/packages/cachekit/src/workers/cloudflare-workers.d.ts b/packages/cachekit/src/workers/cloudflare-workers.d.ts new file mode 100644 index 00000000..351f4298 --- /dev/null +++ b/packages/cachekit/src/workers/cloudflare-workers.d.ts @@ -0,0 +1,6 @@ +// The one export of workerd's built-in module the Workers runtime reads. The +// module is typed here rather than through @cloudflare/workers-types, which +// the package does not depend on. +declare module 'cloudflare:workers' { + export const env: Record | null | undefined; +} diff --git a/packages/cachekit/src/workers/runtime.ts b/packages/cachekit/src/workers/runtime.ts index 0b506b34..58ffba96 100644 --- a/packages/cachekit/src/workers/runtime.ts +++ b/packages/cachekit/src/workers/runtime.ts @@ -22,6 +22,7 @@ import { deriveTenantKeys, } from '@cachekit-io/cachekit-core-wasm'; import * as core from '@cachekit-io/cachekit-core-wasm'; +import * as cloudflareWorkers from 'cloudflare:workers'; import type { CacheOptions, SecureCache } from '../types/cache.js'; import type { CachekitIOBackendConfig } from '../backends/types.js'; import { cachekitio } from '../backends/cachekitio-factory.js'; @@ -30,12 +31,32 @@ import { EncryptionManagerCore, type EncryptionBindings } from '../encryption/ma import { webCryptoEncryptionBindings } from './webcrypto.js'; import { ConfigurationError } from '../errors.js'; import { KEY_GEN_MAX_SIZE } from '../constants.js'; +import { setTelemetryBindingsReader } from '../backends/telemetry.js'; import { nobleBlake2b256Hex, setKeyHashBinding, type Blake2b256Hex, } from '../serialization/blake2b.js'; +// CACHEKIT_TELEMETRY is read from the Worker's bindings as well as from +// process.env (API-94), through the env that cloudflare:workers exports. +// Under disallow_importable_env that env is empty, so a binding set to off +// would go unread: the reader throws instead, and an unreadable source is off. +// The module is read through its namespace, so a workerd without the +// importable env still loads this entry; it reads as unreadable too. +setTelemetryBindingsReader(() => { + const flags = (globalThis as { Cloudflare?: { compatibilityFlags?: Record } }) + .Cloudflare?.compatibilityFlags; + const env = cloudflareWorkers.env; + if (flags?.disallow_importable_env) { + throw new Error("disallow_importable_env hides the Worker's bindings"); + } + if (flags === undefined || env == null) { + throw new Error("this workerd does not say whether it exports the Worker's bindings"); + } + return env.CACHEKIT_TELEMETRY; +}); + let keyHashBindingTried = false; /** diff --git a/packages/cachekit/test/fixtures/npm-undici.ts b/packages/cachekit/test/fixtures/npm-undici.ts new file mode 100644 index 00000000..ad07df43 --- /dev/null +++ b/packages/cachekit/test/fixtures/npm-undici.ts @@ -0,0 +1,40 @@ +/** + * The npm undici that miniflare (from @cloudflare/vitest-plugin) brings: the + * public API an application installs a global dispatcher with, for tests that + * need genuine undici dispatchers beside the undici Node bundles. Without a + * dependency of its own, the version is miniflare's, so it is checked: tests + * written for undici 7 must fail, not pass uncovered, if a bump moves it. + */ +import { createRequire } from 'node:module'; + +export interface Closable { + close(): Promise; +} + +interface Composable extends Closable { + compose(interceptor: unknown): Closable; +} + +export interface NpmUndici { + Agent: new (options?: object) => Composable; + RetryAgent: new (agent: Closable, options: object) => Closable; + Pool: new (origin: string, options: object) => Composable; + MockAgent: new () => Closable & { + disableNetConnect(): void; + get(origin: string): { + intercept(options: object): { reply(status: number, body: string): void }; + }; + }; + interceptors: { retry(options: object): unknown }; + setGlobalDispatcher(dispatcher: Closable): void; +} + +const fromMiniflare = createRequire( + createRequire(import.meta.resolve('@cloudflare/vitest-plugin')).resolve('miniflare') +); +const { version } = fromMiniflare('undici/package.json') as { version: string }; +if (!version.startsWith('7.')) { + throw new Error(`test/fixtures/npm-undici: expected undici 7 from miniflare, got ${version}`); +} + +export const npmUndici = fromMiniflare('undici') as NpmUndici; diff --git a/packages/cachekit/test/workers/miniflare/telemetry-probe.worker.ts b/packages/cachekit/test/workers/miniflare/telemetry-probe.worker.ts new file mode 100644 index 00000000..bdb166a5 --- /dev/null +++ b/packages/cachekit/test/workers/miniflare/telemetry-probe.worker.ts @@ -0,0 +1,47 @@ +/** + * A Worker that drives the SDK's Workers entry for telemetry.miniflare.test.ts. + * Each path makes CachekitIO requests one way; the test reads the headers + * they carried off its outbound service. + */ +import { cachekitio, createCache } from '../../../src/workers/index.js'; + +const API_KEY = 'ck_test_fake-not-a-secret'; // pragma: allowlist secret +const COUNTERS = { l1Hits: 3, l2Hits: 1, misses: 1, l1Enabled: true }; + +type Flags = { Cloudflare?: { compatibilityFlags?: Record } }; + +export default { + async fetch(request: Request, env: Record): Promise { + switch (new URL(request.url).pathname) { + case '/runtime': + return Response.json({ + disallowImportableEnv: (globalThis as Flags).Cloudflare?.compatibilityFlags + ?.disallow_importable_env, + binding: env.CACHEKIT_TELEMETRY ?? null, + }); + case '/instance': + await cachekitio({ apiKey: API_KEY, metricsProvider: () => COUNTERS }).get('svc:k'); + break; + case '/instance-telemetry-true': + await cachekitio({ + apiKey: API_KEY, + telemetry: true, + metricsProvider: () => COUNTERS, + }).get('svc:k'); + break; + case '/config': + case '/io': { + const cache = request.url.endsWith('/io') + ? createCache.io({ apiKey: API_KEY, metrics: false }) + : createCache({ backend: { apiKey: API_KEY }, metrics: false }); + await cache.set('svc:k', 1); + await cache.get('svc:other'); + await cache.close(); + break; + } + default: + return new Response('unknown path', { status: 404 }); + } + return new Response('ok'); + }, +}; diff --git a/packages/cachekit/test/workers/miniflare/telemetry.miniflare.test.ts b/packages/cachekit/test/workers/miniflare/telemetry.miniflare.test.ts new file mode 100644 index 00000000..4d4e5af1 --- /dev/null +++ b/packages/cachekit/test/workers/miniflare/telemetry.miniflare.test.ts @@ -0,0 +1,143 @@ +/** + * Telemetry off through the Worker's bindings where `cloudflare:workers` does + * not export them (protocol spec/saas-api.md, API-94 and API-96). The + * @cloudflare/vitest-plugin pool cannot run under `disallow_importable_env`: + * its own runner reads the importable env. So this test bundles the real + * Workers entry (telemetry-probe.worker.ts) and runs it in workerd through + * Miniflare, and reads each request's headers off the outbound service. + */ +import { readFile } from 'node:fs/promises'; +import { createRequire } from 'node:module'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { build } from 'esbuild'; +import { beforeAll, describe, expect, it } from 'vitest'; + +interface MiniflareLike { + dispatchFetch(url: string): Promise<{ text(): Promise }>; + dispose(): Promise; +} +interface MiniflareModule { + Miniflare: new (options: unknown) => MiniflareLike; + convertV4MiniflareOptions(options: unknown): unknown; +} + +// Miniflare comes with @cloudflare/vitest-plugin, which pins its version. +const { Miniflare, convertV4MiniflareOptions } = createRequire( + import.meta.resolve('@cloudflare/vitest-plugin') +)('miniflare') as MiniflareModule; + +const OPTIONAL_METRICS = [ + 'x-cachekit-session-id', + 'x-cachekit-session-start', + 'x-cachekit-l1-hits', + 'x-cachekit-l2-hits', + 'x-cachekit-misses', + 'x-cachekit-l1-hit-rate', + 'x-cachekit-l1-latency-buckets', + 'x-cachekit-l1-latency-sum-ns', + 'x-cachekit-l1-latency-one-in', +]; +const PATHS = ['/instance', '/instance-telemetry-true', '/config', '/io']; + +let worker: string; +let wasm: Buffer; + +beforeAll(async () => { + let wasmPath: string | undefined; + const bundle = await build({ + entryPoints: [fileURLToPath(new URL('./telemetry-probe.worker.ts', import.meta.url))], + bundle: true, + write: false, + format: 'esm', + platform: 'browser', + conditions: ['workerd'], + external: ['cloudflare:*'], + logLevel: 'silent', + plugins: [ + { + // wrangler's CompiledWasm rule: the .wasm import is a WebAssembly.Module. + name: 'compiled-wasm', + setup(builder) { + builder.onResolve({ filter: /\.wasm$/ }, (args) => { + wasmPath = join(args.resolveDir, args.path); + return { path: './core.wasm', external: true }; + }); + }, + }, + ], + }); + worker = bundle.outputFiles[0].text; + wasm = await readFile(wasmPath!); +}); + +/** Run `paths` in a fresh workerd; return its runtime facts and every request's headers. */ +async function run( + compatibilityFlags: string[], + bindings: Record, + paths: string[] +) { + const sent: Record[] = []; + const mf = new Miniflare( + convertV4MiniflareOptions({ + modules: [ + { type: 'ESModule', path: 'worker.mjs', contents: worker }, + { type: 'CompiledWasm', path: 'core.wasm', contents: wasm }, + ], + compatibilityDate: '2026-07-01', + compatibilityFlags, + bindings, + outboundService: (request: Request) => { + sent.push(Object.fromEntries(request.headers)); + return request.method === 'PUT' ? new Response('{}') : new Response(null, { status: 404 }); + }, + }) + ); + try { + const runtime = JSON.parse(await (await mf.dispatchFetch('http://probe/runtime')).text()); + for (const path of paths) { + expect(await (await mf.dispatchFetch(`http://probe${path}`)).text(), path).toBe('ok'); + } + return { runtime, sent }; + } finally { + await mf.dispose(); + } +} + +function expectNoOptionalHeader(sent: Record[]): void { + expect(sent.length).toBeGreaterThanOrEqual(PATHS.length); + for (const headers of sent) { + for (const name of OPTIONAL_METRICS) expect(headers[name], name).toBeUndefined(); + // Kept with telemetry off: API-1 requires L1-Status on every ck_sdk_ request. + expect(headers['x-cachekit-l1-status']).toBeDefined(); + expect(headers['user-agent']).toMatch(/cachekit/i); + } +} + +describe('Workers telemetry switch in real workerd (API-94, API-96)', () => { + it('control: with the bindings importable and no variable, telemetry is on', async () => { + const { runtime, sent } = await run([], {}, ['/instance']); + expect(runtime).toEqual({ disallowImportableEnv: false, binding: null }); + expect(sent[0]['x-cachekit-session-id']).toBeDefined(); + expect(sent[0]['x-cachekit-l1-hits']).toBe('3'); + }); + + it.each([[[]], [['nodejs_compat']]])( + 'with disallow_importable_env and %j, a binding set to off sends no optional header', + async (extraFlags) => { + const { runtime, sent } = await run( + ['disallow_importable_env', ...extraFlags], + { CACHEKIT_TELEMETRY: 'off' }, + PATHS + ); + expect(runtime).toEqual({ disallowImportableEnv: true, binding: 'off' }); + expectNoOptionalHeader(sent); + } + ); + + it('with disallow_importable_env, telemetry is off even with no binding: it cannot be read', async () => { + const { runtime, sent } = await run(['disallow_importable_env'], {}, PATHS); + expect(runtime).toEqual({ disallowImportableEnv: true, binding: null }); + expectNoOptionalHeader(sent); + }); +}); diff --git a/packages/cachekit/test/workers/telemetry.workers.test.ts b/packages/cachekit/test/workers/telemetry.workers.test.ts new file mode 100644 index 00000000..077a4b0a --- /dev/null +++ b/packages/cachekit/test/workers/telemetry.workers.test.ts @@ -0,0 +1,70 @@ +/** + * Workers has no clock that advances during synchronous execution, so it + * sends no L1-latency header at all (protocol spec/saas-api.md, API-82), and + * it reads CACHEKIT_TELEMETRY from the Worker's bindings (API-94). + */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { env } from 'cloudflare:workers'; +import { createCache, cachekitio } from '../../src/workers/index.js'; +import { L1LatencyReport } from '../../src/backends/telemetry.js'; + +const BACKEND = { apiKey: 'ck_test_fake-not-a-secret' }; // pragma: allowlist secret +const LATENCY = [ + 'X-CacheKit-L1-Latency-Buckets', + 'X-CacheKit-L1-Latency-Sum-Ns', + 'X-CacheKit-L1-Latency-One-In', +]; + +let fetchSpy: ReturnType; +const bindings = env as unknown as Record; + +beforeEach(() => { + fetchSpy = vi.fn(async () => new Response(null, { status: 404 })); + vi.stubGlobal('fetch', fetchSpy); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + delete bindings.CACHEKIT_TELEMETRY; +}); + +function sentHeaders(): Record[] { + return (fetchSpy.mock.calls as [string, RequestInit][]).map( + ([, init]) => init.headers as Record + ); +} + +describe('telemetry on Workers', () => { + it('times no L1 hit and sends no latency header, never zeros (API-82)', async () => { + const recorded = vi.spyOn(L1LatencyReport.prototype, 'record'); + const cache = createCache({ backend: BACKEND, metrics: false }); + // Even with every call drawn for timing, none is: Workers has no clock. + // The field must exist, or the write would force nothing. + const sampling = cache as unknown as { l1LatencyThreshold: number }; + expect(sampling.l1LatencyThreshold, 'CacheImpl.l1LatencyThreshold').toBeTypeOf('number'); + sampling.l1LatencyThreshold = 2 ** 30; + const fn = cache.wrap(async (id: number) => ({ id }), { namespace: 'svc:get', ttl: 60 }); + for (let i = 0; i < 50; i++) await fn(7); + expect(recorded).not.toHaveBeenCalled(); + await cache.get('svc:other'); + const headers = sentHeaders(); + expect(headers.length).toBeGreaterThan(0); + for (const sent of headers) { + for (const name of LATENCY) expect(sent[name], name).toBeUndefined(); + } + // Telemetry itself is on: the session headers still go. + expect(headers.at(-1)!['X-CacheKit-Session-ID']).toBeDefined(); + await cache.close(); + }); + + it("reads CACHEKIT_TELEMETRY from the Worker's bindings: off wins (API-94)", async () => { + bindings.CACHEKIT_TELEMETRY = 'off'; + const backend = cachekitio(BACKEND); + await backend.get('svc:k'); + const sent = sentHeaders().at(-1)!; + expect(sent['X-CacheKit-Session-ID']).toBeUndefined(); + expect(sent['X-CacheKit-L1-Status']).toBeDefined(); + expect(backend.l1Latency).toBeUndefined(); + }); +}); diff --git a/packages/cachekit/vitest.workers.config.ts b/packages/cachekit/vitest.workers.config.ts index c7ff5036..84d47f19 100644 --- a/packages/cachekit/vitest.workers.config.ts +++ b/packages/cachekit/vitest.workers.config.ts @@ -63,38 +63,57 @@ const mockCachekitIOService = createMockCachekitIO(); * test. Without it, vite's own SSR wasm transform tries a runtime fs read, * which cannot work inside workerd; the pool's UnsafeEval binding * (newWasmModule) is miniflare's own mechanism for exactly this. + * + * test/workers/miniflare/** runs in Node instead: each test there starts its + * own workerd through Miniflare, for a configuration the pool cannot run in. */ export default defineConfig({ - plugins: [ - { - name: 'wasm-compiled-module', - enforce: 'pre', - load(id) { - const path = id.split('?')[0]; - if (!path.endsWith('.wasm')) return null; - const base64 = readFileSync(path).toString('base64'); - return ` - import { env } from 'cloudflare:test'; - const bytes = Uint8Array.from(atob(${JSON.stringify(base64)}), (c) => c.charCodeAt(0)); - export default env.__VITEST_POOL_WORKERS_UNSAFE_EVAL.newWasmModule(bytes); - `; + test: { + projects: [ + { + plugins: [ + { + name: 'wasm-compiled-module', + enforce: 'pre', + load(id) { + const path = id.split('?')[0]; + if (!path.endsWith('.wasm')) return null; + const base64 = readFileSync(path).toString('base64'); + return ` + import { env } from 'cloudflare:test'; + const bytes = Uint8Array.from(atob(${JSON.stringify(base64)}), (c) => c.charCodeAt(0)); + export default env.__VITEST_POOL_WORKERS_UNSAFE_EVAL.newWasmModule(bytes); + `; + }, + }, + cloudflareTest({ + miniflare: { + compatibilityDate: '2026-07-01', + // In-memory mock of the CachekitIO API for the backend smoke test: + // every fetch the tests make leaves workerd through this service + // (real Request/Response marshalling, binary-safe bodies). Unknown + // hosts 502 so tests can never reach the network. + outboundService: mockCachekitIOService, + // Real KV binding for the Workers KV backend tests. + kvNamespaces: ['TEST_KV'], + }, + }), + ], + test: { + name: 'workerd', + include: ['test/workers/**/*.test.ts'], + exclude: ['test/workers/miniflare/**'], + testTimeout: 10000, + }, }, - }, - cloudflareTest({ - miniflare: { - compatibilityDate: '2026-07-01', - // In-memory mock of the CachekitIO API for the backend smoke test: - // every fetch the tests make leaves workerd through this service - // (real Request/Response marshalling, binary-safe bodies). Unknown - // hosts 502 so tests can never reach the network. - outboundService: mockCachekitIOService, - // Real KV binding for the Workers KV backend tests. - kvNamespaces: ['TEST_KV'], + { + test: { + name: 'miniflare', + environment: 'node', + include: ['test/workers/miniflare/**/*.test.ts'], + testTimeout: 60000, + }, }, - }), - ], - test: { - include: ['test/workers/**/*.test.ts'], - testTimeout: 10000, + ], }, });