diff --git a/.changeset/tidy-flags-cache.md b/.changeset/tidy-flags-cache.md new file mode 100644 index 00000000..f68327df --- /dev/null +++ b/.changeset/tidy-flags-cache.md @@ -0,0 +1,5 @@ +--- +"@vercel/flags-core": minor +--- + +Add `staleIfError` in seconds to bound cached runtime reads after the first consecutive stream/poll failure or stream disconnect. The default `Infinity` preserves unlimited fallback; finite nonnegative durations (including fractional seconds) use existing evaluation defaults and errors after expiry, and `getDatafile()` follows the same allowance. Accepted updates, valid equal-version responses, or matching stream primed revisions reset the allowance. Storing fallback data does not confirm freshness or renew the failure clock. Build/offline behavior and source scheduling remain unchanged. diff --git a/packages/vercel-flags-core/CLAUDE.md b/packages/vercel-flags-core/CLAUDE.md index e31b0c92..99e506f4 100644 --- a/packages/vercel-flags-core/CLAUDE.md +++ b/packages/vercel-flags-core/CLAUDE.md @@ -42,7 +42,7 @@ src/ ``` createClient(sdkKey, options) - → Controller (state machine, owns all data tagging and source coordination) + → Controller (state machine, selects data origin and coordinates sources/cache) → StreamSource / PollingSource / BundledSource (emit raw DatafileInput) → create-raw-client (ID-based indirection for 'use cache' support) → controller-fns (lookup by ID, evaluate, report) @@ -62,7 +62,7 @@ The Controller's `unauthorized` flag gates usage tracking. It is set on a 401 fr ### Design principles -- **Sources emit raw data** — StreamSource, PollingSource, and BundledSource return/emit raw `DatafileInput`. The Controller is solely responsible for tagging data with its origin (`tagData(data, 'stream')` etc.). +- **Sources emit raw data** — StreamSource, PollingSource, and BundledSource return/emit raw `DatafileInput`. The Controller selects the source origin. For live updates, `DatafileCache.updateFromSource()` checks version acceptance before tagging and storing data; the Controller tags initial/fallback snapshots passed to `seed()`. - **BundledSource is a plain class** — unlike StreamSource and PollingSource which extend TypedEmitter, BundledSource has no event listeners. The Controller calls its methods directly and uses return values. - **Tests are black-box** — all behavioral tests go through the public API (`createClient` from `./index.default`). Mock `readBundledDefinitions` and `internalReportValue` as observable I/O. Use `fetchMock` for network assertions. - **ID-based indirection** — `controller-fns.ts` holds a `controllerInstanceMap` (Map) so that `'use cache'` wrappers in Next.js can pass serializable IDs instead of function references. @@ -99,6 +99,7 @@ type ControllerOptions = { datafile?: Datafile; // Initial datafile for immediate reads stream?: boolean | { initTimeoutMs: number }; // default: true (3000ms) polling?: boolean | { intervalMs: number; initTimeoutMs: number }; // default: true (30s interval, 3s timeout) + staleIfError?: number; // Seconds of fallback after stream/poll failure; default: Infinity buildStep?: boolean; // Override build step auto-detection metricEnvironment?: string; // Environment attached to ingested evaluation metrics waitUntil?: (promise: Promise) => void; // default: @vercel/functions waitUntil @@ -242,7 +243,7 @@ When updating tests for new behavior, preserve the strength of existing assertio ### Stream Connection - Uses fetch with streaming body (NDJSON format) -- Callbacks: `onDatafile` (new data), `onPrimed` (server confirmed revision is current), `onDisconnect` +- Callbacks: `onDatafile` (new data), `onPrimed` (server confirmed revision is current), `onDisconnect`, and `onError` (failure evidence for cache policy) - Sends `X-Revision` header with the current revision number on every connection (including reconnects), allowing the server to respond with a lightweight `primed` message instead of a full datafile when the revision is current - The `primed` message confirms the client's data is up-to-date; it resolves the init promise (like `datafile`) but does not update data — only transitions state to `streaming` - Reconnects with exponential backoff (base: 1s, max: 60s, max retries: 15) @@ -266,7 +267,7 @@ When updating tests for new behavior, preserve the strength of existing assertio ### Data Origin Tagging -The Controller tags all data with its origin using `tagData(data, origin)` from `tagged-data.ts`. Origins map to public `metrics.source` values: +The Controller selects the origin. Initial/fallback snapshots are tagged before `cache.seed()`; live stream/poll updates are tagged inside `cache.updateFromSource()` only after acceptance. Both use `tagData(data, origin)` from `tagged-data.ts`. Origins map to public `metrics.source` values: - `'stream'`, `'poll'`, `'provided'` → `'in-memory'` - `'fetched'` → `'remote'` - `'bundled'` → `'embedded'` @@ -295,17 +296,29 @@ The Controller tags all data with its origin using `tagData(data, origin)` from ### configUpdatedAt Guard -The Controller rejects incoming data (from stream or poll) if its `configUpdatedAt` is older than or equal to the current in-memory data. This prevents stale updates from overwriting newer data. Accepts the update if either side lacks a `configUpdatedAt`. +The DatafileCache rejects incoming data (from stream or poll) if its `configUpdatedAt` is older than or equal to the current in-memory data. This prevents stale updates from overwriting newer data. Accepts the update if either side lacks a `configUpdatedAt`. ### Evaluation Reporting - `internalReportValue` (defined in `lib/report-value.ts`, called from `controller-fns.ts`) reports flag evaluations to the Vercel request context - Reports are sent for all evaluations where `datafile.projectId` exists, including error cases (e.g., FLAG_NOT_FOUND) +### Cache read policy + +`DatafileCache.read()` is the only full-entry read. The cache is configured once +with the internal `staleIfErrorMs`, normalized from the public `staleIfError` +option in seconds. Evaluations and `getDatafile()` share the same serving +boundary. `hasData` and `revision` expose coordination metadata even after expiry, +so retained data is not replaced by fallback and stream reconnects can still send +`X-Revision`. `seed()` never clears failure. Accepted source updates or valid +version/revision confirmations clear it; repeated errors/disconnects do not renew +the first-error deadline. Stream opening/pings and initialization timeout alone +are not recovery/failure evidence respectively. + ### Evaluation Safety - Regex comparators (`REGEX`, `NOT_REGEX`) limit input string length to 10,000 characters to prevent ReDoS -- `read()` and `getDatafile()` return new objects with spread (never mutate `this.data`) +- `read()` and `getDatafile()` apply the same cache read policy and each return new objects with spread (never mutate cached data) ### Debug Mode diff --git a/packages/vercel-flags-core/README.md b/packages/vercel-flags-core/README.md index 3cad2db8..e51e4eb3 100644 --- a/packages/vercel-flags-core/README.md +++ b/packages/vercel-flags-core/README.md @@ -33,6 +33,46 @@ export default app; Outside Vercel, pass an SDK key explicitly: `createClient(process.env.FLAGS)`. +## Cached stream and polling reads + +`staleIfError` controls how many seconds evaluations and `getDatafile()` may use +cached flag definitions after a stream/poll failure or stream disconnect: + +```ts +const client = createClient(process.env.FLAGS!, { + staleIfError: 60, +}); +``` + +The default is `Infinity`, preserving unlimited cached fallback. Use a finite +nonnegative number of seconds to bound fallback. Fractional seconds are supported +(for example, `0.5` allows 500 milliseconds). A positive window includes its exact +deadline; `0` disables cached fallback immediately after failure. +Negative values, `NaN`, and negative infinity throw when creating the client. + +The allowance starts at the first consecutive failure. Repeated errors, +disconnects, and provided or bundled fallback data do not renew it. An accepted +source update, or a finite equal version for the same project and environment, +clears the outage. A stream `primed` message also clears it when its finite numeric +revision and identity match the cached entry. Opening a connection or receiving +a ping alone does not clear a failure. A later failure starts a new allowance. +Responses are observed in completion order, with existing version acceptance. + +After expiry, `evaluate()` returns the caller's default with reason `error`, or +throws the first failure when no default is supplied. `bulkEvaluate()` returns +an error result for each requested flag, with its default value when provided. +`getDatafile()` follows the same allowance and throws after expiry. The entry is +retained for recovery, including its revision for stream reconnection. A clean +stream close or ping timeout records `stream: disconnected` if no earlier failure +exists. `getFallbackDatafile()` remains an independent bundled-data export. + +There is no age-based expiry while the source is healthy, and reads do not trigger +an extra refresh after expiry. Build/offline behavior, source scheduling, retries, +timeouts, metrics categories, and logging are unchanged. An initialization timeout +alone does not start the allowance. Existing startup limitations remain: when +initial polling times out, no recurring interval is started, even if that in-flight +request later completes. + ## Evaluation Metrics To associate evaluation metrics with an environment, pass the diff --git a/packages/vercel-flags-core/src/controller/datafile-cache.test.ts b/packages/vercel-flags-core/src/controller/datafile-cache.test.ts new file mode 100644 index 00000000..bd20a09e --- /dev/null +++ b/packages/vercel-flags-core/src/controller/datafile-cache.test.ts @@ -0,0 +1,675 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { DatafileInput } from '../types'; +import { DatafileCache } from './datafile-cache'; +import type { TaggedData } from './tagged-data'; + +function response(overrides: Partial = {}): DatafileInput { + return { + definitions: {}, + segments: {}, + projectId: 'prj_test', + environment: 'production', + configUpdatedAt: 1, + ...overrides, + }; +} + +function data(_origin: TaggedData['_origin'] = 'provided'): TaggedData { + return { ...response(), _origin }; +} + +let errorSpy: ReturnType; +let warnSpy: ReturnType; + +beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); +}); +afterEach(() => { + try { + expect(errorSpy).not.toHaveBeenCalled(); + expect(warnSpy).not.toHaveBeenCalled(); + } finally { + vi.restoreAllMocks(); + vi.useRealTimers(); + } +}); + +describe('DatafileCache', () => { + it.each([ + 0, + 100, + Infinity, + ])('returns undefined when empty with policy %s without fetching or confirming a failure', (staleIfErrorMs) => { + const fetch = vi.spyOn(globalThis, 'fetch').mockImplementation(() => { + throw new Error('Unexpected fetch'); + }); + const cache = new DatafileCache(staleIfErrorMs); + expect(cache.hasData).toBe(false); + expect(cache.revision).toBeUndefined(); + expect(cache.read()).toBeUndefined(); + + const error = new Error('poll failed before data arrived'); + cache.fail(error); + expect(cache.tryConfirm(response())).toBe(false); + expect(cache.tryConfirm(response({ revision: 1 }), 'revision')).toBe(false); + vi.setSystemTime(1_101); + expect(cache.read()).toBeUndefined(); + const original = data(); + cache.seed(original); + expect(cache.hasData).toBe(true); + if (staleIfErrorMs === Infinity) { + expect(cache.read()).toBe(original); + } else { + expect(() => cache.read()).toThrow(error); + } + expect(fetch).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('stores data without an age-based expiry when no failure exists', () => { + const cache = new DatafileCache(0); + const original = data(); + expect(cache.seed(original)).toBeUndefined(); + expect(cache.read()).toBe(original); + + vi.setSystemTime(1_000_000); + expect(cache.read()).toBe(original); + }); + + it('starts the inclusive allowance at the first failure, not storage time', () => { + const cache = new DatafileCache(100); + const original = { ...data('poll'), revision: 42 }; + cache.seed(original); + vi.setSystemTime(2_000); + const firstError = new Error('first poll failed'); + cache.fail(firstError); + expect(cache.read()).toBe(original); + + vi.setSystemTime(2_100); + cache.fail(new Error('second poll failed')); + expect(cache.read()).toBe(original); + vi.setSystemTime(2_101); + expect(() => cache.read()).toThrow(firstError); + expect(cache.hasData).toBe(true); + expect(cache.revision).toBe(42); + expect(() => cache.read()).toThrow(firstError); + cache.clear(); + expect(cache.hasData).toBe(false); + expect(cache.revision).toBeUndefined(); + expect(cache.read()).toBeUndefined(); + }); + + it.each([0, -1])('fails immediately with policy %s', (staleIfErrorMs) => { + const cache = new DatafileCache(staleIfErrorMs); + const original = data(); + cache.seed(original); + expect(cache.read()).toBe(original); + const error = new Error('poll failed'); + cache.fail(error); + expect(() => cache.read()).toThrow(error); + + vi.setSystemTime(1_000_000); + expect(() => cache.read()).toThrow(error); + }); + + it.each([ + undefined, + Infinity, + ])('allows unlimited stale reads with policy %s', (staleIfErrorMs) => { + const cache = new DatafileCache(staleIfErrorMs); + const original = data(); + cache.seed(original); + cache.fail(new Error('poll failed')); + expect(cache.read()).toBe(original); + + vi.setSystemTime(1_000_000); + cache.fail(new Error('poll still failing')); + expect(cache.read()).toBe(original); + }); + + it('does not clear or renew failure when seeding network data', () => { + const cache = new DatafileCache(100); + cache.seed(data('poll')); + const error = new Error('poll failed'); + cache.fail(error); + vi.setSystemTime(1_050); + const replacement = data('poll'); + cache.seed(replacement); + expect(cache.read()).toBe(replacement); + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + }); + + it.each([ + 'provided', + 'bundled', + ] as const)('restores %s seeds only within the original failure deadline', (origin) => { + const cache = new DatafileCache(100); + const seed = data(origin); + cache.seed(seed); + const firstError = new Error('first poll failed'); + cache.fail(firstError); + + vi.setSystemTime(1_050); + cache.clear(); + expect(cache.hasData).toBe(false); + expect(cache.read()).toBeUndefined(); + expect(cache.tryConfirm(response())).toBe(false); + cache.fail(new Error('poll still failing')); + cache.seed(seed); + expect(cache.read()).toBe(seed); + vi.setSystemTime(1_100); + expect(cache.read()).toBe(seed); + + vi.setSystemTime(1_101); + cache.clear(); + cache.seed(seed); + cache.fail(new Error('poll failed again')); + expect(() => cache.read()).toThrow(firstError); + }); + + it('clears failure on a matching raw source response without replacing data', () => { + const cache = new DatafileCache(100); + const original = data(); + cache.seed(original); + const incoming = response(); + const firstError = new Error('first outage'); + cache.fail(firstError); + vi.setSystemTime(2_000); + expect(() => cache.read()).toThrow(firstError); + + expect(cache.tryConfirm(incoming)).toBe(true); + expect(incoming).not.toHaveProperty('_origin'); + expect(cache.read()).toBe(original); + + const nextError = new Error('next outage'); + cache.fail(nextError); + vi.setSystemTime(2_100); + expect(cache.read()).toBe(original); + vi.setSystemTime(2_101); + expect(() => cache.read()).toThrow(nextError); + }); + + it.each([ + [1, 1], + [1, '1'], + ['1', 1], + ['1', '1'], + [0, '0'], + ])('confirms equal finite versions (%s and %s)', (current, incoming) => { + const cache = new DatafileCache(0); + const original = { ...data(), configUpdatedAt: current }; + cache.seed(original); + const error = new Error('poll failed'); + cache.fail(error); + expect(() => cache.read()).toThrow(error); + + expect(cache.tryConfirm(response({ configUpdatedAt: incoming }))).toBe( + true, + ); + expect(cache.read()).toBe(original); + }); + + it.each([ + ['wrong project', { projectId: 'prj_other' }], + ['wrong environment', { environment: 'preview' }], + ['older version', { configUpdatedAt: 0 }], + ['newer version', { configUpdatedAt: 2 }], + ['missing version', { configUpdatedAt: undefined }], + ['invalid version', { configUpdatedAt: 'invalid' }], + ['NaN version', { configUpdatedAt: NaN }], + ['string NaN version', { configUpdatedAt: 'NaN' }], + ['infinite version', { configUpdatedAt: Infinity }], + ['string infinite version', { configUpdatedAt: 'Infinity' }], + ['negative infinite version', { configUpdatedAt: -Infinity }], + ] satisfies [ + string, + Partial, + ][])('rejects %s without changing storage or the failure deadline', (_, overrides) => { + const cache = new DatafileCache(100); + const original = data(); + cache.seed(original); + const error = new Error('first outage'); + cache.fail(error); + vi.setSystemTime(1_050); + + expect(cache.tryConfirm(response(overrides))).toBe(false); + expect(cache.read()).toBe(original); + vi.setSystemTime(1_100); + expect(cache.read()).toBe(original); + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + }); + + it.each([ + undefined, + 'invalid', + NaN, + 'NaN', + Infinity, + 'Infinity', + -Infinity, + ])('never confirms invalid current version %s, even for the same object', (configUpdatedAt) => { + const cache = new DatafileCache(100); + const original = { ...data(), configUpdatedAt }; + cache.seed(original); + const error = new Error('poll failed'); + cache.fail(error); + + expect(cache.tryConfirm(response())).toBe(false); + expect(cache.tryConfirm(response({ configUpdatedAt }))).toBe(false); + expect(cache.tryConfirm(original)).toBe(false); + expect(cache.read()).toBe(original); + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + }); + + describe.each([ + 0, 100, + ])('seed replacements with policy %s', (staleIfErrorMs) => { + it.each([ + undefined, + 'invalid', + NaN, + Infinity, + ])('retains failure when seeding a replacement with version %s', (configUpdatedAt) => { + const cache = new DatafileCache(staleIfErrorMs); + cache.seed(data()); + const error = new Error('poll failed'); + cache.fail(error); + vi.setSystemTime(1_050); + const replacement = { ...data('poll'), configUpdatedAt }; + cache.seed(replacement); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(replacement); + } + vi.setSystemTime(1_100); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(replacement); + } + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + }); + }); + + it('rejects an old response after storing a newer replacement', () => { + const cache = new DatafileCache(100); + cache.seed(data()); + const oldResponse = response(); + const replacement = { ...data('poll'), configUpdatedAt: 2 }; + cache.seed(replacement); + const error = new Error('replacement outage'); + cache.fail(error); + + expect(cache.tryConfirm(oldResponse)).toBe(false); + expect(cache.read()).toBe(replacement); + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + }); + + describe('revision confirmation', () => { + it.each([ + 0, 42, + ])('confirms revision %s after expiry without replacing data and starts a fresh allowance', (revision) => { + const cache = new DatafileCache(100); + const original = Object.freeze({ ...data('bundled'), revision }); + cache.seed(original); + expect(cache.read()).toBe(original); + const incoming = Object.freeze({ + revision, + projectId: 'prj_test', + environment: 'production', + }); + const firstError = new Error('first outage'); + cache.fail(firstError); + vi.setSystemTime(2_000); + expect(() => cache.read()).toThrow(firstError); + expect(cache.hasData).toBe(true); + expect(cache.revision).toBe(revision); + + expect(cache.tryConfirm(incoming)).toBe(false); + expect(() => cache.read()).toThrow(firstError); + expect(cache.tryConfirm(incoming, 'revision')).toBe(true); + expect(cache.read()).toBe(original); + expect(original._origin).toBe('bundled'); + expect(incoming).toEqual({ + revision, + projectId: 'prj_test', + environment: 'production', + }); + + const nextError = new Error('second outage'); + cache.fail(nextError); + vi.setSystemTime(2_100); + expect(cache.read()).toBe(original); + vi.setSystemTime(2_101); + expect(() => cache.read()).toThrow(nextError); + expect(cache.revision).toBe(revision); + }); + + it.each([ + ['wrong project', { projectId: 'prj_other' }], + ['wrong environment', { environment: 'preview' }], + ['older revision', { revision: 41 }], + ['newer revision', { revision: 43 }], + ['missing revision', { revision: undefined }], + ['string revision', { revision: '42' }], + ['malformed revision', { revision: 'invalid' }], + ['null revision', { revision: null }], + ['NaN revision', { revision: NaN }], + ['infinite revision', { revision: Infinity }], + ['negative infinite revision', { revision: -Infinity }], + ])('rejects %s without replacing data or changing the failure deadline', (_, overrides) => { + const cache = new DatafileCache(100); + const original = Object.freeze({ ...data('bundled'), revision: 42 }); + cache.seed(original); + // Network payloads can contain malformed revisions despite the static type. + const incoming = Object.freeze({ + configUpdatedAt: 1, + revision: 42, + projectId: 'prj_test', + environment: 'production', + ...overrides, + }) as Parameters[0]; + const error = new Error('first outage'); + cache.fail(error); + vi.setSystemTime(1_050); + + expect(cache.tryConfirm(incoming, 'revision')).toBe(false); + expect(cache.read()).toBe(original); + vi.setSystemTime(1_100); + expect(cache.read()).toBe(original); + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + expect(cache.revision).toBe(42); + expect(cache.tryConfirm(incoming, 'revision')).toBe(false); + expect(() => cache.read()).toThrow(error); + + expect(cache.tryConfirm(original, 'revision')).toBe(true); + expect(cache.read()).toBe(original); + }); + + it.each([ + undefined, + '42', + 'invalid', + null, + NaN, + Infinity, + -Infinity, + ])('never confirms invalid stored revision %s, even for the same object', (revision) => { + const cache = new DatafileCache(100); + const original = Object.freeze({ + ...data('bundled'), + revision, + }) as TaggedData; + cache.seed(original); + const error = new Error('outage'); + cache.fail(error); + vi.setSystemTime(1_050); + + expect(cache.tryConfirm({ ...original, revision: 42 }, 'revision')).toBe( + false, + ); + expect(cache.tryConfirm(original, 'revision')).toBe(false); + expect(cache.read()).toBe(original); + vi.setSystemTime(1_101); + expect(cache.tryConfirm(original, 'revision')).toBe(false); + expect(() => cache.read()).toThrow(error); + expect(cache.hasData).toBe(true); + }); + }); + + describe('updateFromSource', () => { + it.each([ + 'poll', + 'stream', + ] as const)('accepts the first %s response and clears a failure recorded while empty', (origin) => { + const cache = new DatafileCache(0); + cache.fail(new Error('failed before data arrived')); + vi.setSystemTime(2_000); + const incoming = response({ configUpdatedAt: NaN }); + const snapshot = { ...incoming }; + + expect(cache.updateFromSource(incoming, origin)).toBeUndefined(); + expect(cache.read()).toBe(incoming); + expect(incoming).toEqual({ ...snapshot, _origin: origin }); + expect(vi.getTimerCount()).toBe(0); + }); + + it.each([ + [1, 2], + [1, '2'], + ['1', 2], + ['9', '10'], + [1, undefined], + [1, 'invalid'], + [1, 'NaN'], + [1, Infinity], + [1, 'Infinity'], + [undefined, 0], + ['invalid', 0], + ['NaN', 0], + [undefined, undefined], + ['invalid', 'invalid'], + [undefined, NaN], + [NaN, undefined], + [NaN, 'invalid'], + [-Infinity, 0], + [Infinity, undefined], + ])('accepts version %s → %s and automatically starts a fresh allowance on the next failure', (current, next) => { + const cache = new DatafileCache(100); + cache.seed({ ...data('bundled'), configUpdatedAt: current }); + const firstError = new Error('first outage'); + cache.fail(firstError); + vi.setSystemTime(2_000); + expect(() => cache.read()).toThrow(firstError); + const incoming = response({ configUpdatedAt: next }); + const snapshot = { ...incoming }; + + expect(cache.updateFromSource(incoming, 'poll')).toBeUndefined(); + expect(cache.read()).toBe(incoming); + expect(incoming).toEqual({ ...snapshot, _origin: 'poll' }); + expect(cache.read()?.definitions).toBe(incoming.definitions); + + const nextError = new Error('second outage'); + cache.fail(nextError); + vi.setSystemTime(2_100); + expect(cache.read()).toBe(incoming); + vi.setSystemTime(2_101); + expect(() => cache.read()).toThrow(nextError); + }); + + it.each([ + { projectId: 'prj_other' }, + { environment: 'preview' }, + ])('preserves acceptance of a newer version despite mismatched identity %j', (overrides) => { + const cache = new DatafileCache(0); + cache.seed(data('bundled')); + cache.fail(new Error('outage')); + const incoming = response({ ...overrides, configUpdatedAt: 2 }); + + cache.updateFromSource(incoming, 'stream'); + expect(cache.read()).toBe(incoming); + expect(incoming).toEqual({ + ...response({ ...overrides, configUpdatedAt: 2 }), + _origin: 'stream', + }); + }); + + it.each([ + [1, 1], + [1, '1'], + ['1', 1], + ['1', '1'], + [0, '0'], + ['0', 0], + ])('confirms equal finite versions %s and %s without tagging or replacing either object', (current, next) => { + const cache = new DatafileCache(100); + const original = Object.freeze({ + ...data('bundled'), + configUpdatedAt: current, + }); + cache.seed(original); + const incoming = Object.freeze(response({ configUpdatedAt: next })); + const snapshot = { ...incoming }; + const error = new Error('outage'); + cache.fail(error); + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + + expect(cache.updateFromSource(incoming, 'poll')).toBeUndefined(); + expect(cache.read()).toBe(original); + expect(original._origin).toBe('bundled'); + expect(incoming).toEqual(snapshot); + expect(incoming).not.toHaveProperty('_origin'); + }); + + it.each([ + 'reused', + 'distinct', + ])('confirms a %s tagged response without changing its origin', (kind) => { + const cache = new DatafileCache(0); + const original = Object.freeze(data('bundled')); + cache.seed(original); + const incoming = + kind === 'reused' ? original : Object.freeze(data('provided')); + const snapshot = { ...incoming }; + cache.fail(new Error('outage')); + + cache.updateFromSource(incoming, 'poll'); + expect(cache.read()).toBe(original); + expect(original._origin).toBe('bundled'); + expect(incoming).toEqual(snapshot); + }); + + describe.each([ + 0, 100, + ])('rejected responses with policy %s', (staleIfErrorMs) => { + it.each([ + ['older version', { configUpdatedAt: 0 }], + ['older numeric string', { configUpdatedAt: '0' }], + ['wrong project', { projectId: 'prj_other' }], + ['wrong environment', { environment: 'preview' }], + ['numeric NaN', { configUpdatedAt: NaN }], + ['negative Infinity', { configUpdatedAt: -Infinity }], + ['string negative Infinity', { configUpdatedAt: '-Infinity' }], + ] satisfies [ + string, + Partial, + ][])('rejects %s without mutation or clearing the original error/deadline', (_, overrides) => { + const cache = new DatafileCache(staleIfErrorMs); + const original = Object.freeze(data('bundled')); + cache.seed(original); + expect(cache.read()).toBe(original); + const incoming = Object.freeze(response(overrides)); + const snapshot = { ...incoming }; + const error = new Error('first outage'); + cache.fail(error); + vi.setSystemTime(1_050); + + expect(cache.updateFromSource(incoming, 'poll')).toBeUndefined(); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(original); + } + expect(original._origin).toBe('bundled'); + expect(incoming).toEqual(snapshot); + expect(incoming).not.toHaveProperty('_origin'); + cache.fail(new Error('repeated outage')); + vi.setSystemTime(1_100); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(original); + } + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + expect(cache.tryConfirm(original)).toBe(true); + expect(cache.read()).toBe(original); + }); + + it.each([ + [NaN, 1], + [NaN, NaN], + [Infinity, 1], + [Infinity, Infinity], + [Infinity, 'Infinity'], + ['Infinity', Infinity], + [-Infinity, -Infinity], + [-Infinity, '-Infinity'], + ])('cannot recover from a rejected response with nonfinite current version %s and incoming version %s', (current, next) => { + const cache = new DatafileCache(staleIfErrorMs); + const original = Object.freeze({ + ...data('bundled'), + configUpdatedAt: current, + }); + cache.seed(original); + expect(cache.read()).toBe(original); + const incoming = Object.freeze(response({ configUpdatedAt: next })); + const snapshot = { ...incoming }; + const error = new Error('first outage'); + cache.fail(error); + vi.setSystemTime(1_050); + + cache.updateFromSource(incoming, 'poll'); + cache.updateFromSource(original, 'poll'); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(original); + } + expect(original._origin).toBe('bundled'); + expect(incoming).toEqual(snapshot); + expect(incoming).not.toHaveProperty('_origin'); + vi.setSystemTime(1_100); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(original); + } + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + }); + + it('rejects an old response after an accepted replacement without renewing its failure deadline', () => { + const cache = new DatafileCache(staleIfErrorMs); + const oldResponse = Object.freeze(data('bundled')); + cache.seed(oldResponse); + const replacement = response({ configUpdatedAt: 2 }); + cache.updateFromSource(replacement, 'poll'); + expect(cache.read()).toBe(replacement); + expect(cache.read()?._origin).toBe('poll'); + const error = new Error('replacement outage'); + cache.fail(error); + vi.setSystemTime(1_050); + + cache.updateFromSource(oldResponse, 'stream'); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(replacement); + } + expect(oldResponse._origin).toBe('bundled'); + vi.setSystemTime(1_100); + if (staleIfErrorMs === 0) { + expect(() => cache.read()).toThrow(error); + } else { + expect(cache.read()).toBe(replacement); + } + vi.setSystemTime(1_101); + expect(() => cache.read()).toThrow(error); + expect(cache.tryConfirm(replacement)).toBe(true); + expect(cache.read()).toBe(replacement); + }); + }); + }); +}); diff --git a/packages/vercel-flags-core/src/controller/datafile-cache.ts b/packages/vercel-flags-core/src/controller/datafile-cache.ts new file mode 100644 index 00000000..97752a79 --- /dev/null +++ b/packages/vercel-flags-core/src/controller/datafile-cache.ts @@ -0,0 +1,117 @@ +import type { DatafileInput } from '../types'; +import { type DataOrigin, type TaggedData, tagData } from './tagged-data'; + +type Confirmation = Pick< + DatafileInput, + 'configUpdatedAt' | 'revision' | 'projectId' | 'environment' +>; + +/** + * Parses a configUpdatedAt value (number or string) into a numeric timestamp. + * Returns undefined if the value is missing or cannot be parsed. + */ +function parseConfigUpdatedAt(value: unknown): number | undefined { + if (typeof value === 'number') return value; + if (typeof value === 'string') { + const parsed = Number(value); + return Number.isNaN(parsed) ? undefined : parsed; + } + return undefined; +} + +/** Storage and failure-relative read policy; callers provide source evidence. */ +export class DatafileCache { + private data: TaggedData | undefined; + private failure: { error: Error; startedAt: number } | undefined; + + constructor(private readonly staleIfErrorMs = Infinity) {} + + get hasData(): boolean { + return this.data !== undefined; + } + + /** Retained revisions remain available for reconnecting after serving expires. */ + get revision(): number | undefined { + return this.data?.revision; + } + + /** Stores initial or fallback data without confirming recovery from a failure. */ + seed(data: TaggedData): void { + this.data = data; + } + + /** Accepts a source update or confirms the current version without replacing it. */ + updateFromSource(incoming: DatafileInput, origin: DataOrigin): void { + if (this.isNewerData(incoming)) { + this.data = tagData(incoming, origin); + this.failure = undefined; + return; + } + this.tryConfirm(incoming); + } + + /** Confirms a same-version source response without replacing stored data. */ + tryConfirm( + incoming: Confirmation, + version: 'configUpdatedAt' | 'revision' = 'configUpdatedAt', + ): boolean { + if (!this.data) return false; + + const currentTs = + version === 'revision' + ? this.data.revision + : parseConfigUpdatedAt(this.data.configUpdatedAt); + const incomingTs = + version === 'revision' + ? incoming.revision + : parseConfigUpdatedAt(incoming.configUpdatedAt); + if ( + !Number.isFinite(currentTs) || + !Number.isFinite(incomingTs) || + currentTs !== incomingTs || + this.data.projectId !== incoming.projectId || + this.data.environment !== incoming.environment + ) { + return false; + } + + this.failure = undefined; + return true; + } + + /** Preserves existing acceptance, including missing or unparseable versions. */ + private isNewerData(incoming: DatafileInput): boolean { + if (!this.data) return true; + + const currentTs = parseConfigUpdatedAt(this.data.configUpdatedAt); + const incomingTs = parseConfigUpdatedAt(incoming.configUpdatedAt); + + if (currentTs === undefined || incomingTs === undefined) { + return true; + } + + return incomingTs > currentTs; + } + + fail(error: Error): void { + this.failure ??= { error, startedAt: Date.now() }; + } + + read(): TaggedData | undefined { + if (!this.data) return undefined; + + if (!this.failure || this.staleIfErrorMs === Infinity) return this.data; + + const withinAllowance = + this.staleIfErrorMs > 0 && + Date.now() - this.failure.startedAt <= this.staleIfErrorMs; + if (!withinAllowance) { + throw this.failure.error; + } + return this.data; + } + + clear(): void { + this.data = undefined; + } +} diff --git a/packages/vercel-flags-core/src/controller/index.ts b/packages/vercel-flags-core/src/controller/index.ts index 9d3fd69c..44a2140f 100644 --- a/packages/vercel-flags-core/src/controller/index.ts +++ b/packages/vercel-flags-core/src/controller/index.ts @@ -11,6 +11,7 @@ import type { TrackEvaluationOptions } from '../utils/usage/flags-evaluation'; import { UsageTracker } from '../utils/usage-tracker'; import { unauthorizedMessage } from './auth'; import { BundledSource } from './bundled-source'; +import { DatafileCache } from './datafile-cache'; import { fetchDatafile } from './fetch-datafile'; import { type ControllerOptions, @@ -18,7 +19,7 @@ import { normalizeOptions, } from './normalized-options'; import { PollingSource } from './polling-source'; -import { UnauthorizedError } from './stream-connection'; +import { type PrimedMessage, UnauthorizedError } from './stream-connection'; import { StreamSource } from './stream-source'; import { originToMetricsSource, type TaggedData, tagData } from './tagged-data'; @@ -27,23 +28,6 @@ export type { ControllerOptions } from './normalized-options'; export { PollingSource } from './polling-source'; export { StreamSource } from './stream-source'; -// --------------------------------------------------------------------------- -// Internal helpers -// --------------------------------------------------------------------------- - -/** - * Parses a configUpdatedAt value (number or string) into a numeric timestamp. - * Returns undefined if the value is missing or cannot be parsed. - */ -function parseConfigUpdatedAt(value: unknown): number | undefined { - if (typeof value === 'number') return value; - if (typeof value === 'string') { - const parsed = Number(value); - return Number.isNaN(parsed) ? undefined : parsed; - } - return undefined; -} - // --------------------------------------------------------------------------- // Internal types // --------------------------------------------------------------------------- @@ -97,10 +81,10 @@ export class Controller implements ControllerInterface { private state: State = 'idle'; // Data state — tagged with origin - private data: TaggedData | undefined; + private readonly cache: DatafileCache; // Memoized data spread for read() / getDatafile(). - // Rebuilt only when `this.data` reference changes (e.g. on stream/poll update). + // Rebuilt only when the cached data reference changes (e.g. on stream/poll update). // Holds the result of stripping `_origin`; metrics are appended per-call. private dataViewSource: TaggedData | undefined = undefined; private dataViewBase: DatafileInput | undefined = undefined; @@ -124,11 +108,12 @@ export class Controller implements ControllerInterface { constructor(options: ControllerOptions) { this.options = normalizeOptions(options); + this.cache = new DatafileCache(this.options.staleIfErrorMs); // Create source modules this.streamSource = new StreamSource( this.options, - () => this.data?.revision, + () => this.cache.revision, ); this.pollingSource = new PollingSource(this.options); @@ -143,7 +128,7 @@ export class Controller implements ControllerInterface { // If datafile provided, use it immediately if (this.options.datafile) { - this.data = tagData(this.options.datafile, 'provided'); + this.cache.seed(tagData(this.options.datafile, 'provided')); } this.usageTracker = new UsageTracker(this.options); @@ -152,12 +137,11 @@ export class Controller implements ControllerInterface { // Source event handlers (stored for cleanup) private onStreamData = (data: DatafileInput) => { this.unauthorized = false; - if (this.isNewerData(data)) { - this.data = tagData(data, 'stream'); - } + this.cache.updateFromSource(data, 'stream'); }; - private onStreamPrimed = () => { + private onStreamPrimed = (message: PrimedMessage) => { this.unauthorized = false; + this.cache.tryConfirm(message, 'revision'); // The server confirmed our revision is current — no new data needed. // Transition to streaming like a normal connected event. if (this.state === 'degraded' || this.state === 'initializing:stream') { @@ -170,18 +154,21 @@ export class Controller implements ControllerInterface { } }; private onStreamDisconnected = () => { + this.cache.fail(new Error('stream: disconnected')); if (this.state === 'streaming') { this.transition('degraded'); } }; + private onStreamError = (error: Error) => { + this.cache.fail(error); + }; private onPollData = (data: DatafileInput) => { this.unauthorized = false; - if (this.isNewerData(data)) { - this.data = tagData(data, 'poll'); - } + this.cache.updateFromSource(data, 'poll'); }; private onPollError = (error: Error) => { this.noteUnauthorized(error); + this.cache.fail(error); console.error('@vercel/flags-core: Poll failed:', error); }; @@ -194,6 +181,7 @@ export class Controller implements ControllerInterface { this.streamSource.on('primed', this.onStreamPrimed); this.streamSource.on('connected', this.onStreamConnected); this.streamSource.on('disconnected', this.onStreamDisconnected); + this.streamSource.on('error', this.onStreamError); this.pollingSource.on('data', this.onPollData); this.pollingSource.on('error', this.onPollError); } @@ -203,6 +191,7 @@ export class Controller implements ControllerInterface { this.streamSource.off('primed', this.onStreamPrimed); this.streamSource.off('connected', this.onStreamConnected); this.streamSource.off('disconnected', this.onStreamDisconnected); + this.streamSource.off('error', this.onStreamError); this.pollingSource.off('data', this.onPollData); this.pollingSource.off('error', this.onPollError); } @@ -252,18 +241,18 @@ export class Controller implements ControllerInterface { } // Hydrate from provided datafile if not already set (e.g., after shutdown) - if (!this.data && this.options.datafile) { - this.data = tagData(this.options.datafile, 'provided'); + if (!this.cache.hasData && this.options.datafile) { + this.cache.seed(tagData(this.options.datafile, 'provided')); } // If no data yet, try loading bundled definitions eagerly so we can // send the revision to the stream and potentially get a lightweight // "primed" response instead of a full datafile. - if (!this.data) { + if (!this.cache.hasData) { try { const bundled = await this.bundledSource.tryLoad(); if (bundled) { - this.data = tagData(bundled, 'bundled'); + this.cache.seed(tagData(bundled, 'bundled')); } } catch { // Bundled definitions not available — proceed without revision @@ -274,7 +263,7 @@ export class Controller implements ControllerInterface { // start updates. Both streaming and polling wait for initial data before // being considered initialized, so we know we have fresh data. // For no-updates (offline), return immediately since we already have usable data. - if (this.data) { + if (this.cache.hasData) { if (this.options.stream.enabled) { this.transition('initializing:stream'); await this.tryInitializeStream(); @@ -313,27 +302,23 @@ export class Controller implements ControllerInterface { */ async read(): Promise { const startTime = Date.now(); - const cacheHadDefinitions = this.data !== undefined; + const cacheHadDefinitions = this.cache.hasData; const isFirstRead = this.isFirstGetData; this.isFirstGetData = false; const [result, cacheStatus] = await this.resolveData(); - const readMs = Date.now() - startTime; - const source = originToMetricsSource(result._origin); - this.trackRead(startTime, cacheHadDefinitions, isFirstRead, source); - if (this.dataViewSource !== result) { const { _origin, ...rest } = result; this.dataViewBase = rest; this.dataViewSource = result; } - return { + const datafile = { ...(this.dataViewBase as DatafileInput), metrics: { - readMs, - source, + readMs: Date.now() - startTime, + source: originToMetricsSource(result._origin), cacheStatus, connectionState: this.isConnected ? ('connected' as const) @@ -341,6 +326,9 @@ export class Controller implements ControllerInterface { mode: this.mode, }, } satisfies Datafile; + + this.trackRead(startTime, cacheHadDefinitions, isFirstRead, datafile); + return datafile; } /** @@ -350,9 +338,10 @@ export class Controller implements ControllerInterface { this.unwireSourceEvents(); this.streamSource.stop(); this.pollingSource.stop(); - this.data = this.options.datafile - ? tagData(this.options.datafile, 'provided') - : undefined; + this.cache.clear(); + if (this.options.datafile) { + this.cache.seed(tagData(this.options.datafile, 'provided')); + } this.transition('shutdown'); await this.usageTracker.shutdown(); } @@ -366,32 +355,26 @@ export class Controller implements ControllerInterface { const startTime = Date.now(); this.isFirstGetData = false; - let result: TaggedData; + let result = this.cache.read(); let cacheStatus: Metrics['cacheStatus']; if (this.options.buildStep) { [result, cacheStatus] = await this.resolveDataForBuildStep(); - } else if (this.data) { + } else if (result) { cacheStatus = this.isConnected ? 'HIT' : 'STALE'; - result = this.data; } else { - // No in-memory data — try bundled, then one-time fetch + // Preserve snapshot loading without starting stream/poll initialization. const bundled = await this.bundledSource.tryLoad(); if (bundled) { - this.data = tagData(bundled, 'bundled'); - result = this.data; - cacheStatus = 'MISS'; + this.cache.seed(tagData(bundled, 'bundled')); } else { - // One-time fetch as last resort try { const fetched = await fetchDatafile({ host: this.options.host, auth: this.options.auth, fetch: this.options.fetch, }); - this.data = tagData(fetched, 'fetched'); - result = this.data; - cacheStatus = 'MISS'; + this.cache.seed(tagData(fetched, 'fetched')); } catch (error) { this.noteUnauthorized(error); throw this.noDefinitionsError( @@ -399,10 +382,10 @@ export class Controller implements ControllerInterface { ); } } + cacheStatus = 'MISS'; + result = this.cache.read()!; } - const source = originToMetricsSource(result._origin); - if (this.dataViewSource !== result) { const { _origin, ...rest } = result; this.dataViewBase = rest; @@ -413,7 +396,7 @@ export class Controller implements ControllerInterface { ...(this.dataViewBase as DatafileInput), metrics: { readMs: Date.now() - startTime, - source, + source: originToMetricsSource(result._origin), cacheStatus, connectionState: this.isConnected ? ('connected' as const) @@ -447,9 +430,10 @@ export class Controller implements ControllerInterface { return this.resolveDataForBuildStep(); } - if (this.data) { + const data = this.cache.read(); + if (data) { const cacheStatus = this.isConnected ? 'HIT' : 'STALE'; - return [this.data, cacheStatus]; + return [data, cacheStatus]; } return this.resolveDataWithFallbacks(); @@ -525,7 +509,7 @@ export class Controller implements ControllerInterface { if (this.options.polling.initTimeoutMs <= 0) { try { await pollPromise; - if (this.data) { + if (this.cache.hasData) { this.pollingSource.startInterval(); return true; } @@ -555,7 +539,7 @@ export class Controller implements ControllerInterface { return false; } - if (this.data) { + if (this.cache.hasData) { this.pollingSource.startInterval(); return true; } @@ -583,24 +567,25 @@ export class Controller implements ControllerInterface { * Initializes data for build step environments. */ private async initializeForBuildStep(): Promise { - if (this.data) return; + if (this.cache.hasData) return; if (!this.buildDataPromise) { this.buildDataPromise = this.loadBuildData(); } - this.data = await this.buildDataPromise; + this.cache.seed(await this.buildDataPromise); } /** * Retrieves data during build steps. * Concurrent callers share a single load promise. The first caller to - * populate `this.data` gets cacheStatus MISS; subsequent callers get HIT. + * populate the cache gets cacheStatus MISS; subsequent callers get HIT. */ private async resolveDataForBuildStep(): Promise< [TaggedData, Metrics['cacheStatus']] > { - if (this.data) { - return [this.data, 'HIT']; + const cached = this.cache.read(); + if (cached) { + return [cached, 'HIT']; } if (!this.buildDataPromise) { @@ -609,11 +594,11 @@ export class Controller implements ControllerInterface { const data = await this.buildDataPromise; - if (!this.data) { - this.data = data; - return [data, 'MISS']; + if (!this.cache.hasData) { + this.cache.seed(data); + return [this.cache.read()!, 'MISS']; } - return [this.data, 'HIT']; + return [this.cache.read()!, 'HIT']; } /** @@ -650,14 +635,14 @@ export class Controller implements ControllerInterface { private async initializeFromFallbacks(): Promise { this.transition('initializing:fallback'); - if (this.data) { + if (this.cache.hasData) { this.transition('degraded'); return; } const bundled = await this.bundledSource.tryLoad(); if (bundled) { - this.data = tagData(bundled, 'bundled'); + this.cache.seed(tagData(bundled, 'bundled')); this.transition('degraded'); return; } @@ -670,7 +655,7 @@ export class Controller implements ControllerInterface { auth: this.options.auth, fetch: this.options.fetch, }); - this.data = tagData(fetched, 'fetched'); + this.cache.seed(tagData(fetched, 'fetched')); this.transition('degraded'); return; } catch { @@ -709,16 +694,16 @@ export class Controller implements ControllerInterface { if (this.options.stream.enabled) { this.transition('initializing:stream'); const streamSuccess = await this.tryInitializeStream(); - if (streamSuccess && this.data) { + if (streamSuccess && this.cache.hasData) { this.transition('streaming'); - return [this.data, 'MISS']; + return [this.cache.read()!, 'MISS']; } } else if (this.options.polling.enabled) { this.transition('initializing:polling'); const pollingSuccess = await this.tryInitializePolling(); - if (pollingSuccess && this.data) { + if (pollingSuccess && this.cache.hasData) { this.transition('polling'); - return [this.data, 'MISS']; + return [this.cache.read()!, 'MISS']; } } @@ -726,33 +711,36 @@ export class Controller implements ControllerInterface { this.transition('initializing:fallback'); if (this.options.datafile) { - this.data = tagData(this.options.datafile, 'provided'); + this.cache.seed(tagData(this.options.datafile, 'provided')); this.transition('degraded'); - return [this.data, 'STALE']; + return [this.cache.read()!, 'STALE']; } const bundled = await this.bundledSource.tryLoad(); if (bundled) { console.warn('@vercel/flags-core: Using bundled definitions as fallback'); - this.data = tagData(bundled, 'bundled'); + this.cache.seed(tagData(bundled, 'bundled')); this.transition('degraded'); - return [this.data, 'STALE']; + return [this.cache.read()!, 'STALE']; } // Last resort: one-time fetch (only when no stream/poll configured) if (!this.options.stream.enabled && !this.options.polling.enabled) { + let fetched: DatafileInput | undefined; try { - const fetched = await fetchDatafile({ + fetched = await fetchDatafile({ host: this.options.host, auth: this.options.auth, fetch: this.options.fetch, }); - this.data = tagData(fetched, 'fetched'); - this.transition('degraded'); - return [this.data, 'MISS']; } catch { // fetch failed — fall through to throw } + if (fetched) { + this.cache.seed(tagData(fetched, 'fetched')); + this.transition('degraded'); + return [this.cache.read()!, 'MISS']; + } } throw this.noDefinitionsError( @@ -760,34 +748,6 @@ export class Controller implements ControllerInterface { ); } - // --------------------------------------------------------------------------- - // Data comparison - // --------------------------------------------------------------------------- - - /** - * Checks if the incoming data is newer than the current in-memory data. - * Returns true if the update should proceed, false if it should be skipped. - * - * Always accepts the update if: - * - There is no current data - * - The current data has no configUpdatedAt - * - The incoming data has no configUpdatedAt - * - * Skips the update only when both have configUpdatedAt and incoming is not newer. - */ - private isNewerData(incoming: DatafileInput): boolean { - if (!this.data) return true; - - const currentTs = parseConfigUpdatedAt(this.data.configUpdatedAt); - const incomingTs = parseConfigUpdatedAt(incoming.configUpdatedAt); - - if (currentTs === undefined || incomingTs === undefined) { - return true; - } - - return incomingTs > currentTs; - } - // --------------------------------------------------------------------------- // Usage tracking // --------------------------------------------------------------------------- @@ -800,14 +760,14 @@ export class Controller implements ControllerInterface { startTime: number, cacheHadDefinitions: boolean, isFirstRead: boolean, - source: Metrics['source'], + datafile: Datafile, ): void { if (this.unauthorized) return; if (this.options.buildStep && this.buildReadTracked) return; if (this.options.buildStep) this.buildReadTracked = true; const configOrigin: 'in-memory' | 'embedded' = - source === 'embedded' ? 'embedded' : 'in-memory'; + datafile.metrics.source === 'embedded' ? 'embedded' : 'in-memory'; const cacheAction: 'FOLLOWING' | 'REFRESHING' | 'NONE' = this.state === 'streaming' ? 'FOLLOWING' @@ -824,11 +784,11 @@ export class Controller implements ControllerInterface { mode: mode === 'streaming' ? 'stream' : mode === 'polling' ? 'poll' : mode, }; - const configUpdatedAt = this.data?.configUpdatedAt; + const configUpdatedAt = datafile.configUpdatedAt; if (typeof configUpdatedAt === 'number') { trackOptions.configUpdatedAt = configUpdatedAt; } - const revision = this.data?.revision; + const revision = datafile.revision; if (typeof revision === 'number') { trackOptions.revision = revision; } diff --git a/packages/vercel-flags-core/src/controller/normalized-options.ts b/packages/vercel-flags-core/src/controller/normalized-options.ts index a8320316..78a77a00 100644 --- a/packages/vercel-flags-core/src/controller/normalized-options.ts +++ b/packages/vercel-flags-core/src/controller/normalized-options.ts @@ -45,6 +45,18 @@ export type ControllerOptions = { */ polling?: boolean | PollingOptions; + /** + * How long runtime reads may use cached data after the first consecutive + * stream/poll failure or stream disconnect. Accepts nonnegative seconds or Infinity. + * Fractional seconds are supported. + * Zero disables fallback immediately; positive windows include the deadline. + * Accepted updates, matching versions, or matching stream primed revisions + * reset the allowance. Applies to evaluations and getDatafile(). + * Build/offline behavior is unchanged. + * @default Infinity + */ + staleIfError?: number; + /** * Override build step detection * - `true`: Treat as build step (use datafile/bundled only, no network) @@ -91,6 +103,7 @@ export type NormalizedOptions = { datafile: DatafileInput | undefined; stream: { enabled: boolean; initTimeoutMs: number }; polling: { enabled: boolean; intervalMs: number; initTimeoutMs: number }; + staleIfErrorMs: number; buildStep: boolean; fetch: typeof globalThis.fetch; waitUntil: WaitUntil; @@ -103,6 +116,13 @@ export type NormalizedOptions = { export function normalizeOptions( options: ControllerOptions, ): NormalizedOptions { + const staleIfError = options.staleIfError ?? Infinity; + if (typeof staleIfError !== 'number' || !(staleIfError >= 0)) { + throw new Error( + '@vercel/flags-core: staleIfError must be a nonnegative number of seconds or Infinity.', + ); + } + const autoDetectedBuildStep = process.env.CI === '1' || process.env.NEXT_PHASE === 'phase-production-build'; @@ -144,6 +164,7 @@ export function normalizeOptions( datafile: options.datafile, stream, polling, + staleIfErrorMs: staleIfError * 1000, buildStep, fetch: options.fetch ?? globalThis.fetch, waitUntil: options.waitUntil ?? defaultWaitUntil, diff --git a/packages/vercel-flags-core/src/controller/stream-connection.ts b/packages/vercel-flags-core/src/controller/stream-connection.ts index 0683deb9..c274ac61 100644 --- a/packages/vercel-flags-core/src/controller/stream-connection.ts +++ b/packages/vercel-flags-core/src/controller/stream-connection.ts @@ -48,6 +48,7 @@ export type StreamCallbacks = { onDatafile: (data: BundledDefinitions) => void; onPrimed?: (message: PrimedMessage) => void; onDisconnect?: () => void; + onError?: (error: Error) => void; }; export type StreamConfig = { @@ -71,10 +72,19 @@ export async function connectStream( callbacks: StreamCallbacks, ): Promise { const { host, abortController, fetch: fetchFn = globalThis.fetch } = config; - const { onDatafile, onPrimed, onDisconnect } = callbacks; + const { onDatafile, onPrimed, onDisconnect, onError } = callbacks; let retryCount = 0; let lastAttemptTime = 0; + const reportError = (error: unknown): void => { + if (abortController.signal.aborted) return; + onError?.( + error instanceof Error + ? error + : new Error('Unknown stream error', { cause: error }), + ); + }; + let resolveInit: () => void; let rejectInit: (error: unknown) => void; const initPromise = new Promise((resolve, reject) => { @@ -155,15 +165,17 @@ export async function connectStream( signal: connectionAbort.signal, }); - if (!response.ok) { - if (response.status === 401) { - if (!initialDataReceived) { - rejectInit!(new UnauthorizedError(config.sourceProjectId)); - } - abortController.abort(); - break; + if (!response.ok && response.status === 401) { + const error = new UnauthorizedError(config.sourceProjectId); + reportError(error); + if (!initialDataReceived) { + rejectInit!(error); } + abortController.abort(); + break; + } + if (!response.ok) { throw new Error(`stream was not ok: ${response.status}`); } @@ -264,6 +276,9 @@ export async function connectStream( if (abortController.signal.aborted) { break; } + if (!connectionAbort.signal.aborted) { + reportError(error); + } if (error instanceof TokenResolutionError && !initialDataReceived) { rejectInit!(error.cause); abortController.abort(); diff --git a/packages/vercel-flags-core/src/controller/stream-source.ts b/packages/vercel-flags-core/src/controller/stream-source.ts index 3157c1ab..593d2d8f 100644 --- a/packages/vercel-flags-core/src/controller/stream-source.ts +++ b/packages/vercel-flags-core/src/controller/stream-source.ts @@ -8,6 +8,7 @@ export type StreamSourceEvents = { primed: (message: PrimedMessage) => void; connected: () => void; disconnected: () => void; + error: (error: Error) => void; }; /** @@ -73,6 +74,9 @@ export class StreamSource extends TypedEmitter { onDisconnect: () => { this.emit('disconnected'); }, + onError: (error) => { + this.emit('error', error); + }, }, ); diff --git a/packages/vercel-flags-core/src/stale-if-error.test.ts b/packages/vercel-flags-core/src/stale-if-error.test.ts new file mode 100644 index 00000000..dac7426f --- /dev/null +++ b/packages/vercel-flags-core/src/stale-if-error.test.ts @@ -0,0 +1,513 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + type BundledDefinitions, + type CreateClientOptions, + createClient, + type FlagsClient, +} from './index.default'; +import { readBundledDefinitions } from './utils/read-bundled-definitions'; + +vi.mock('./utils/read-bundled-definitions', () => ({ + readBundledDefinitions: vi.fn(), +})); +vi.mock('./lib/report-value', () => ({ internalReportValue: vi.fn() })); + +function data(overrides: Partial = {}): BundledDefinitions { + return { + definitions: { + flagA: { environments: { production: 1 }, variants: [false, true] }, + }, + segments: {}, + environment: 'production', + projectId: 'prj_123', + configUpdatedAt: 10, + revision: 1, + digest: 'test', + ...overrides, + }; +} + +// Custom fetch supports exceptional version values without JSON coercion. +function response(value: BundledDefinitions): Response { + const result = new Response(); + result.json = async () => value; + return result; +} + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: Error) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +const poll = vi.fn<() => Promise>(); +const fetchMock = vi.fn(); +let clients: FlagsClient[]; +let errorSpy: ReturnType; +let warnSpy: ReturnType; + +function client(options: CreateClientOptions = {}): FlagsClient { + const result = createClient('vf_server_fake', { + buildStep: false, + stream: false, + polling: { intervalMs: 30_000, initTimeoutMs: 3_000 }, + fetch: fetchMock, + disableMetrics: true, + ...options, + }); + clients.push(result); + return result; +} + +function expectErrors(...errors: Error[]) { + expect(errorSpy.mock.calls).toEqual( + errors.map((error) => ['@vercel/flags-core: Poll failed:', error]), + ); + errorSpy.mockClear(); +} + +function rejectPollOnce(error: Error) { + for (let attempt = 0; attempt < 3; attempt++) { + poll.mockRejectedValueOnce(error); + } +} + +beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + clients = []; + poll.mockReset().mockImplementation(async () => response(data())); + fetchMock.mockReset().mockImplementation((input) => { + if (String(input).endsWith('/v1/datafile')) return poll(); + return Promise.resolve(new Response()); + }); + vi.mocked(readBundledDefinitions).mockReset().mockResolvedValue({ + definitions: null, + state: 'missing-file', + }); + errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); +}); + +afterEach(async () => { + try { + for (const instance of clients) await instance.shutdown(); + expect(errorSpy).not.toHaveBeenCalled(); + expect(warnSpy).not.toHaveBeenCalled(); + } finally { + vi.restoreAllMocks(); + vi.useRealTimers(); + } +}); + +describe('polling stale-if-error through the public API', () => { + it.each([ + -1, + NaN, + -Infinity, + ])('rejects invalid duration %s', (staleIfError) => { + expect(() => client({ staleIfError })).toThrow( + '@vercel/flags-core: staleIfError must be a nonnegative number of seconds or Infinity.', + ); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it.each([ + undefined, + Infinity, + ])('keeps unlimited fallback for %s', async (staleIfError) => { + const instance = client({ staleIfError }); + const initial = await instance.evaluate('flagA'); + const failure = new Error('offline'); + poll.mockRejectedValue(failure); + await vi.advanceTimersByTimeAsync(90_300); + expect(await instance.evaluate('flagA')).toEqual(initial); + expect(poll).toHaveBeenCalledTimes(10); + expectErrors(failure, failure, failure); + }); + + it('includes the finite deadline, preserves the first error, and uses existing error/default/bulk conventions', async () => { + const instance = client({ staleIfError: 30 }); + const initial = await instance.evaluate('flagA'); + expect(initial.metrics).toEqual({ + readMs: 0, + evaluationMs: 0, + source: 'in-memory', + cacheStatus: 'STALE', + connectionState: 'disconnected', + mode: 'polling', + }); + const snapshot = await instance.getDatafile(); + const first = new Error('first failure'); + const repeated = new Error('repeated failure'); + rejectPollOnce(first); + poll.mockRejectedValue(repeated); + await vi.advanceTimersByTimeAsync(30_300); + expect(await instance.evaluate('flagA')).toEqual(initial); + await vi.advanceTimersByTimeAsync(30_000); + expect(await instance.evaluate('flagA')).toEqual(initial); + await vi.advanceTimersByTimeAsync(1); + await expect(instance.evaluate('flagA')).rejects.toBe(first); + const fallback = { + value: false, + variantId: null, + reason: 'error', + errorMessage: first.message, + }; + expect(await instance.evaluate('flagA', false)).toEqual(fallback); + expect( + await instance.bulkEvaluate([ + { key: 'flagA', defaultValue: false }, + { key: 'missing' }, + ]), + ).toEqual({ + flagA: fallback, + missing: { ...fallback, value: undefined }, + }); + await expect(instance.getDatafile()).rejects.toBe(first); + expect(poll).toHaveBeenCalledTimes(7); + poll.mockResolvedValueOnce(response(data())); + await vi.advanceTimersByTimeAsync(29_699); + const retained = await instance.getDatafile(); + expect(retained).toEqual(snapshot); + expect(retained.definitions).toBe(snapshot.definitions); + expect(poll).toHaveBeenCalledTimes(8); + expectErrors(first, repeated); + }); + + it('does not expire healthy data between polls or start a poll from reads', async () => { + const instance = client({ staleIfError: 0.001 }); + const initial = await instance.evaluate('flagA'); + await vi.advanceTimersByTimeAsync(29_999); + expect(await instance.evaluate('flagA')).toEqual(initial); + expect(await instance.evaluate('flagA')).toEqual(initial); + expect(poll).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(await instance.evaluate('flagA')).toEqual(initial); + expect(poll).toHaveBeenCalledTimes(2); + }); + + it('accepts fractional seconds and expires just after the inclusive millisecond deadline', async () => { + const instance = client({ staleIfError: 0.25 }); + const initial = await instance.evaluate('flagA'); + const failure = new Error('offline'); + rejectPollOnce(failure); + await vi.advanceTimersByTimeAsync(30_300); + await vi.advanceTimersByTimeAsync(249); + expect(await instance.evaluate('flagA')).toEqual(initial); + await vi.advanceTimersByTimeAsync(1); + expect(await instance.evaluate('flagA')).toEqual(initial); + await vi.advanceTimersByTimeAsync(1); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + await expect(instance.getDatafile()).rejects.toBe(failure); + expect(poll).toHaveBeenCalledTimes(4); + expectErrors(failure); + }); + + it.each([ + 'provided', + 'bundled', + ] as const)('blocks %s seeds immediately with zero and permits equal-version recovery without replacement', async (seed) => { + const supplied = data(); + if (seed === 'bundled') { + vi.mocked(readBundledDefinitions).mockResolvedValue({ + definitions: supplied, + state: 'ok', + }); + } + const failure = new Error('initial failure'); + rejectPollOnce(failure); + const instance = client({ + staleIfError: 0, + ...(seed === 'provided' ? { datafile: supplied } : {}), + }); + const snapshot = await instance.getDatafile(); + const evaluation = instance.evaluate('flagA'); + const evaluationOutcome = expect(evaluation).rejects.toBe(failure); + await vi.advanceTimersByTimeAsync(300); + await evaluationOutcome; + await expect(instance.getDatafile()).rejects.toBe(failure); + expect(Date.now()).toBe(300); + expect(snapshot.definitions).toBe(supplied.definitions); + expect(snapshot.metrics.source).toBe( + seed === 'provided' ? 'in-memory' : 'embedded', + ); + await vi.advanceTimersByTimeAsync(30_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).toBe( + snapshot.definitions, + ); + expect(poll).toHaveBeenCalledTimes(4); + expectErrors(failure); + }); + + it.each([ + 10, + '10', + 11, + undefined, + 'invalid', + ])('recovers on accepted or confirmed version %s and starts a second outage', async (version) => { + const instance = client({ staleIfError: 0.1 }); + await instance.evaluate('flagA'); + const snapshot = await instance.getDatafile(); + const failure = new Error('offline'); + poll.mockRejectedValue(failure); + await vi.advanceTimersByTimeAsync(30_401); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + const updated = data({ configUpdatedAt: version as number }); + poll.mockResolvedValueOnce(response(updated)); + await vi.advanceTimersByTimeAsync(29_599); + expect((await instance.evaluate('flagA')).value).toBe(true); + const recovered = await instance.getDatafile(); + expect(recovered.definitions).toBe( + version === 10 || version === '10' + ? snapshot.definitions + : updated.definitions, + ); + await vi.advanceTimersByTimeAsync(30_400); + expect((await instance.evaluate('flagA')).value).toBe(true); + await vi.advanceTimersByTimeAsync(1); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + expect(poll).toHaveBeenCalledTimes(8); + expectErrors(failure, failure); + }); + + it('recovers when polling reuses the same bundled object without changing its embedded origin', async () => { + const supplied = data(); + vi.mocked(readBundledDefinitions).mockResolvedValue({ + definitions: supplied, + state: 'ok', + }); + const instance = client({ staleIfError: 0 }); + const initial = await instance.evaluate('flagA'); + expect(initial.value).toBe(true); + expect(initial.metrics).toEqual({ + readMs: 0, + evaluationMs: 0, + source: 'embedded', + cacheStatus: 'STALE', + connectionState: 'disconnected', + mode: 'offline', // Bundled initialization retains the existing lifecycle state. + }); + const snapshot = await instance.getDatafile(); + expect(snapshot.definitions).toBe(supplied.definitions); + expect(snapshot.metrics.source).toBe('embedded'); + const failure = new Error('polling outage'); + rejectPollOnce(failure); + await vi.advanceTimersByTimeAsync(30_300); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + + // Return the exact bundled object, including the origin attached at startup. + poll.mockResolvedValueOnce(response(supplied)); + await vi.advanceTimersByTimeAsync(29_700); + const recovered = await instance.evaluate('flagA'); + expect(recovered).toEqual(initial); + const retained = await instance.getDatafile(); + expect(retained).toEqual(snapshot); + expect(retained.definitions).toBe(supplied.definitions); + expect(retained.segments).toBe(supplied.segments); + expect(poll).toHaveBeenCalledTimes(5); + expectErrors(failure); + }); + + it.each([ + { configUpdatedAt: 9 }, + { projectId: 'other' }, + { environment: 'preview' }, + { configUpdatedAt: NaN }, + { configUpdatedAt: -Infinity }, + ])('does not confirm rejected data %j', async (override) => { + const instance = client({ staleIfError: 0 }); + await instance.evaluate('flagA'); + const snapshot = await instance.getDatafile(); + const failure = new Error('offline'); + rejectPollOnce(failure); + poll.mockResolvedValue(response(data(override))); + await vi.advanceTimersByTimeAsync(60_000); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + await expect(instance.getDatafile()).rejects.toBe(failure); + expect(poll).toHaveBeenCalledTimes(5); + poll.mockResolvedValueOnce(response(data())); + await vi.advanceTimersByTimeAsync(30_000); + expect((await instance.getDatafile()).definitions).toBe( + snapshot.definitions, + ); + expect(poll).toHaveBeenCalledTimes(6); + expectErrors(failure); + }); + + it.each([ + NaN, + Infinity, + -Infinity, + ])('does not confirm equal nonfinite version %s', async (configUpdatedAt) => { + poll.mockResolvedValue(response(data({ configUpdatedAt }))); + const instance = client({ staleIfError: 0 }); + await instance.evaluate('flagA'); + const failure = new Error('offline'); + rejectPollOnce(failure); + await vi.advanceTimersByTimeAsync(60_000); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + expectErrors(failure); + }); + + it('retains main acceptance of a newer mismatched identity and positive Infinity', async () => { + const instance = client({ staleIfError: 0 }); + await instance.evaluate('flagA'); + const failure = new Error('offline'); + rejectPollOnce(failure); + const accepted = data({ configUpdatedAt: Infinity, projectId: 'other' }); + poll.mockResolvedValue(response(accepted)); + await vi.advanceTimersByTimeAsync(60_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).toBe( + accepted.definitions, + ); + expectErrors(failure); + }); + + it('does not start SIE at initialization timeout; a late actual error does', async () => { + const pending = deferred(); + poll.mockReturnValue(pending.promise); + const instance = client({ staleIfError: 0, datafile: data() }); + const evaluation = instance.evaluate('flagA'); + await vi.advanceTimersByTimeAsync(3_000); + expect((await evaluation).value).toBe(true); + expect(warnSpy.mock.calls).toEqual([ + [ + '@vercel/flags-core: Polling initialization timeout, falling back while continuing to poll in the background', + ], + ]); + warnSpy.mockClear(); + await vi.advanceTimersByTimeAsync(1_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + const failure = new Error('late failure'); + pending.reject(failure); + await vi.advanceTimersByTimeAsync(300); + await expect(instance.evaluate('flagA')).rejects.toBe(failure); + await vi.advanceTimersByTimeAsync(60_000); + expect(poll).toHaveBeenCalledTimes(3); // Main starts no interval after this timeout. + expectErrors(failure); + }); + + it.each([ + 'provided', + 'bundled', + ] as const)('does not renew an expired allowance when restoring a %s seed', async (seed) => { + const supplied = data(); + if (seed === 'bundled') { + vi.mocked(readBundledDefinitions).mockResolvedValue({ + definitions: supplied, + state: 'ok', + }); + } + const failure = new Error('initial failure'); + poll.mockRejectedValue(failure); + const instance = client({ + staleIfError: 0.1, + ...(seed === 'provided' ? { datafile: supplied } : {}), + }); + const initial = instance.evaluate('flagA'); + await vi.advanceTimersByTimeAsync(300); + expect((await initial).value).toBe(true); + await vi.advanceTimersByTimeAsync(101); + await instance.shutdown(); + // Main permits reinitialization but does not rewire source events. + // Restoring a seed must not turn that limitation into a policy bypass. + const restored = instance.evaluate('flagA'); + const restoredOutcome = expect(restored).rejects.toBe(failure); + await vi.advanceTimersByTimeAsync(300); + await restoredOutcome; + await expect(instance.getDatafile()).rejects.toBe(failure); + expect(poll).toHaveBeenCalledTimes(6); + expectErrors(failure); + }); + + it('settles a timed-out poll before the next interval can recover', async () => { + const instance = client({ staleIfError: 0 }); + await instance.evaluate('flagA'); + const delayedBody = deferred(); + const delayedResponse = new Response(); + delayedResponse.json = () => delayedBody.promise; + poll.mockResolvedValueOnce(delayedResponse); + await vi.advanceTimersByTimeAsync(40_000); + await expect(instance.evaluate('flagA')).rejects.toThrow( + '@vercel/flags-core: Datafile fetch deadline exceeded', + ); + + // Completing the abandoned response cannot overwrite the timeout. + delayedBody.resolve(data()); + await vi.advanceTimersByTimeAsync(0); + await expect(instance.evaluate('flagA')).rejects.toThrow( + '@vercel/flags-core: Datafile fetch deadline exceeded', + ); + + await vi.advanceTimersByTimeAsync(20_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect(poll).toHaveBeenCalledTimes(3); + expect(errorSpy).toHaveBeenCalledTimes(1); + errorSpy.mockClear(); + }); + + it('keeps healthy streaming data with zero even when polling is configured', async () => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue( + new TextEncoder().encode( + `${JSON.stringify({ type: 'datafile', data: data() })}\n`, + ), + ); + }, + }); + fetchMock.mockResolvedValue(new Response(body)); + const instance = client({ stream: true, staleIfError: 0 }); + const initial = await instance.evaluate('flagA'); + expect(initial.metrics?.mode).toBe('streaming'); + await vi.advanceTimersByTimeAsync(60_000); + expect(await instance.evaluate('flagA')).toEqual(initial); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(poll).not.toHaveBeenCalled(); + }); + + it('applies zero allowance to an initial stream error without starting polling', async () => { + fetchMock.mockResolvedValue(new Response(null, { status: 401 })); + const instance = client({ + stream: true, + datafile: data(), + staleIfError: 0, + }); + await expect(instance.evaluate('flagA')).rejects.toThrow( + 'stream: unauthorized (401)', + ); + await vi.advanceTimersByTimeAsync(60_000); + await expect(instance.evaluate('flagA')).rejects.toThrow( + 'stream: unauthorized (401)', + ); + await expect(instance.getDatafile()).rejects.toThrow( + 'stream: unauthorized (401)', + ); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(poll).not.toHaveBeenCalled(); + }); + + it.each([ + true, + false, + ])('retains build/offline serving with zero (buildStep=%s)', async (buildStep) => { + const instance = client({ + buildStep, + polling: false, + staleIfError: 0, + datafile: data(), + }); + expect((await instance.evaluate('flagA')).value).toBe(true); + await vi.advanceTimersByTimeAsync(60_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/vercel-flags-core/src/stream-stale-if-error.test.ts b/packages/vercel-flags-core/src/stream-stale-if-error.test.ts new file mode 100644 index 00000000..7d26c7ef --- /dev/null +++ b/packages/vercel-flags-core/src/stream-stale-if-error.test.ts @@ -0,0 +1,608 @@ +import { getVercelOidcToken } from '@vercel/oidc'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + type BundledDefinitions, + type CreateClientOptions, + createClient, + type FlagsClient, +} from './index.default'; +import { readBundledDefinitions } from './utils/read-bundled-definitions'; + +vi.mock('@vercel/oidc', () => ({ getVercelOidcToken: vi.fn() })); +vi.mock('./utils/read-bundled-definitions', () => ({ + readBundledDefinitions: vi.fn(), +})); +vi.mock('./lib/report-value', () => ({ internalReportValue: vi.fn() })); + +function data(overrides: Partial = {}): BundledDefinitions { + return { + definitions: { + flagA: { environments: { production: 1 }, variants: [false, true] }, + }, + segments: {}, + environment: 'production', + projectId: 'prj_123', + configUpdatedAt: 10, + revision: 7, + digest: 'test', + ...overrides, + }; +} + +function primed(overrides: Record = {}) { + return { + type: 'primed', + revision: 7, + projectId: 'prj_123', + environment: 'production', + ...overrides, + }; +} + +// Use real NDJSON decoding, including malformed field values from the server. +function mockStream() { + let controller!: ReadableStreamDefaultController; + const body = new ReadableStream({ + start(value) { + controller = value; + }, + }); + const raw = (line: string) => { + controller.enqueue(new TextEncoder().encode(`${line}\n`)); + }; + return { + response: new Response(body), + raw, + push: (message: unknown) => raw(JSON.stringify(message)), + fail: (error: Error) => controller.error(error), + close: () => controller.close(), + }; +} + +const streamFetch = vi.fn(); +const fetchMock = vi.fn(); +let clients: FlagsClient[]; +let errorSpy: ReturnType; +let warnSpy: ReturnType; + +function client(options: CreateClientOptions = {}, oidc = false): FlagsClient { + const instance = createClient(oidc ? undefined : 'vf_server_fake', { + buildStep: false, + stream: { initTimeoutMs: 3_000 }, + // Streaming must remain exclusive even with polling configured. + polling: { intervalMs: 30_000, initTimeoutMs: 3_000 }, + fetch: fetchMock, + disableMetrics: true, + ...options, + }); + clients.push(instance); + return instance; +} + +async function start(options: CreateClientOptions = {}, oidc = false) { + const stream = mockStream(); + streamFetch.mockResolvedValueOnce(stream.response); + const instance = client(options, oidc); + const evaluation = instance.evaluate('flagA'); + stream.push({ type: 'datafile', data: data() }); + await vi.advanceTimersByTimeAsync(0); + expect(await evaluation).toMatchObject({ + value: true, + metrics: { source: 'in-memory', mode: 'streaming' }, + }); + return { instance, stream }; +} + +function expectRequests(attempts: string[], revision: string | null = '7') { + expect(streamFetch).toHaveBeenCalledTimes(attempts.length); + for (const [index, attempt] of attempts.entries()) { + const [url, init] = streamFetch.mock.calls[index]!; + expect(url).toBe('https://flags.vercel.com/v1/stream'); + const headers = new Headers(init?.headers); + expect(headers.get('X-Retry-Attempt')).toBe(attempt); + expect(headers.get('X-Revision')).toBe(index === 0 ? null : revision); + } +} + +async function expectExpired(instance: FlagsClient, error: Error) { + await expect(instance.evaluate('flagA')).rejects.toBe(error); + await expect(instance.getDatafile()).rejects.toBe(error); +} + +function expectInitTimeout() { + expect(warnSpy.mock.calls).toEqual([ + [ + '@vercel/flags-core: Stream initialization timeout, falling back while continuing to connect in the background', + ], + ]); + warnSpy.mockClear(); +} + +beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(0); + vi.spyOn(Math, 'random').mockReturnValue(0); + clients = []; + streamFetch + .mockReset() + .mockRejectedValue(new Error('unexpected stream fetch')); + fetchMock.mockReset().mockImplementation((input, init) => { + if (String(input).endsWith('/v1/stream')) return streamFetch(input, init); + return Promise.reject(new Error(`Unexpected fetch: ${String(input)}`)); + }); + vi.mocked(readBundledDefinitions).mockReset().mockResolvedValue({ + state: 'missing-file', + definitions: null, + }); + vi.mocked(getVercelOidcToken) + .mockReset() + .mockResolvedValue('test-oidc-token'); + errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); +}); + +afterEach(async () => { + try { + for (const instance of clients) await instance.shutdown(); + // Reads, reconnects, and shutdown must not introduce polling or fetches. + expect(fetchMock.mock.calls).toEqual(streamFetch.mock.calls); + expect(errorSpy).not.toHaveBeenCalled(); + expect(warnSpy).not.toHaveBeenCalled(); + } finally { + vi.restoreAllMocks(); + vi.useRealTimers(); + } +}); + +describe('stream stale-if-error through the public API', () => { + it.each([ + undefined, + Infinity, + ])('retains unlimited fallback for %s', async (staleIfError) => { + const { instance, stream } = await start({ staleIfError }); + const snapshot = await instance.getDatafile(); + const failure = new Error('offline'); + streamFetch.mockRejectedValue(failure); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(70_000); + expect(await instance.evaluate('flagA')).toMatchObject({ value: true }); + expect((await instance.getDatafile()).definitions).toBe( + snapshot.definitions, + ); + expectRequests(['0', '1', '2', '3', '4', '5', '6', '7']); + }); + + it('keeps the inclusive first-error deadline through retries, HTTP open, pings, and connected events', async () => { + const { instance, stream } = await start({ staleIfError: 3 }); + const first = new Error('first stream read failed'); + const repeated = new Error('reconnect failed'); + const reconnect = mockStream(); + streamFetch + .mockRejectedValueOnce(repeated) + .mockResolvedValueOnce(reconnect.response); + stream.fail(first); + await vi.advanceTimersByTimeAsync(0); + expect(await instance.evaluate('flagA')).toMatchObject({ value: true }); + await vi.advanceTimersByTimeAsync(999); + expectRequests(['0']); + await vi.advanceTimersByTimeAsync(1); + expectRequests(['0', '1']); + await vi.advanceTimersByTimeAsync(999); + expectRequests(['0', '1']); + await vi.advanceTimersByTimeAsync(1); + expectRequests(['0', '1', '2']); + reconnect.push({ type: 'ping' }); + // These messages emit connected but neither confirms the cached snapshot. + reconnect.push(primed({ revision: 6 })); + reconnect.push({ type: 'datafile', data: data({ configUpdatedAt: 9 }) }); + await vi.advanceTimersByTimeAsync(1_000); + expect(await instance.evaluate('flagA')).toMatchObject({ + value: true, + metrics: { mode: 'streaming', connectionState: 'connected' }, + }); + expect((await instance.getDatafile()).configUpdatedAt).toBe(10); + await vi.advanceTimersByTimeAsync(1); + await expectExpired(instance, first); + const fallback = { + value: false, + variantId: null, + reason: 'error', + errorMessage: first.message, + }; + expect(await instance.evaluate('flagA', false)).toEqual(fallback); + expect( + await instance.bulkEvaluate([ + { key: 'flagA', defaultValue: false }, + { key: 'missing' }, + ]), + ).toEqual({ + flagA: fallback, + missing: { ...fallback, value: undefined }, + }); + expectRequests(['0', '1', '2']); + }); + + it.each([ + 10, + '10', + 11, + ])('recovers on stream configUpdatedAt %s, replacing only accepted data', async (configUpdatedAt) => { + const { instance, stream } = await start({ staleIfError: 0 }); + const snapshot = await instance.getDatafile(); + const reconnect = mockStream(); + streamFetch.mockResolvedValueOnce(reconnect.response); + const failure = new Error('outage'); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure); + await vi.advanceTimersByTimeAsync(1_000); + await expectExpired(instance, failure); // HTTP 200 alone is not recovery. + reconnect.push({ + type: 'datafile', + data: data({ + configUpdatedAt: configUpdatedAt as number, + revision: 8, + definitions: { + flagA: { environments: { production: 0 }, variants: [false, true] }, + }, + }), + }); + await vi.advanceTimersByTimeAsync(0); + const replaced = configUpdatedAt === 11; + expect((await instance.evaluate('flagA')).value).toBe(!replaced); + const recovered = await instance.getDatafile(); + expect(recovered.revision).toBe(replaced ? 8 : 7); + if (replaced) { + expect(recovered.definitions).not.toBe(snapshot.definitions); + } else { + expect(recovered.definitions).toBe(snapshot.definitions); + expect(recovered.segments).toBe(snapshot.segments); + } + const second = new Error('second outage'); + reconnect.fail(second); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, second); + expectRequests(['0', '1']); + }); + + it('uses a retained revision after expiry, confirms primed without replacement, and starts a fresh second allowance', async () => { + const { instance, stream } = await start({ staleIfError: 0.1 }); + const snapshot = await instance.getDatafile(); + const reconnect = mockStream(); + streamFetch.mockResolvedValueOnce(reconnect.response); + const first = new Error('first outage'); + stream.fail(first); + await vi.advanceTimersByTimeAsync(101); + await expectExpired(instance, first); + expectRequests(['0']); + await vi.advanceTimersByTimeAsync(899); + expectRequests(['0', '1']); + await expectExpired(instance, first); + reconnect.push(primed()); + await vi.advanceTimersByTimeAsync(0); + expect((await instance.evaluate('flagA')).value).toBe(true); + const recovered = await instance.getDatafile(); + expect(recovered.definitions).toBe(snapshot.definitions); + expect(recovered.segments).toBe(snapshot.segments); + expect(recovered.metrics).toMatchObject({ + source: 'in-memory', + mode: 'streaming', + }); + const second = new Error('second outage'); + reconnect.fail(second); + await vi.advanceTimersByTimeAsync(100); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).toBe( + snapshot.definitions, + ); + await vi.advanceTimersByTimeAsync(1); + await expectExpired(instance, second); + expectRequests(['0', '1']); + }); + + it('rejects old, unequal, malformed, missing, nonfinite, or wrong-identity primed confirmations', async () => { + const { instance, stream } = await start({ staleIfError: 0 }); + const snapshot = await instance.getDatafile(); + const reconnect = mockStream(); + streamFetch.mockResolvedValueOnce(reconnect.response); + const failure = new Error('outage'); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(1_000); + for (const override of [ + { revision: 6 }, + { revision: 8 }, + { revision: '7' }, + { revision: 'invalid' }, + { revision: null }, + { revision: undefined }, + { projectId: 'other' }, + { projectId: undefined }, + { environment: 'preview' }, + { environment: undefined }, + ]) { + reconnect.push(primed(override)); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure); + } + // JSON supports overflowing numeric literals; stringify(Infinity) is null. + for (const revision of ['1e400', '-1e400']) { + reconnect.raw( + `{"type":"primed","revision":${revision},"projectId":"prj_123","environment":"production"}`, + ); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure); + } + reconnect.push(primed()); + await vi.advanceTimersByTimeAsync(0); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).toBe( + snapshot.definitions, + ); + expectRequests(['0', '1']); + }); + + it.each([ + '7', + undefined, + Infinity, + ])('cannot confirm a retained nonnumeric or nonfinite revision %s', async (revision) => { + const supplied = data({ revision: revision as number }); + const stream = mockStream(); + const reconnect = mockStream(); + streamFetch + .mockResolvedValueOnce(stream.response) + .mockResolvedValueOnce(reconnect.response); + const instance = client({ datafile: supplied, staleIfError: 0 }); + const initial = instance.evaluate('flagA'); + stream.push(primed()); + await vi.advanceTimersByTimeAsync(0); + expect((await initial).value).toBe(true); + const failure = new Error('outage'); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(1_000); + if (revision === Infinity) { + reconnect.raw( + '{"type":"primed","revision":1e400,"projectId":"prj_123","environment":"production"}', + ); + } else { + reconnect.push(primed({ revision })); + } + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure); + reconnect.push({ type: 'datafile', data: data({ configUpdatedAt: 11 }) }); + await vi.advanceTimersByTimeAsync(0); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).not.toBe( + supplied.definitions, + ); + expect(streamFetch).toHaveBeenCalledTimes(2); + }); + + it('does not confirm rejected same-version data with mismatched identity or older versions', async () => { + const { instance, stream } = await start({ staleIfError: 0 }); + const snapshot = await instance.getDatafile(); + const reconnect = mockStream(); + streamFetch.mockResolvedValueOnce(reconnect.response); + const failure = new Error('outage'); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(1_000); + for (const override of [ + { configUpdatedAt: 9 }, + { projectId: 'other' }, + { environment: 'preview' }, + ]) { + reconnect.push({ type: 'datafile', data: data(override) }); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure); + } + reconnect.push({ type: 'datafile', data: data() }); + await vi.advanceTimersByTimeAsync(0); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).toBe( + snapshot.definitions, + ); + expectRequests(['0', '1']); + }); + + it.each([ + 'provided', + 'bundled', + ] as const)('does not renew an exhausted allowance when initialization falls back to a %s seed', async (seed) => { + const supplied = data(); + if (seed === 'bundled') { + vi.mocked(readBundledDefinitions).mockResolvedValue({ + state: 'ok', + definitions: supplied, + }); + } + const failure = new Error('initial connection failed'); + const reconnect = mockStream(); + streamFetch + .mockRejectedValueOnce(failure) + .mockResolvedValueOnce(reconnect.response); + const instance = client({ + staleIfError: 0, + ...(seed === 'provided' ? { datafile: supplied } : {}), + }); + const evaluation = instance.evaluate('flagA', false); + await vi.advanceTimersByTimeAsync(3_000); + expect(await evaluation).toEqual({ + value: false, + variantId: null, + reason: 'error', + errorMessage: failure.message, + }); + expectInitTimeout(); + await expectExpired(instance, failure); + reconnect.push(primed()); + await vi.advanceTimersByTimeAsync(0); + expect((await instance.evaluate('flagA')).value).toBe(true); + const recovered = await instance.getDatafile(); + expect(recovered.definitions).toBe(supplied.definitions); + expect(recovered.metrics.source).toBe( + seed === 'provided' ? 'in-memory' : 'embedded', + ); + expect(streamFetch).toHaveBeenCalledTimes(2); + for (const [, init] of streamFetch.mock.calls) { + expect(new Headers(init?.headers).get('X-Revision')).toBe('7'); + } + }); + + it('does not start SIE on initialization timeout or ping, but does on a late failure', async () => { + const stream = mockStream(); + streamFetch.mockResolvedValueOnce(stream.response); + const supplied = data(); + const instance = client({ datafile: supplied, staleIfError: 0 }); + const initialized = vi.fn(); + const initialization = Promise.resolve(instance.initialize()).then( + initialized, + ); + await vi.advanceTimersByTimeAsync(2_999); + stream.push({ type: 'ping' }); + await vi.advanceTimersByTimeAsync(0); + expect(initialized).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + await initialization; + expect(initialized).toHaveBeenCalledOnce(); + expectInitTimeout(); + await vi.advanceTimersByTimeAsync(10_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + expect((await instance.getDatafile()).definitions).toBe( + supplied.definitions, + ); + const reconnect = mockStream(); + streamFetch.mockResolvedValueOnce(reconnect.response); + const failure = new Error('late stream failure'); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure); + // A long-lived connection has already satisfied the minimum retry gap. + expect(streamFetch).toHaveBeenCalledTimes(2); + expect( + new Headers(streamFetch.mock.calls[1]![1]?.headers).get('X-Revision'), + ).toBe('7'); + }); + + it('records a clean disconnect immediately with zero and does not renew on repeated closes', async () => { + const { instance, stream } = await start({ staleIfError: 0 }); + const reconnect = mockStream(); + streamFetch.mockResolvedValueOnce(reconnect.response); + stream.close(); + await vi.advanceTimersByTimeAsync(0); + const failure = await instance + .evaluate('flagA') + .catch((error: unknown) => error); + expect(failure).toBeInstanceOf(Error); + expect(failure).toMatchObject({ message: 'stream: disconnected' }); + await expectExpired(instance, failure as Error); + await vi.advanceTimersByTimeAsync(1_000); + reconnect.close(); + await vi.advanceTimersByTimeAsync(0); + await expectExpired(instance, failure as Error); + expectRequests(['0', '1']); + }); + + it('starts SIE only at ping timeout and keeps the original failure through another ping timeout', async () => { + const { instance } = await start({ staleIfError: 0.1 }); + const reconnect = mockStream(); + const third = mockStream(); + streamFetch + .mockResolvedValueOnce(reconnect.response) + .mockResolvedValueOnce(third.response); + await vi.advanceTimersByTimeAsync(89_999); + expect((await instance.evaluate('flagA')).value).toBe(true); + expectRequests(['0']); + await vi.advanceTimersByTimeAsync(2); + expectRequests(['0', '1']); + reconnect.push({ type: 'ping' }); + await vi.advanceTimersByTimeAsync(99); + expect((await instance.evaluate('flagA')).value).toBe(true); + await vi.advanceTimersByTimeAsync(1); + const failure = await instance + .evaluate('flagA') + .catch((error: unknown) => error); + expect(failure).toBeInstanceOf(Error); + expect(failure).toMatchObject({ message: 'stream: disconnected' }); + await expectExpired(instance, failure as Error); + await vi.advanceTimersByTimeAsync(89_901); + await expectExpired(instance, failure as Error); + expectRequests(['0', '1', '1']); + }); + + it('stops silently on terminal 401 without replacing the first error or renewing its deadline', async () => { + const { instance, stream } = await start({ staleIfError: 1 }); + const first = new Error('stream failed before unauthorized reconnect'); + streamFetch.mockResolvedValueOnce(new Response(null, { status: 401 })); + stream.fail(first); + await vi.advanceTimersByTimeAsync(1_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + await vi.advanceTimersByTimeAsync(1); + await expectExpired(instance, first); + await vi.advanceTimersByTimeAsync(60_000); + expectRequests(['0', '1']); + }); + + it('fast-fails initial 401 against a zero allowance without retries or timeout logging', async () => { + streamFetch.mockResolvedValueOnce(new Response(null, { status: 401 })); + const instance = client({ datafile: data(), staleIfError: 0 }); + const failure = await instance + .evaluate('flagA') + .catch((error: unknown) => error); + expect(failure).toBeInstanceOf(Error); + expect(failure).toMatchObject({ message: 'stream: unauthorized (401)' }); + await expectExpired(instance, failure as Error); + expect(Date.now()).toBe(0); + await vi.advanceTimersByTimeAsync(60_000); + expect(streamFetch).toHaveBeenCalledTimes(1); + }); + + it('forwards initial OIDC resolution failure without fetching or waiting for initialization timeout', async () => { + vi.mocked(getVercelOidcToken).mockRejectedValue( + new Error('OIDC unavailable'), + ); + const instance = client({ datafile: data(), staleIfError: 0 }, true); + const failure = await instance + .evaluate('flagA') + .catch((error: unknown) => error); + expect(failure).toBeInstanceOf(Error); + expect(failure).toMatchObject({ + message: 'stream: token resolution failed', + cause: expect.objectContaining({ + message: expect.stringContaining( + '@vercel/flags-core: Failed to get OIDC token.', + ), + }), + }); + await expectExpired(instance, failure as Error); + expect(Date.now()).toBe(0); + await vi.advanceTimersByTimeAsync(60_000); + expect(getVercelOidcToken).toHaveBeenCalledOnce(); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it('logs only the final error when retries are exhausted', async () => { + const { instance, stream } = await start(); + const failure = new Error('persistent connection failure'); + streamFetch.mockRejectedValue(failure); + stream.fail(failure); + await vi.advanceTimersByTimeAsync(700_000); + expect((await instance.evaluate('flagA')).value).toBe(true); + expectRequests(Array.from({ length: 16 }, (_, index) => String(index))); + expect(errorSpy.mock.calls).toEqual([ + ['@vercel/flags-core: Max retry count exceeded', failure], + ]); + errorSpy.mockClear(); + await vi.advanceTimersByTimeAsync(60_000); + expect(streamFetch).toHaveBeenCalledTimes(16); + }); + + it('cancels a pending retry on shutdown without new warnings or errors', async () => { + const { instance, stream } = await start({ staleIfError: 0 }); + stream.fail(new Error('disconnected before shutdown')); + await vi.advanceTimersByTimeAsync(0); + await instance.shutdown(); + clients = clients.filter((entry) => entry !== instance); + await vi.advanceTimersByTimeAsync(60_000); + expectRequests(['0']); + }); +});