diff --git a/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.configmap.test.ts b/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.configmap.test.ts new file mode 100644 index 0000000000..646b995a88 --- /dev/null +++ b/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.configmap.test.ts @@ -0,0 +1,101 @@ +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; + +import { integrations } from '@launchdarkly/js-server-sdk-common'; + +import LDClientNode from '../src/LDClientNode'; + +// Kubernetes updates a mounted ConfigMap or Secret with an atomic symbolic link swap: the data +// lives in a timestamped directory, `..data` links to it, and each file in the mount is a link +// into `..data`. An update writes a new timestamped directory, links `..data_tmp` to it, and +// renames `..data_tmp` over `..data`. The configured file name never appears in an event. +// Symbolic links need privileges on Windows, so the test runs elsewhere. + +const document = (value: string) => JSON.stringify({ flagValues: { flag: value } }); + +const settle = (ms: number) => + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +async function waitFor(condition: () => Promise, timeoutMs: number = 5000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + // eslint-disable-next-line no-await-in-loop + if (await condition()) { + return; + } + // eslint-disable-next-line no-await-in-loop + await settle(20); + } + throw new Error('timed out waiting for the condition'); +} + +const itOnPosix = process.platform === 'win32' ? it.skip : it; + +describe('given a file data source with automatic updates over a ConfigMap-style mount', () => { + let base: string; + let mount: string; + let client: LDClientNode | undefined; + let version = 0; + + const writeVersion = (value: string) => { + version += 1; + const name = `..2026_${version}`; + fs.mkdirSync(path.join(mount, name)); + fs.writeFileSync(path.join(mount, name, 'flags.json'), document(value)); + return name; + }; + + const swapTo = (name: string) => { + const previous = fs.readlinkSync(path.join(mount, '..data')); + fs.symlinkSync(name, path.join(mount, '..data_tmp')); + fs.renameSync(path.join(mount, '..data_tmp'), path.join(mount, '..data')); + fs.rmSync(path.join(mount, previous), { recursive: true }); + }; + + beforeEach(() => { + base = fs.mkdtempSync(path.join(os.tmpdir(), 'file-data-source-configmap-')); + mount = path.join(base, 'config'); + fs.mkdirSync(mount); + const first = writeVersion('a'); + fs.symlinkSync(first, path.join(mount, '..data')); + fs.symlinkSync(path.join('..data', 'flags.json'), path.join(mount, 'flags.json')); + }); + + afterEach(() => { + client?.close(); + client = undefined; + fs.rmSync(base, { recursive: true, force: true }); + }); + + const start = async () => { + const source = new integrations.FileDataSourceFactory({ + paths: [path.join(mount, 'flags.json')], + autoUpdate: true, + }); + client = new LDClientNode('sdk-key', { + updateProcessor: source.getFactory(), + sendEvents: false, + }); + client.on('error', () => { + // A failed load is reported here. The tests observe the flag values instead. + }); + await client.waitForInitialization({ timeout: 10 }); + }; + + const valueIs = (value: string) => async () => + (await client!.variation('flag', { key: 'user' }, 'default')) === value; + + itOnPosix('detects each symbolic link swap', async () => { + await start(); + expect(await valueIs('a')()).toBe(true); + + swapTo(writeVersion('b')); + await waitFor(valueIs('b'), 3000); + + swapTo(writeVersion('c')); + await waitFor(valueIs('c'), 3000); + }); +}); diff --git a/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.directory.test.ts b/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.directory.test.ts new file mode 100644 index 0000000000..b690d73500 --- /dev/null +++ b/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.directory.test.ts @@ -0,0 +1,84 @@ +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; + +import { integrations } from '@launchdarkly/js-server-sdk-common'; + +import LDClientNode from '../src/LDClientNode'; + +// These tests run on real files and real timers. The source waits for change notifications to +// settle for 100 ms and retries a failed load after one second. + +const document = (value: string) => JSON.stringify({ flagValues: { flag: value } }); + +const settle = (ms: number) => + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +async function waitFor(condition: () => Promise, timeoutMs: number = 5000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + // eslint-disable-next-line no-await-in-loop + if (await condition()) { + return; + } + // eslint-disable-next-line no-await-in-loop + await settle(20); + } + throw new Error('timed out waiting for the condition'); +} + +describe('given a file data source with automatic updates over a real directory', () => { + let base: string; + let directory: string; + let filePath: string; + let client: LDClientNode | undefined; + + beforeEach(() => { + base = fs.mkdtempSync(path.join(os.tmpdir(), 'file-data-source-test-')); + directory = path.join(base, 'data'); + filePath = path.join(directory, 'flags.json'); + fs.mkdirSync(directory); + }); + + afterEach(() => { + client?.close(); + client = undefined; + fs.rmSync(base, { recursive: true, force: true }); + }); + + const start = async () => { + const source = new integrations.FileDataSourceFactory({ paths: [filePath], autoUpdate: true }); + client = new LDClientNode('sdk-key', { + updateProcessor: source.getFactory(), + sendEvents: false, + }); + client.on('error', () => { + // A failed load is reported here. The tests observe the flag values instead. + }); + await client.waitForInitialization({ timeout: 10 }); + }; + + const valueIs = (value: string) => async () => + (await client!.variation('flag', { key: 'user' }, 'default')) === value; + + it('detects edits after the directory is deleted and created again', async () => { + fs.writeFileSync(filePath, document('a')); + await start(); + expect(await valueIs('a')()).toBe(true); + + fs.rmSync(directory, { recursive: true }); + await settle(300); + expect(await valueIs('a')()).toBe(true); + + // The directory comes back with the file. The retry or the new watch loads it. + fs.mkdirSync(directory); + fs.writeFileSync(filePath, document('b')); + await waitFor(valueIs('b')); + + // Only a watch on the new directory can detect a later edit. + fs.writeFileSync(filePath, document('c')); + await waitFor(valueIs('c'), 3000); + }); +}); diff --git a/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.symlink.test.ts b/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.symlink.test.ts new file mode 100644 index 0000000000..f63eeb8b29 --- /dev/null +++ b/packages/sdk/server-node/__tests__/LDClientNode.fileDataSource.symlink.test.ts @@ -0,0 +1,93 @@ +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; + +import { integrations } from '@launchdarkly/js-server-sdk-common'; + +import LDClientNode from '../src/LDClientNode'; + +// A configured path can be a symbolic link to a file in another directory. The directory that +// contains the link never reports a change to the target, so the source also watches the path +// itself, which follows the link. Symbolic links need privileges on Windows, so the test runs +// elsewhere. + +const document = (value: string) => JSON.stringify({ flagValues: { flag: value } }); + +const settle = (ms: number) => + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +async function waitFor(condition: () => Promise, timeoutMs: number = 5000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + // eslint-disable-next-line no-await-in-loop + if (await condition()) { + return; + } + // eslint-disable-next-line no-await-in-loop + await settle(20); + } + throw new Error('timed out waiting for the condition'); +} + +const itOnPosix = process.platform === 'win32' ? it.skip : it; + +describe('given a file data source whose configured path links to a file elsewhere', () => { + let base: string; + let target: string; + let link: string; + let client: LDClientNode | undefined; + + const replaceTarget = (value: string) => { + const temporary = `${target}.tmp`; + fs.writeFileSync(temporary, document(value)); + fs.renameSync(temporary, target); + }; + + beforeEach(() => { + base = fs.mkdtempSync(path.join(os.tmpdir(), 'file-data-source-symlink-')); + fs.mkdirSync(path.join(base, 'real')); + fs.mkdirSync(path.join(base, 'config')); + target = path.join(base, 'real', 'flags.json'); + link = path.join(base, 'config', 'flags.json'); + fs.writeFileSync(target, document('a')); + fs.symlinkSync(path.join('..', 'real', 'flags.json'), link); + }); + + afterEach(() => { + client?.close(); + client = undefined; + fs.rmSync(base, { recursive: true, force: true }); + }); + + const start = async () => { + const source = new integrations.FileDataSourceFactory({ paths: [link], autoUpdate: true }); + client = new LDClientNode('sdk-key', { + updateProcessor: source.getFactory(), + sendEvents: false, + }); + client.on('error', () => { + // A failed load is reported here. The tests observe the flag values instead. + }); + await client.waitForInitialization({ timeout: 10 }); + }; + + const valueIs = (value: string) => async () => + (await client!.variation('flag', { key: 'user' }, 'default')) === value; + + itOnPosix('detects edits and replacements of the target', async () => { + await start(); + expect(await valueIs('a')()).toBe(true); + + // An in-place edit of the target. + fs.writeFileSync(target, document('b')); + await waitFor(valueIs('b'), 3000); + + // An atomic replacement of the target, twice: the direct watch is renewed after each. + replaceTarget('c'); + await waitFor(valueIs('c'), 3000); + replaceTarget('d'); + await waitFor(valueIs('d'), 3000); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.formatDetection.test.ts b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.formatDetection.test.ts new file mode 100644 index 0000000000..6e330049ae --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.formatDetection.test.ts @@ -0,0 +1,136 @@ +import { ClientContext, Filesystem, WatchHandle } from '@launchdarkly/js-sdk-common'; + +import { FileDataSourceFactory } from '../../src/integrations'; +import Configuration from '../../src/options/Configuration'; +import AsyncStoreFacade from '../../src/store/AsyncStoreFacade'; +import InMemoryFeatureStore from '../../src/store/InMemoryFeatureStore'; +import VersionedDataKinds from '../../src/store/VersionedDataKinds'; +import { createBasicPlatform } from '../createBasicPlatform'; +import TestLogger from '../Logger'; + +// The file data source chooses the parser by file extension only. A .yml or .yaml file uses the +// YAML parser, and every other file uses JSON.parse, whatever the content looks like. + +const jsonDocument = '{"flagValues": {"flag1": "value1"}}'; +const yamlDocument = 'flagValues:\n flag1: value1\n'; + +class MockFilesystem implements Filesystem { + public fileData: Record = {}; + + async getFileTimestamp(): Promise { + return 0; + } + + async readFile(path: string): Promise { + if (!(path in this.fileData)) { + throw new Error('FILE NOT FOUND'); + } + return this.fileData[path]; + } + + watch(): WatchHandle { + return { close: () => {} }; + } +} + +describe('given a file data source over files with different extensions', () => { + let filesystem: MockFilesystem; + let featureStore: InMemoryFeatureStore; + let asyncFeatureStore: AsyncStoreFacade; + let errorHandler: jest.Mock; + let yamlParser: jest.Mock; + + beforeEach(() => { + jest.useFakeTimers(); + filesystem = new MockFilesystem(); + featureStore = new InMemoryFeatureStore(); + asyncFeatureStore = new AsyncStoreFacade(featureStore); + errorHandler = jest.fn(); + yamlParser = jest.fn(() => ({ flagValues: { flag1: 'from-yaml-parser' } })); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + const load = async (path: string, data: string, withYamlParser: boolean = true) => { + filesystem.fileData[path] = data; + const factory = new FileDataSourceFactory({ + paths: [path], + yamlParser: withYamlParser ? yamlParser : undefined, + }); + const source = factory.create( + new ClientContext('', new Configuration({ featureStore, logger: new TestLogger() }), { + ...createBasicPlatform(), + fileSystem: filesystem, + }), + featureStore, + () => {}, + errorHandler, + ); + source.start(); + await jest.runAllTimersAsync(); + source.close(); + }; + + it('parses a file without a YAML extension as JSON and does not consult the YAML parser', async () => { + await load('data.txt', jsonDocument); + + expect(await asyncFeatureStore.initialized()).toBe(true); + const flags = await asyncFeatureStore.all(VersionedDataKinds.Features); + expect(flags.flag1.variations).toEqual(['value1']); + expect(yamlParser).not.toHaveBeenCalled(); + expect(errorHandler).not.toHaveBeenCalled(); + }); + + it('reports a JSON error for a .json file with YAML content even when a YAML parser is configured', async () => { + await load('data.json', yamlDocument); + + expect(errorHandler).toHaveBeenCalledTimes(1); + expect(errorHandler.mock.calls[0][0].message).toMatch(/json/i); + expect(yamlParser).not.toHaveBeenCalled(); + expect(await asyncFeatureStore.initialized()).toBe(false); + }); + + it.each(['yml', 'yaml'])( + 'parses a .%s file with the YAML parser even when its content looks like JSON', + async (extension) => { + await load(`data.${extension}`, jsonDocument); + + expect(yamlParser).toHaveBeenCalledWith(jsonDocument); + const flags = await asyncFeatureStore.all(VersionedDataKinds.Features); + expect(flags.flag1.variations).toEqual(['from-yaml-parser']); + expect(errorHandler).not.toHaveBeenCalled(); + }, + ); + + it.each(['yml', 'yaml'])( + 'reports the missing parser error for a .%s file whose content looks like JSON', + async (extension) => { + await load(`data.${extension}`, jsonDocument, false); + + expect(errorHandler).toHaveBeenCalledTimes(1); + expect(errorHandler.mock.calls[0][0].message).toEqual( + `Attempted to parse yaml file (data.${extension}) without parser.`, + ); + expect(await asyncFeatureStore.initialized()).toBe(false); + }, + ); + + it('accepts a document whose members are not objects keyed by key', async () => { + await load('data.json', '{"flags": [], "flagValues": null, "segments": 0}'); + + expect(errorHandler).not.toHaveBeenCalled(); + expect(await asyncFeatureStore.initialized()).toBe(true); + expect(await asyncFeatureStore.all(VersionedDataKinds.Features)).toEqual({}); + expect(await asyncFeatureStore.all(VersionedDataKinds.Segments)).toEqual({}); + }); + + it('stores a flag entry as written, without filling a missing key from the map key', async () => { + await load('data.json', '{"flags": {"flag1": {"on": false, "version": 2}}}'); + + expect(errorHandler).not.toHaveBeenCalled(); + const flags = await asyncFeatureStore.all(VersionedDataKinds.Features); + expect(flags.flag1).toBeUndefined(); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.policy.test.ts b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.policy.test.ts new file mode 100644 index 0000000000..f44ac13bed --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.policy.test.ts @@ -0,0 +1,115 @@ +import { FileDataDocument, mergeDocuments } from '../../src/data_sources/filedata'; +import { fileDataInitializerPolicy } from '../../src/data_sources/fileDataInitilizerFDv2'; +import { fileDataSourcePolicy } from '../../src/data_sources/FileDataSource'; +import { Flag } from '../../src/evaluation/data/Flag'; + +// The policies are the one place where the file data source and the FDv2 file data +// initializer differ from the other file-based sources in how documents become data. + +const onFlag = (key: string, value: any, version: number): Flag => ({ + key, + on: true, + fallthrough: { variation: 0 }, + variations: [value], + version, +}); + +describe('given the file data source policy', () => { + const policy = fileDataSourcePolicy(); + + it('expands a flag value into a flag that is on and serves the value by fallthrough', () => { + expect(policy.makeFlagWithValue('flag', 'a', undefined)).toEqual(onFlag('flag', 'a', 1)); + }); + + it('keeps the version of the previous flag when the value is unchanged', () => { + expect(policy.makeFlagWithValue('flag', 'a', onFlag('flag', 'a', 3))).toEqual( + onFlag('flag', 'a', 3), + ); + }); + + it('increments the version when the value changes', () => { + expect(policy.makeFlagWithValue('flag', 'b', onFlag('flag', 'a', 3))).toEqual( + onFlag('flag', 'b', 4), + ); + expect(policy.makeFlagWithValue('flag', { x: 1 }, onFlag('flag', { x: 2 }, 1)).version).toBe(2); + expect(policy.makeFlagWithValue('flag', { x: 1 }, onFlag('flag', { x: 1 }, 1)).version).toBe(1); + }); + + it('carries the version forward from a previous full flag definition', () => { + const previous: Flag = { + key: 'flag', + on: false, + fallthrough: { variation: 0 }, + variations: ['a', 'b'], + version: 7, + }; + expect(policy.makeFlagWithValue('flag', 'a', previous).version).toBe(7); + expect(policy.makeFlagWithValue('flag', 'z', previous).version).toBe(8); + }); + + it('fails the load on a duplicate key with the existing message', () => { + expect(() => policy.resolveDuplicateKey('flag', 'flag1')).toThrow( + 'found duplicate key: "flag1"', + ); + expect(() => policy.resolveDuplicateKey('segment', 'seg1')).toThrow( + 'found duplicate key: "seg1"', + ); + }); + + it('treats a missing file as a failed load', () => { + expect(policy.missingFile).toEqual('fail'); + }); + + it('parses by extension and passes the YAML parser through', () => { + const yamlParser = jest.fn(() => ({ flagValues: { flag: 'yaml' } })); + const withParser = fileDataSourcePolicy(yamlParser); + expect(withParser.parseDocument('data.yaml', 'x')).toEqual({ flagValues: { flag: 'yaml' } }); + expect(withParser.parseDocument('data.json', '{"flagValues": {"flag": "json"}}')).toEqual({ + flagValues: { flag: 'json' }, + }); + expect(yamlParser).toHaveBeenCalledTimes(1); + }); +}); + +describe('given the file data initializer policy', () => { + const policy = fileDataInitializerPolicy(); + + it('expands a flag value into an on flag with version 1 regardless of the previous flag', () => { + expect(policy.makeFlagWithValue('flag', 'a', undefined)).toEqual(onFlag('flag', 'a', 1)); + expect(policy.makeFlagWithValue('flag', 'b', onFlag('flag', 'a', 3))).toEqual( + onFlag('flag', 'b', 1), + ); + }); + + it('keeps the last definition of a duplicate key', () => { + expect(policy.resolveDuplicateKey('flag', 'flag1')).toEqual('keepLast'); + expect(policy.resolveDuplicateKey('segment', 'seg1')).toEqual('keepLast'); + }); + + it('treats a missing file as a failed load', () => { + expect(policy.missingFile).toEqual('fail'); + }); +}); + +describe('given the entry keys of the two policies', () => { + it('the file data source keys an entry by its own key property, as it has always stored it', () => { + const policy = fileDataSourcePolicy(); + expect(policy.entryKey('map-key', { key: 'own-key' } as Flag)).toEqual('own-key'); + expect(policy.entryKey('map-key', {} as Flag)).toEqual('undefined'); + }); + + it('the file data source detects a duplicate by the key property across map keys', () => { + const documents: FileDataDocument[] = [ + { flags: { a: { key: 'x', version: 1 } as Flag } }, + { flags: { b: { key: 'x', version: 1 } as Flag } }, + ]; + expect(() => mergeDocuments(fileDataSourcePolicy(), documents)).toThrow( + 'found duplicate key: "x"', + ); + }); + + it('the initializer keys an entry by its map key, as it has always merged it', () => { + const policy = fileDataInitializerPolicy(); + expect(policy.entryKey('map-key', { key: 'own-key' } as Flag)).toEqual('map-key'); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.robustness.test.ts b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.robustness.test.ts new file mode 100644 index 0000000000..f020c09aa1 --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.robustness.test.ts @@ -0,0 +1,318 @@ +import { ClientContext } from '@launchdarkly/js-sdk-common'; + +import { FileDataSourceFactory } from '../../src/integrations'; +import Configuration from '../../src/options/Configuration'; +import AsyncStoreFacade from '../../src/store/AsyncStoreFacade'; +import InMemoryFeatureStore from '../../src/store/InMemoryFeatureStore'; +import VersionedDataKinds from '../../src/store/VersionedDataKinds'; +import { createBasicPlatform } from '../createBasicPlatform'; +import TestLogger, { LogLevel } from '../Logger'; +import MockFilesystem from './filedata/MockFilesystem'; + +// These tests run on real timers. The source waits for change notifications to settle for +// 100 ms and retries a failed load after one second. + +const directory = '/data'; +const path = `${directory}/flags.json`; +const document = (value: string) => JSON.stringify({ flagValues: { flag: value } }); + +const settle = (ms: number) => + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +async function waitFor(condition: () => Promise, timeoutMs: number = 5000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + // eslint-disable-next-line no-await-in-loop + if (await condition()) { + return; + } + // eslint-disable-next-line no-await-in-loop + await settle(20); + } + throw new Error('timed out waiting for the condition'); +} + +describe('given a file data source with automatic updates over a mock filesystem', () => { + let filesystem: MockFilesystem; + let featureStore: InMemoryFeatureStore; + let asyncFeatureStore: AsyncStoreFacade; + let logger: TestLogger; + let errorHandler: jest.Mock; + let initSuccessHandler: jest.Mock; + let source: ReturnType | undefined; + + beforeEach(() => { + filesystem = new MockFilesystem(); + featureStore = new InMemoryFeatureStore(); + asyncFeatureStore = new AsyncStoreFacade(featureStore); + logger = new TestLogger(); + errorHandler = jest.fn(); + initSuccessHandler = jest.fn(); + }); + + afterEach(() => { + source?.close(); + source = undefined; + }); + + const start = () => { + source = new FileDataSourceFactory({ paths: [path], autoUpdate: true }).create( + new ClientContext('', new Configuration({ featureStore, logger }), { + ...createBasicPlatform(), + fileSystem: filesystem, + }), + featureStore, + initSuccessHandler, + errorHandler, + ); + source.start(); + }; + + const flagValue = async () => + (await asyncFeatureStore.get(VersionedDataKinds.Features, 'flag'))?.variations[0]; + const flagVersion = async () => + (await asyncFeatureStore.get(VersionedDataKinds.Features, 'flag'))?.version; + const valueIs = (value: string) => async () => (await flagValue()) === value; + + it('does not leave an unhandled rejection when a file disappears during a reload', async () => { + const unhandled = jest.fn(); + process.on('unhandledRejection', unhandled); + try { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + // The file is removed and the directory reports it. The reload fails to read the file. + filesystem.remove(path); + filesystem.emit(directory, 'rename'); + await waitFor(async () => errorHandler.mock.calls.length > 0); + expect(errorHandler.mock.calls[0][0].message).toMatch(/ENOENT/); + expect(await flagValue()).toEqual('a'); + + // The file comes back without a notification. The retry loads it. + filesystem.set(path, document('b')); + await waitFor(valueIs('b')); + + await settle(50); + expect(unhandled).not.toHaveBeenCalled(); + } finally { + process.off('unhandledRejection', unhandled); + } + }); + + it('detects a file that is replaced by a rename', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + // A rename-based write replaces the file. The directory reports a rename, not a change + // to the watched file. + filesystem.set(path, document('b')); + filesystem.emit(directory, 'rename'); + + await waitFor(valueIs('b')); + }); + + it('retries a failed reload and recovers without another change notification', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + filesystem.set(path, '{"flagValues"'); + filesystem.emit(directory, 'change'); + await waitFor(async () => errorHandler.mock.calls.length > 0); + expect(errorHandler.mock.calls[0][0].message).toMatch(/json/i); + // The last good data stays in effect. + expect(await flagValue()).toEqual('a'); + + // The file is fixed with no further notification. Only the retry can observe it. + filesystem.set(path, document('b')); + await waitFor(valueIs('b')); + expect(errorHandler).toHaveBeenCalledTimes(1); + }); + + it('reports every failed edit, not only the first', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + filesystem.set(path, '{"flagValues"'); + filesystem.emit(directory, 'change'); + await waitFor(async () => errorHandler.mock.calls.length === 1); + + // A second edit that still fails is reported again. Only automatic retries are demoted. + filesystem.set(path, '{"flagValues": '); + filesystem.emit(directory, 'change'); + await waitFor(async () => errorHandler.mock.calls.length === 2); + expect(errorHandler).toHaveBeenCalledTimes(2); + }); + + it('keeps the version baseline intact across a failed reload', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + expect(await flagVersion()).toEqual(1); + + filesystem.set(path, document('b')); + filesystem.emit(directory, 'change'); + await waitFor(valueIs('b')); + expect(await flagVersion()).toEqual(2); + + // A malformed edit fails the reload. The internal state of the last good load is intact. + filesystem.set(path, '{"flagValues"'); + filesystem.emit(directory, 'change'); + await waitFor(async () => errorHandler.mock.calls.length > 0); + + // The next value change continues the version sequence from the last good load. + filesystem.set(path, document('c')); + filesystem.emit(directory, 'change'); + await waitFor(valueIs('c')); + expect(await flagVersion()).toEqual(3); + }); + + it('waits for change notifications to settle before reloading', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + const init = jest.spyOn(featureStore, 'init'); + + // Five edits arrive closer together than the settle window. + await ['b', 'c', 'd', 'e', 'f'].reduce>( + (previous, value) => + previous.then(async () => { + filesystem.set(path, document(value)); + filesystem.emit(directory, 'change'); + await settle(20); + }), + Promise.resolve(), + ); + + await waitFor(valueIs('f')); + await settle(150); + // One reload for the whole burst. + expect(init).toHaveBeenCalledTimes(1); + }); + + it('reports a file that is missing at start and loads it when it appears', async () => { + start(); + + await waitFor(async () => errorHandler.mock.calls.length > 0); + expect(errorHandler.mock.calls[0][0].message).toMatch(/ENOENT/); + expect(initSuccessHandler).not.toHaveBeenCalled(); + expect(await asyncFeatureStore.initialized()).toBe(false); + + // The file appears without a notification. The retry loads it and the source initializes. + filesystem.set(path, document('a')); + await waitFor(valueIs('a')); + expect(initSuccessHandler).toHaveBeenCalledTimes(1); + }); + + it('reports a file that is written slowly once', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + // Each retry sees more of the file and fails to parse it at a different position. + filesystem.set(path, '{"flagValues"'); + filesystem.emit(directory, 'change'); + await waitFor(async () => errorHandler.mock.calls.length > 0); + filesystem.set(path, '{"flagValues": {"flag": "b'); + await waitFor(async () => logger.getCount(LogLevel.Debug) > 0); + filesystem.set(path, '{"flagValues": {"flag": "b"}, "flags": {'); + await waitFor(async () => logger.getCount(LogLevel.Debug) > 1); + + // One report through the error handler and one error-level log for the whole write. + expect(errorHandler).toHaveBeenCalledTimes(1); + expect(logger.getCount(LogLevel.Error)).toEqual(1); + expect(await flagValue()).toEqual('a'); + + filesystem.set(path, document('b')); + await waitFor(valueIs('b')); + + // A failure of another kind is reported again. + filesystem.remove(path); + filesystem.emit(directory, 'rename'); + await waitFor(async () => errorHandler.mock.calls.length > 1); + expect(errorHandler.mock.calls[1][0].message).toMatch(/ENOENT/); + expect(logger.getCount(LogLevel.Error)).toEqual(2); + }); + + it('watches the directory again after it is deleted and recreated', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + // The directory is deleted. The platform delivers one last event for it, and the reload + // fails to read the file. The watch on the deleted directory is closed. + filesystem.removeDirectory(directory); + filesystem.emit(directory, 'rename'); + await waitFor(async () => errorHandler.mock.calls.length > 0); + await waitFor(async () => filesystem.activeWatches(directory).length === 0); + expect(await flagValue()).toEqual('a'); + + // The directory is created again with the file in it. The watch is set up again and the + // file is loaded. + filesystem.set(path, document('b')); + await waitFor(valueIs('b')); + await waitFor(async () => filesystem.activeWatches(directory).length === 1); + + // A later edit is detected through the new watch. + filesystem.set(path, document('c')); + filesystem.emit(directory, 'change'); + await waitFor(valueIs('c')); + }); + + it('closes the watch on a deleted directory when a load fails without an event', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + // A malformed edit arms the retry while the directory still exists. + filesystem.set(path, '{"flagValues"'); + filesystem.emit(directory, 'change'); + await waitFor(async () => errorHandler.mock.calls.length > 0); + expect(filesystem.activeWatches(directory)).toHaveLength(1); + + // The directory is deleted and the platform delivers no event for it. The retry fails to + // read the file, and the failed load makes the watcher check its directories. + filesystem.removeDirectory(directory); + await waitFor(async () => filesystem.activeWatches(directory).length === 0, 3000); + expect(await flagValue()).toEqual('a'); + + filesystem.set(path, document('b')); + await waitFor(valueIs('b')); + await waitFor(async () => filesystem.activeWatches(directory).length === 1); + filesystem.set(path, document('c')); + filesystem.emit(directory, 'change'); + await waitFor(valueIs('c')); + }); + + it('reloads under a flood of events for a deleted directory', async () => { + filesystem.set(path, document('a')); + start(); + await waitFor(valueIs('a')); + + // Some platforms deliver an endless stream of events for a deleted directory, faster than + // the settle window. The stream ends when the watch is closed. + filesystem.removeDirectory(directory); + let events = 0; + const flood = setInterval(() => { + filesystem.emit(directory, 'rename'); + events += 1; + }, 1); + try { + await waitFor(async () => errorHandler.mock.calls.length > 0, 2000); + expect(errorHandler.mock.calls[0][0].message).toMatch(/ENOENT/); + await waitFor(async () => filesystem.activeWatches(directory).length === 0); + expect(events).toBeGreaterThan(0); + expect(await flagValue()).toEqual('a'); + } finally { + clearInterval(flood); + } + + filesystem.set(path, document('b')); + await waitFor(valueIs('b')); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.test.ts b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.test.ts index df0a2c920c..a169f9a08a 100644 --- a/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.test.ts +++ b/packages/shared/sdk-server/__tests__/data_sources/FileDataSource.test.ts @@ -327,11 +327,15 @@ describe('given a mock filesystem and memory feature store', () => { ); expect(await asyncFeatureStore.initialized()).toBeTruthy(); - expect(filesystem.watch).toHaveBeenCalledTimes(2); + // Each file is watched directly, as before, and the directory that contains them is watched + // too. Both files are in the same directory. + expect(filesystem.watch).toHaveBeenCalledTimes(3); + expect(filesystem.watches['.'].length).toEqual(1); expect(filesystem.watches['file1.json'].length).toEqual(1); expect(filesystem.watches['file2.json'].length).toEqual(1); fds.close(); + expect(filesystem.watches['.'].length).toEqual(0); expect(filesystem.watches['file1.json'].length).toEqual(0); expect(filesystem.watches['file2.json'].length).toEqual(0); }); @@ -349,7 +353,7 @@ describe('given a mock filesystem and memory feature store', () => { // Need to update the timestamp, or it will think the file has not changed. filesystem.fileData['file1.json'] = { timestamp: 100, data: segmentOnlyJson }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); await jest.runAllTimersAsync(); const flags2 = await asyncFeatureStore.all(VersionedDataKinds.Features); @@ -372,13 +376,13 @@ describe('given a mock filesystem and memory feature store', () => { // Trigger several change callbacks. filesystem.fileData['file1.json'] = { timestamp: 100, data: segmentOnlyJson }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); filesystem.fileData['file1.json'] = { timestamp: 101, data: segmentOnlyJson }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); filesystem.fileData['file1.json'] = { timestamp: 102, data: segmentOnlyJson }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); filesystem.fileData['file1.json'] = { timestamp: 103, data: segmentOnlyJson }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); // The handling of the file loading is async, and additionally we debounce // the callback. So we have to wait a bit to account for the awaits and the debounce. @@ -398,8 +402,8 @@ describe('given a mock filesystem and memory feature store', () => { const segments = await asyncFeatureStore.all(VersionedDataKinds.Segments); expect(Object.keys(segments).length).toEqual(0); - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); await jest.runAllTimersAsync(); // Once for the start. @@ -480,7 +484,7 @@ describe('given a mock filesystem and memory feature store', () => { } }`, }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); await jest.runAllTimersAsync(); @@ -524,7 +528,7 @@ describe('given a mock filesystem and memory feature store', () => { } }`, }; - filesystem.watches['file1.json'][0].cb('change', 'file1.json'); + filesystem.watches['.'][0].cb('change', 'file1.json'); await jest.runAllTimersAsync(); diff --git a/packages/shared/sdk-server/__tests__/data_sources/fileDataInitilizerFDv2.applyFailure.test.ts b/packages/shared/sdk-server/__tests__/data_sources/fileDataInitilizerFDv2.applyFailure.test.ts new file mode 100644 index 0000000000..f1e843c7c1 --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/fileDataInitilizerFDv2.applyFailure.test.ts @@ -0,0 +1,55 @@ +import { + DataSourceErrorKind, + Filesystem, + LDPollingError, + Platform, + subsystem, +} from '@launchdarkly/js-sdk-common'; + +import { FileSystemDataSourceConfiguration } from '../../src/api'; +import FileDataInitializerFDv2 from '../../src/data_sources/fileDataInitilizerFDv2'; +import { createBasicPlatform } from '../createBasicPlatform'; +import TestLogger, { LogLevel } from '../Logger'; +import MockFilesystem from './filedata/MockFilesystem'; + +// A YAML entry with no body parses to null, and the initializer keys entries by their map key, +// so the null reaches the payload processing step. That step must report it as invalid data, +// as it always has, rather than reject a promise nobody awaits. +it('reports a flag entry that is not an object as invalid data instead of rejecting', async () => { + jest.useFakeTimers(); + const filesystem = new MockFilesystem(); + filesystem.set('/data/flags.json', '{"flags": {"my-flag": null}}'); + const platform: Platform = { + ...createBasicPlatform(), + fileSystem: filesystem as unknown as Filesystem, + }; + const logger = new TestLogger(); + const dataCallback = jest.fn(); + const statusCallback = jest.fn(); + const rejections: unknown[] = []; + const onRejection = (reason: unknown) => rejections.push(reason); + process.on('unhandledRejection', onRejection); + try { + const options: FileSystemDataSourceConfiguration = { + type: 'file', + paths: ['/data/flags.json'], + }; + const initializer = new FileDataInitializerFDv2(options, platform, logger); + initializer.start(dataCallback, statusCallback); + await jest.runAllTimersAsync(); + + const closed = statusCallback.mock.calls.find( + (call) => call[0] === subsystem.DataSourceState.Closed, + ); + expect(closed).toBeDefined(); + const [, error] = closed ?? []; + expect(error).toBeInstanceOf(LDPollingError); + expect((error as LDPollingError).kind).toBe(DataSourceErrorKind.InvalidData); + expect(dataCallback).not.toHaveBeenCalled(); + expect(logger.getCount(LogLevel.Error)).toBeGreaterThan(0); + expect(rejections).toHaveLength(0); + } finally { + process.off('unhandledRejection', onRejection); + jest.useRealTimers(); + } +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/fileDataInitilizerFDv2.formatDetection.test.ts b/packages/shared/sdk-server/__tests__/data_sources/fileDataInitilizerFDv2.formatDetection.test.ts new file mode 100644 index 0000000000..1c2dafa55d --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/fileDataInitilizerFDv2.formatDetection.test.ts @@ -0,0 +1,142 @@ +import { + DataSourceErrorKind, + Filesystem, + LDPollingError, + Platform, + subsystem, + WatchHandle, +} from '@launchdarkly/js-sdk-common'; + +import FileDataInitializerFDv2 from '../../src/data_sources/fileDataInitilizerFDv2'; +import { createBasicPlatform } from '../createBasicPlatform'; +import TestLogger, { LogLevel } from '../Logger'; + +// The FDv2 file data initializer chooses the parser by file extension only. A .yml or .yaml file +// uses the YAML parser, and every other file uses JSON.parse, whatever the content looks like. + +const jsonDocument = '{"flagValues": {"flag1": "value1"}}'; +const yamlDocument = 'flagValues:\n flag1: value1\n'; + +class MockFilesystem implements Filesystem { + public fileData: Record = {}; + + async getFileTimestamp(): Promise { + return 0; + } + + async readFile(path: string): Promise { + if (!(path in this.fileData)) { + throw new Error('FILE NOT FOUND'); + } + return this.fileData[path]; + } + + watch(): WatchHandle { + return { close: () => {} }; + } +} + +function flagValueOf(dataCallback: jest.Mock, key: string): any { + const { payload } = dataCallback.mock.calls[0][1]; + const update = payload.updates.find((item: any) => item.kind === 'flag' && item.key === key); + return update?.object.variations[0]; +} + +describe('given a file data initializer over files with different extensions', () => { + let filesystem: MockFilesystem; + let platform: Platform; + let logger: TestLogger; + let dataCallback: jest.Mock; + let statusCallback: jest.Mock; + let yamlParser: jest.Mock; + + beforeEach(() => { + jest.useFakeTimers(); + filesystem = new MockFilesystem(); + platform = { ...createBasicPlatform(), fileSystem: filesystem }; + logger = new TestLogger(); + dataCallback = jest.fn(); + statusCallback = jest.fn(); + yamlParser = jest.fn(() => ({ flagValues: { flag1: 'from-yaml-parser' } })); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + const load = async (path: string, data: string, withYamlParser: boolean = true) => { + filesystem.fileData[path] = data; + const initializer = new FileDataInitializerFDv2( + { type: 'file', paths: [path], yamlParser: withYamlParser ? yamlParser : undefined }, + platform, + logger, + ); + initializer.start(dataCallback, statusCallback); + await jest.runAllTimersAsync(); + initializer.stop(); + }; + + const expectMalformedDataFailure = () => { + expect(dataCallback).not.toHaveBeenCalled(); + const closed = statusCallback.mock.calls.find( + (call) => call[0] === subsystem.DataSourceState.Closed, + ); + expect(closed).toBeDefined(); + expect(closed![1]).toBeInstanceOf(LDPollingError); + expect(closed![1].kind).toEqual(DataSourceErrorKind.InvalidData); + expect(closed![1].message).toEqual('Malformed data in file response'); + logger.expectMessages([{ level: LogLevel.Error, matches: /File contained invalid data/ }]); + }; + + it('parses a file without a YAML extension as JSON and does not consult the YAML parser', async () => { + await load('data.txt', jsonDocument); + + expect(statusCallback).toHaveBeenCalledWith(subsystem.DataSourceState.Valid); + expect(flagValueOf(dataCallback, 'flag1')).toEqual('value1'); + expect(yamlParser).not.toHaveBeenCalled(); + }); + + it('reports malformed data for a .json file with YAML content even when a YAML parser is configured', async () => { + await load('data.json', yamlDocument); + + expectMalformedDataFailure(); + expect(yamlParser).not.toHaveBeenCalled(); + }); + + it.each(['yml', 'yaml'])( + 'parses a .%s file with the YAML parser even when its content looks like JSON', + async (extension) => { + await load(`data.${extension}`, jsonDocument); + + expect(yamlParser).toHaveBeenCalledWith(jsonDocument); + expect(statusCallback).toHaveBeenCalledWith(subsystem.DataSourceState.Valid); + expect(flagValueOf(dataCallback, 'flag1')).toEqual('from-yaml-parser'); + }, + ); + + it.each(['yml', 'yaml'])( + 'reports malformed data for a .%s file when no YAML parser is configured', + async (extension) => { + await load(`data.${extension}`, jsonDocument, false); + + expectMalformedDataFailure(); + // The logged error names the file and the missing parser. + logger.expectMessages([ + { + level: LogLevel.Error, + matches: new RegExp( + `File contained invalid data.*Attempted to parse yaml file \\(data\\.${extension}\\) without parser\\.`, + ), + }, + ]); + }, + ); + + it('accepts a document whose members are not objects keyed by key', async () => { + await load('data.json', '{"flags": [], "flagValues": null, "segments": 0}'); + + expect(statusCallback).toHaveBeenCalledWith(subsystem.DataSourceState.Valid); + expect(dataCallback).toHaveBeenCalledTimes(1); + expect(dataCallback.mock.calls[0][1].payload.updates).toEqual([]); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts new file mode 100644 index 0000000000..3ba9dbca9d --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts @@ -0,0 +1,511 @@ +import { directoryOf, FileDirectoryWatcher } from '../../../src/data_sources/filedata'; +import TestLogger, { LogLevel } from '../../Logger'; +import MockFilesystem from './MockFilesystem'; + +it.each([ + ['/data/one.json', '/data'], + ['/one.json', '/'], + ['one.json', '.'], + ['relative/one.json', 'relative'], + ['C:\\one.json', 'C:\\'], + ['C:\\data\\one.json', 'C:\\data'], +])('determines the directory of %s', (path, expected) => { + expect(directoryOf(path)).toEqual(expected); +}); + +describe('given a directory watcher over a mock filesystem', () => { + let filesystem: MockFilesystem; + let logger: TestLogger; + let onChange: jest.Mock; + let watcher: FileDirectoryWatcher; + + beforeEach(() => { + jest.useFakeTimers(); + filesystem = new MockFilesystem(); + logger = new TestLogger(); + onChange = jest.fn(); + }); + + afterEach(() => { + watcher?.close(); + jest.useRealTimers(); + }); + + const startWatcher = (paths: string[]) => { + watcher = new FileDirectoryWatcher(filesystem, paths, onChange, logger); + watcher.start(); + return watcher; + }; + + // The watcher also watches each configured file directly. These tests are about the directory + // watches, so the file watches are left out of the counts. + const directoryWatches = () => + filesystem.activeWatches().filter((watch) => !watch.path.endsWith('.json')); + + it('watches the directory of each file once', () => { + startWatcher(['/a/one.json', '/a/two.json', '/b/three.json']); + + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a', '/b']); + expect(onChange).not.toHaveBeenCalled(); + }); + + it('watches each configured file directly and sets the watch up again after an event', () => { + // A direct watch follows a symbolic link to a file in another directory, which the directory + // watch cannot see. It ends when the file is replaced, so it is renewed after each event. + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [before] = filesystem.activeWatches('/a/one.json'); + expect(before).toBeDefined(); + + filesystem.emit('/a/one.json', 'change'); + expect(onChange).toHaveBeenCalledTimes(1); + const [after] = filesystem.activeWatches('/a/one.json'); + expect(before.closed).toBe(true); + expect(after).toBeDefined(); + expect(after).not.toBe(before); + + watcher.close(); + expect(filesystem.activeWatches('/a/one.json')).toHaveLength(0); + }); + + it('reloads and sets the watch up again when a direct file watch reports an error', () => { + // The file can have changed while its watch was failing, and when the path is a link to a + // file in another directory, the directory watch does not see that change. + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [before] = filesystem.activeWatches('/a/one.json'); + + filesystem.emit('/a/one.json', 'error'); + expect(onChange).toHaveBeenCalledTimes(1); + const [after] = filesystem.activeWatches('/a/one.json'); + expect(before.closed).toBe(true); + expect(after).toBeDefined(); + expect(after).not.toBe(before); + }); + + it('invokes the callback for a change in a watched directory', () => { + startWatcher(['/a/one.json']); + + filesystem.emit('/a', 'rename'); + filesystem.emit('/a', 'change'); + expect(onChange).toHaveBeenCalledTimes(2); + }); + + it('ignores an event that names an entry which is not a configured file', () => { + startWatcher(['/a/one.json']); + + filesystem.emit('/a', 'change', 'other.json'); + expect(onChange).not.toHaveBeenCalled(); + filesystem.emit('/a', 'change', 'one.json'); + expect(onChange).toHaveBeenCalledTimes(1); + // Without a name from the platform, every event counts. + filesystem.emit('/a', 'change'); + expect(onChange).toHaveBeenCalledTimes(2); + }); + + it('runs the callback for an event naming another entry when a configured file changed through it', async () => { + // A mounted ConfigMap is updated by swapping a symbolic link that the configured file + // points through. The event names the link, not the file, and the file's metadata changes. + filesystem.set('/a/one.json', '{}', 1, 2); + startWatcher(['/a/one.json']); + await jest.advanceTimersByTimeAsync(0); + + filesystem.set('/a/one.json', '{}', 2, 2); + filesystem.emit('/a', 'rename', '..data'); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(1); + + // The same metadata again is not a change. + filesystem.emit('/a', 'rename', '..data_tmp'); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(1); + }); + + it('renews the direct watch when a directory event names the configured file', () => { + // A link that is retargeted in place produces an event in its directory that names it. The + // direct watch is still on the old target, so it is set up again on the new one. + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [before] = filesystem.activeWatches('/a/one.json'); + + filesystem.emit('/a', 'rename', 'one.json'); + expect(onChange).toHaveBeenCalledTimes(1); + const [after] = filesystem.activeWatches('/a/one.json'); + expect(before.closed).toBe(true); + expect(after).toBeDefined(); + expect(after).not.toBe(before); + }); + + it('renews the direct watch when a configured file changed through another entry', async () => { + filesystem.set('/a/one.json', '{}', 1, 2); + startWatcher(['/a/one.json']); + await jest.advanceTimersByTimeAsync(0); + const [before] = filesystem.activeWatches('/a/one.json'); + + filesystem.set('/a/one.json', '{}', 2, 2); + filesystem.emit('/a', 'rename', '..data'); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(1); + const [after] = filesystem.activeWatches('/a/one.json'); + expect(before.closed).toBe(true); + expect(after).toBeDefined(); + expect(after).not.toBe(before); + }); + + it('notifies for a change that a check in flight observed when the request arrived', async () => { + // An event naming the file starts a metadata read that does not notify, because the event + // did. A link swap during that read produces an event for another entry, whose request to + // notify must not be lost when the read in flight already recorded the new metadata. + filesystem.set('/a/one.json', '{}', 1, 2); + startWatcher(['/a/one.json']); + await jest.advanceTimersByTimeAsync(0); + + filesystem.deferStats = true; + filesystem.emit('/a', 'change', 'one.json'); + expect(onChange).toHaveBeenCalledTimes(1); + filesystem.set('/a/one.json', '{}', 2, 2); + filesystem.emit('/a', 'rename', '..data'); + filesystem.completePendingStats(); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(2); + + // The read that was requested during the first one finds nothing new. + filesystem.completePendingStats(); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(2); + }); + + it('treats an event for another entry during the first metadata read as a change', async () => { + // The first read has nothing to compare to, so a link swap that happens while it is in + // flight cannot be told from the baseline. The event for the swap must still reload. + filesystem.set('/a/one.json', '{}', 1, 2); + filesystem.deferStats = true; + startWatcher(['/a/one.json']); + + filesystem.set('/a/one.json', '{}', 2, 2); + filesystem.emit('/a', 'rename', '..data'); + filesystem.completePendingStats(); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(1); + + filesystem.completePendingStats(); + await jest.advanceTimersByTimeAsync(0); + expect(onChange).toHaveBeenCalledTimes(1); + }); + + it('reads the metadata of every configured file in the directory, not only the first', () => { + startWatcher(['/a/one.json', '/a/two.json']); + + filesystem.emit('/a', 'change', 'two.json'); + expect(onChange).toHaveBeenCalledTimes(1); + }); + + it('still checks the directory on an event for an entry that is not configured', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [watch] = filesystem.activeWatches('/a'); + + filesystem.removeDirectory('/a'); + filesystem.emit('/a', 'rename', 'other.json'); + expect(onChange).not.toHaveBeenCalled(); + await jest.advanceTimersByTimeAsync(0); + + expect(watch.closed).toBe(true); + }); + + it('closes the watches and ignores later events', () => { + startWatcher(['/a/one.json', '/b/two.json']); + watcher.close(); + + expect(directoryWatches()).toHaveLength(0); + filesystem.watches.forEach((watch) => watch.callback('change', watch.path)); + expect(onChange).not.toHaveBeenCalled(); + }); + + it('signals a directory that appears while another directory stays unwatchable', async () => { + // The catch-up after a watch gap must not wait for every directory: a configured path in a + // directory that never exists is a valid steady state, and a file written into the + // recovered directory before its watch was in place would otherwise never be loaded. + filesystem.failingDirectories.add('/a'); + filesystem.failingDirectories.add('/b'); + startWatcher(['/a/one.json', '/b/two.json']); + expect(directoryWatches()).toHaveLength(0); + + filesystem.failingDirectories.delete('/a'); + filesystem.set('/a/one.json', '{"flagValues":{"from-gap":true}}'); + await jest.advanceTimersByTimeAsync(1000); + + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a']); + expect(onChange).toHaveBeenCalledTimes(1); + + // The retry continues for the directory that is still missing, and signals once for it too. + filesystem.failingDirectories.delete('/b'); + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a', '/b']); + expect(onChange).toHaveBeenCalledTimes(2); + }); + + it('signals a directory that is deleted and recreated while another directory stays unwatchable', async () => { + filesystem.failingDirectories.add('/b'); + startWatcher(['/a/one.json', '/b/two.json']); + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a']); + + // The directory goes, which an event reports, and comes back with a new file before the + // retry sets its watch up again. + filesystem.removeDirectory('/a'); + filesystem.emit('/a', 'rename', 'one.json'); + await jest.advanceTimersByTimeAsync(0); + expect(directoryWatches()).toHaveLength(0); + const callsAfterLoss = onChange.mock.calls.length; + filesystem.restoreDirectory('/a'); + filesystem.set('/a/one.json', '{"flagValues":{"from-gap":true}}'); + await jest.advanceTimersByTimeAsync(1000); + + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a']); + expect(onChange).toHaveBeenCalledTimes(callsAfterLoss + 1); + }); + + it('logs a directory that cannot be watched and retries until it can', async () => { + filesystem.failingDirectories.add('/b'); + startWatcher(['/a/one.json', '/b/two.json']); + + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a']); + logger.expectMessages([{ level: LogLevel.Error, matches: /Unable to watch directory "\/b"/ }]); + expect(onChange).not.toHaveBeenCalled(); + + // The retry fails again while the directory stays unavailable. + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a']); + expect(logger.getCount(LogLevel.Error)).toEqual(2); + + // When the retry succeeds, the callback runs once to pick up changes made in the meantime. + filesystem.failingDirectories.delete('/b'); + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches().map((watch) => watch.path)).toEqual(['/a', '/b']); + expect(onChange).toHaveBeenCalledTimes(1); + + await jest.advanceTimersByTimeAsync(5000); + expect(onChange).toHaveBeenCalledTimes(1); + }); + + it('stops retrying when closed', async () => { + filesystem.failingDirectories.add('/b'); + startWatcher(['/b/two.json']); + watcher.close(); + + filesystem.failingDirectories.delete('/b'); + await jest.advanceTimersByTimeAsync(5000); + expect(directoryWatches()).toHaveLength(0); + expect(onChange).not.toHaveBeenCalled(); + }); + + it('keeps the watch on a directory that still exists', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [watch] = filesystem.activeWatches('/a'); + + filesystem.emit('/a', 'change'); + watcher.verify(); + await jest.advanceTimersByTimeAsync(0); + + expect(filesystem.activeWatches('/a')).toEqual([watch]); + expect(onChange).toHaveBeenCalledTimes(1); + expect(logger.getCount(LogLevel.Warn)).toEqual(0); + }); + + it('closes the watch on a deleted directory so that a flood of events stops', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [watch] = filesystem.activeWatches('/a'); + + // Some platforms deliver an endless stream of events for a deleted directory. The first + // ones arrive before the existence check completes. + filesystem.removeDirectory('/a'); + filesystem.emit('/a', 'rename'); + filesystem.emit('/a', 'rename'); + filesystem.emit('/a', 'rename'); + expect(onChange).toHaveBeenCalledTimes(3); + await jest.advanceTimersByTimeAsync(0); + + // The check finds the directory gone and closes the watch. Events the platform still + // delivers for it no longer reach the callback. + expect(watch.closed).toBe(true); + expect(directoryWatches()).toHaveLength(0); + logger.expectMessages([ + { level: LogLevel.Warn, matches: /Directory "\/a" no longer exists/ }, + { level: LogLevel.Error, matches: /Unable to watch directory "\/a"/ }, + ]); + Array.from({ length: 100 }).forEach(() => watch.callback('rename', '/a')); + expect(onChange).toHaveBeenCalledTimes(3); + + // The watch is set up again once the directory exists, and the callback runs to pick up + // changes made in the meantime. + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches()).toHaveLength(0); + filesystem.restoreDirectory('/a'); + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches().map((active) => active.path)).toEqual(['/a']); + expect(onChange).toHaveBeenCalledTimes(4); + }); + + it('checks again when an event arrives during a check', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [watch] = filesystem.activeWatches('/a'); + + // The first event starts a check. The second event arrives while it is in progress. + filesystem.deferStats = true; + filesystem.emit('/a', 'rename'); + filesystem.emit('/a', 'rename'); + // The directory check is pending once. The files' metadata read is separate. + const directoryChecks = () => filesystem.pendingStats.filter((lookup) => lookup.path === '/a'); + expect(directoryChecks()).toHaveLength(1); + + // The first check observes the directory before it is deleted, so the watch stays. + filesystem.completePendingStats(); + await jest.advanceTimersByTimeAsync(0); + expect(filesystem.activeWatches('/a')).toEqual([watch]); + + // The second event is not lost: a check runs again and observes the deletion. + expect(directoryChecks()).toHaveLength(1); + filesystem.removeDirectory('/a'); + filesystem.completePendingStats(); + await jest.advanceTimersByTimeAsync(0); + expect(watch.closed).toBe(true); + expect(directoryWatches()).toHaveLength(0); + expect(filesystem.pendingStats).toHaveLength(0); + }); + + it('checks the directories on request and replaces the watch on a deleted one', async () => { + filesystem.set('/a/one.json', '{}'); + filesystem.set('/b/two.json', '{}'); + startWatcher(['/a/one.json', '/b/two.json']); + const [watchA] = filesystem.activeWatches('/a'); + const [watchB] = filesystem.activeWatches('/b'); + + // Some platforms deliver no event at all for a deleted directory. The owner asks for a check + // after a failed load. + filesystem.removeDirectory('/a'); + watcher.verify(); + await jest.advanceTimersByTimeAsync(0); + + expect(watchA.closed).toBe(true); + expect(directoryWatches()).toEqual([watchB]); + expect(onChange).not.toHaveBeenCalled(); + + filesystem.restoreDirectory('/a'); + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches().map((active) => active.path)).toEqual(['/b', '/a']); + expect(onChange).toHaveBeenCalledTimes(1); + }); + + it('replaces a watch that reports an error', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + const [watch] = filesystem.activeWatches('/a'); + + expect(() => filesystem.emit('/a', 'error')).not.toThrow(); + + // The failed watch is closed and a new one is in place, so a change made in the meantime + // is picked up. + expect(watch.closed).toBe(true); + expect(directoryWatches().map((active) => active.path)).toEqual(['/a']); + expect(filesystem.activeWatches('/a')[0]).not.toBe(watch); + expect(onChange).toHaveBeenCalledTimes(1); + logger.expectMessages([ + { level: LogLevel.Warn, matches: /The watch on directory "\/a" failed/ }, + ]); + }); + + it('retries a watch that reports an error while its directory is gone', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + + filesystem.removeDirectory('/a'); + filesystem.emit('/a', 'error'); + expect(directoryWatches()).toHaveLength(0); + expect(onChange).not.toHaveBeenCalled(); + + filesystem.restoreDirectory('/a'); + await jest.advanceTimersByTimeAsync(1000); + expect(directoryWatches().map((active) => active.path)).toEqual(['/a']); + expect(onChange).toHaveBeenCalledTimes(1); + }); +}); + +// A platform watch on a file that does not exist fails. The shared mock accepts a watch on any +// path, so this variant refuses a watch on a file path that has no file, as the platforms do. +class StrictMockFilesystem extends MockFilesystem { + override watch( + path: string, + callback: (eventType: string, filename: string, changedName?: string) => void, + ) { + if (path.endsWith('.json') && !this.files[path]) { + const err = new Error(`ENOENT: no such file or directory, watch '${path}'`) as Error & { + code: string; + }; + err.code = 'ENOENT'; + throw err; + } + return super.watch(path, callback); + } +} + +describe('given a directory watcher over a filesystem that refuses a watch on an absent file', () => { + let filesystem: StrictMockFilesystem; + let onChange: jest.Mock; + let watcher: FileDirectoryWatcher; + + beforeEach(() => { + jest.useFakeTimers(); + filesystem = new StrictMockFilesystem(); + onChange = jest.fn(); + }); + + afterEach(() => { + watcher?.close(); + jest.useRealTimers(); + }); + + const startWatcher = (paths: string[]) => { + watcher = new FileDirectoryWatcher(filesystem, paths, onChange, new TestLogger()); + watcher.start(); + return watcher; + }; + + it('sets the direct file watch up again when a deleted directory is restored with its file', async () => { + filesystem.set('/a/one.json', '{}'); + startWatcher(['/a/one.json']); + expect(filesystem.activeWatches('/a/one.json')).toHaveLength(1); + + // The deletion reaches the direct watch, which ends with the file, and the directory watch, + // whose check finds the directory gone and closes it. + filesystem.removeDirectory('/a'); + filesystem.emit('/a/one.json', 'rename'); + filesystem.emit('/a', 'rename', 'one.json'); + await jest.advanceTimersByTimeAsync(0); + expect(filesystem.activeWatches('/a')).toHaveLength(0); + expect(filesystem.activeWatches('/a/one.json')).toHaveLength(0); + + // The directory comes back with the file in place. The retry restores the directory watch, + // and the direct watch on the file comes back with it. + filesystem.set('/a/one.json', '{}'); + await jest.advanceTimersByTimeAsync(1000); + expect(filesystem.activeWatches('/a')).toHaveLength(1); + expect(filesystem.activeWatches('/a/one.json')).toHaveLength(1); + }); + + it('sets the direct file watch up when a directory missing at start appears with its file', async () => { + filesystem.removeDirectory('/a'); + startWatcher(['/a/one.json']); + expect(filesystem.activeWatches('/a')).toHaveLength(0); + expect(filesystem.activeWatches('/a/one.json')).toHaveLength(0); + + // The directory appears with the file in place. The retry sets up the directory watch and + // the direct watch on the file together. + filesystem.set('/a/one.json', '{}'); + await jest.advanceTimersByTimeAsync(1000); + expect(filesystem.activeWatches('/a')).toHaveLength(1); + expect(filesystem.activeWatches('/a/one.json')).toHaveLength(1); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/filedata/FileReloader.test.ts b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileReloader.test.ts new file mode 100644 index 0000000000..ace23c0667 --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileReloader.test.ts @@ -0,0 +1,695 @@ +import { + FileReloader, + FileReloaderConfig, + LoadFailure, + ReloadResult, +} from '../../../src/data_sources/filedata'; +import TestLogger, { LogLevel } from '../../Logger'; +import MockFilesystem from './MockFilesystem'; +import testPolicy from './testPolicy'; + +const path = '/data/data.json'; +const second = '/data/second.json'; + +// A reload starts on a microtask after it is requested. Let it begin before asserting. +const flushMicrotasks = () => jest.advanceTimersByTimeAsync(0); + +async function repeat(count: number, action: () => Promise): Promise { + await Array.from({ length: count }).reduce>( + (previous) => previous.then(action), + Promise.resolve(), + ); +} + +describe('given a reloader over a mock filesystem', () => { + let filesystem: MockFilesystem; + let logger: TestLogger; + let applied: ReloadResult[]; + let failures: LoadFailure[]; + let reloader: FileReloader; + + beforeEach(() => { + jest.useFakeTimers(); + filesystem = new MockFilesystem(); + logger = new TestLogger(); + applied = []; + failures = []; + }); + + afterEach(() => { + reloader?.close(); + jest.useRealTimers(); + }); + + const makeReloader = (configure?: (config: FileReloaderConfig) => void) => { + const config: FileReloaderConfig = { + paths: [path], + policy: testPolicy({ missingFile: 'fail' }), + filesystem, + logger, + apply: (result) => applied.push(result), + onFailure: (failure) => failures.push(failure), + debounceDelayMs: 0, + retryDelayMs: 0, + skipUnchanged: false, + }; + configure?.(config); + reloader = new FileReloader(config); + return reloader; + }; + + const flagKeys = (result: ReloadResult) => result.flags.map((flag) => flag.key); + + it('applies the initial load', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader(); + + await reloader.reloadNow(); + + expect(applied).toHaveLength(1); + expect(flagKeys(applied[0])).toEqual(['flag1']); + expect(applied[0].files).toEqual([{ path, present: true, flags: 1, segments: 0 }]); + expect(failures).toHaveLength(0); + }); + + it('fails on a missing file when the policy says so, with the original error', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.paths = [path, second]; + }); + + await reloader.reloadNow(); + + expect(applied).toHaveLength(0); + expect(failures).toHaveLength(1); + expect(failures[0].kind).toEqual('read'); + expect(failures[0].path).toEqual(second); + expect(failures[0].error.message).toMatch(/ENOENT/); + expect(failures[0].repeated).toBe(false); + }); + + it('skips a missing file when the policy says so', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.paths = [path, second]; + config.policy = testPolicy({ missingFile: 'skip' }); + config.skipUnchanged = true; + }); + + // Step 1: one file exists and one does not. The reload succeeds with the existing file. + await reloader.reloadNow(); + expect(applied).toHaveLength(1); + expect(flagKeys(applied[0])).toEqual(['flag1']); + expect(applied[0].files).toEqual([ + { path, present: true, flags: 1, segments: 0 }, + { path: second, present: false, flags: 0, segments: 0 }, + ]); + + // Step 2: the missing file appears. Its data is merged in. + filesystem.set(second, '{"flagValues": {"flag2": true}}'); + await reloader.reloadNow(); + expect(applied).toHaveLength(2); + expect(flagKeys(applied[1])).toEqual(['flag1', 'flag2']); + expect(applied[1].files[1]).toEqual({ path: second, present: true, flags: 1, segments: 0 }); + + // Step 3: the file is deleted. Its data is gone and the reload still succeeds. + filesystem.remove(second); + await reloader.reloadNow(); + expect(applied).toHaveLength(3); + expect(flagKeys(applied[2])).toEqual(['flag1']); + expect(failures).toHaveLength(0); + }); + + it('reports a read failure that is not a missing file even when missing files are skipped', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + jest.spyOn(filesystem, 'readFile').mockRejectedValue(new Error('EACCES: permission denied')); + makeReloader((config) => { + config.policy = testPolicy({ missingFile: 'skip' }); + }); + + await reloader.reloadNow(); + + expect(applied).toHaveLength(0); + expect(failures).toHaveLength(1); + expect(failures[0].kind).toEqual('read'); + expect(failures[0].path).toEqual(path); + expect(failures[0].error.message).toEqual('EACCES: permission denied'); + }); + + it('reports a parse failure with the parser error and applies nothing', async () => { + filesystem.set(path, '{"flagValues"'); + makeReloader(); + + await reloader.reloadNow(); + + expect(applied).toHaveLength(0); + expect(failures).toHaveLength(1); + expect(failures[0].kind).toEqual('parse'); + expect(failures[0].path).toEqual(path); + expect(failures[0].error.message).toMatch(/json/i); + }); + + it('reports a merge failure', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.set(second, '{"flagValues": {"flag1": false}}'); + makeReloader((config) => { + config.paths = [path, second]; + }); + + await reloader.reloadNow(); + + expect(applied).toHaveLength(0); + expect(failures).toHaveLength(1); + expect(failures[0].kind).toEqual('merge'); + expect(failures[0].path).toBeUndefined(); + expect(failures[0].error.message).toEqual('duplicate flag flag1'); + }); + + it('stops reading after the first failing file', async () => { + filesystem.set(second, '{"flagValues": {"flag2": true}}'); + const readFile = jest.spyOn(filesystem, 'readFile'); + makeReloader((config) => { + config.paths = [path, second]; + }); + + await reloader.reloadNow(); + + expect(failures).toHaveLength(1); + expect(readFile).toHaveBeenCalledTimes(1); + }); + + it('merges multiple files in the configured order', async () => { + filesystem.set(path, '{"flagValues": {"flag1": "first"}}'); + filesystem.set(second, '{"flagValues": {"flag1": "second"}}'); + makeReloader((config) => { + config.paths = [path, second]; + config.policy = testPolicy({ missingFile: 'fail', resolveDuplicateKey: () => 'keepFirst' }); + }); + + await reloader.reloadNow(); + + expect(applied).toHaveLength(1); + expect(applied[0].flags).toHaveLength(1); + expect(applied[0].flags[0].item.variations).toEqual(['first']); + expect(applied[0].files).toEqual([ + { path, present: true, flags: 1, segments: 0 }, + { path: second, present: true, flags: 0, segments: 0 }, + ]); + }); + + it('hands the policy the flags of the last applied load and keeps them across a failure', async () => { + const seen: Array = []; + filesystem.set(path, '{"flagValues": {"flag1": "a"}}'); + makeReloader((config) => { + config.policy = testPolicy({ + missingFile: 'fail', + makeFlagWithValue: (key, value, previous) => { + seen.push(previous?.variations[0]); + return testPolicy().makeFlagWithValue(key, value, previous); + }, + }); + }); + + await reloader.reloadNow(); + filesystem.set(path, '{"flagValues": {"flag1": "b"}}'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + // A failed reload does not disturb the remembered flags. + filesystem.set(path, '{"flagValues"'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + filesystem.set(path, '{"flagValues": {"flag1": "c"}}'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + + expect(seen).toEqual([undefined, 'a', 'b']); + expect(applied).toHaveLength(3); + expect(failures).toHaveLength(1); + }); + + it('reloads after a trigger and logs the reason', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader(); + await reloader.reloadNow(); + + filesystem.set(path, '{"flagValues": {"flag1": false}}'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + + expect(applied).toHaveLength(2); + expect(applied[1].flags[0].item.variations).toEqual([false]); + logger.expectMessages([ + { level: LogLevel.Info, matches: /Reloading flag data after detecting a change/ }, + ]); + }); + + it('coalesces a burst of triggers into one reload', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.debounceDelayMs = 400; + }); + await reloader.reloadNow(); + + filesystem.set(path, '{"flagValues": {"flag1": false}}'); + await repeat(20, async () => { + reloader.trigger(); + await jest.advanceTimersByTimeAsync(1); + }); + expect(applied).toHaveLength(1); + + await jest.advanceTimersByTimeAsync(400); + expect(applied).toHaveLength(2); + + await jest.advanceTimersByTimeAsync(1000); + expect(applied).toHaveLength(2); + }); + + it('extends the debounce window with each trigger', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.debounceDelayMs = 250; + }); + await reloader.reloadNow(); + + // Triggers arrive closer together than the window for three windows. No reload runs while + // they keep arriving. + await repeat(15, async () => { + reloader.trigger(); + await jest.advanceTimersByTimeAsync(50); + }); + expect(applied).toHaveLength(1); + + // One reload runs after the stream stops. + await jest.advanceTimersByTimeAsync(250); + expect(applied).toHaveLength(2); + await jest.advanceTimersByTimeAsync(500); + expect(applied).toHaveLength(2); + }); + + it('reloads no later than the maximum debounce delay under a stream of triggers', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.debounceDelayMs = 250; + }); + await reloader.reloadNow(); + + // Triggers keep arriving closer together than the window for three seconds. A stream that + // never settles cannot postpone the reload without end. + await repeat(21, async () => { + reloader.trigger(); + await jest.advanceTimersByTimeAsync(50); + }); + expect(applied).toHaveLength(2); + await repeat(40, async () => { + reloader.trigger(); + await jest.advanceTimersByTimeAsync(50); + }); + expect(applied).toHaveLength(4); + + // The reload after the stream stops still happens. + await jest.advanceTimersByTimeAsync(250); + expect(applied).toHaveLength(5); + }); + + it('reports an exception from apply as a failure and keeps reloading afterwards', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + let applyThrows = true; + makeReloader((config) => { + config.skipUnchanged = true; + config.apply = (result) => { + if (applyThrows) { + throw new Error('bad entry'); + } + applied.push(result); + }; + }); + const rejections: unknown[] = []; + const onRejection = (reason: unknown) => rejections.push(reason); + process.on('unhandledRejection', onRejection); + try { + await reloader.reloadNow(); + await jest.advanceTimersByTimeAsync(0); + expect(failures).toHaveLength(1); + expect(failures[0].kind).toEqual('apply'); + expect(failures[0].error.message).toEqual('bad entry'); + expect(applied).toHaveLength(0); + + // The rejected result is not the baseline: the same content applies once apply accepts it, + // and the chain still runs reloads. + applyThrows = false; + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(applied).toHaveLength(1); + expect(rejections).toHaveLength(0); + } finally { + process.off('unhandledRejection', onRejection); + } + }); + + it('keeps retrying when the failure callback itself throws', async () => { + filesystem.set(path, '{"flagValues"'); + makeReloader((config) => { + config.retryDelayMs = 10; + config.onFailure = (failure) => { + failures.push(failure); + throw new Error('handler failed'); + }; + }); + + await reloader.reloadNow(); + expect(failures).toHaveLength(1); + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + await jest.advanceTimersByTimeAsync(10); + expect(applied).toHaveLength(1); + logger.expectMessages([ + { level: LogLevel.Error, matches: /Error while reporting a flag data load failure/ }, + ]); + }); + + it('marks a repeated failure and re-arms after a success', async () => { + filesystem.set(path, '{"flagValues"'); + makeReloader((config) => { + config.retryDelayMs = 10; + }); + + await reloader.reloadNow(); + expect(failures).toHaveLength(1); + expect(failures[0].repeated).toBe(false); + + // The automatic retries keep failing identically. Each report is marked as a repeat. + await jest.advanceTimersByTimeAsync(35); + expect(failures.length).toBeGreaterThan(1); + expect(failures.slice(1).every((failure) => failure.repeated)).toBe(true); + + // A failure of a different kind is a new report. + filesystem.remove(path); + await jest.advanceTimersByTimeAsync(10); + const different = failures[failures.length - 1]; + expect(different.repeated).toBe(false); + expect(different.kind).toEqual('read'); + + // A success re-arms reporting. The same failure afterward is a new report. + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + await jest.advanceTimersByTimeAsync(10); + expect(applied).toHaveLength(1); + filesystem.set(path, '{"flagValues"'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(failures[failures.length - 1].repeated).toBe(false); + logger.expectMessages([ + { level: LogLevel.Debug, matches: /Retrying flag data load after earlier failure/ }, + ]); + }); + + it('reports a failure after a change notification as new even when it matches the last one', async () => { + filesystem.set(path, '{"flagValues"'); + makeReloader((config) => { + config.retryDelayMs = 10; + }); + + await reloader.reloadNow(); + await jest.advanceTimersByTimeAsync(25); + expect(failures.length).toBeGreaterThan(1); + expect(failures[failures.length - 1].repeated).toBe(true); + + // The file changed and still fails the same way. That is news, not a repeat. + filesystem.set(path, '{"flagValues": '); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(failures[failures.length - 1].repeated).toBe(false); + }); + + it('marks parse failures with different messages on the same path as repeats', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.set(second, '{"flagValues": {"flag2": true}}'); + makeReloader((config) => { + config.paths = [path, second]; + config.retryDelayMs = 10; + }); + await reloader.reloadNow(); + + // A file that is being written fails to parse at a different position on each attempt. + // That is one problem, reported once. + filesystem.set(path, '{"flagValues"'); + reloader.trigger(); + await flushMicrotasks(); + filesystem.set(path, '{"flagValues": {"flag1": tr'); + await jest.advanceTimersByTimeAsync(10); + filesystem.set(path, '{"flagValues": {"flag1": true}, "flags": {'); + await jest.advanceTimersByTimeAsync(10); + expect(failures).toHaveLength(3); + expect(new Set(failures.map((failure) => failure.error.message)).size).toEqual(3); + expect(failures.map((failure) => failure.repeated)).toEqual([false, true, true]); + + // The same kind of failure on another path is a new report. + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.set(second, '{"flagValues"'); + await jest.advanceTimersByTimeAsync(10); + expect(failures).toHaveLength(4); + expect(failures[3].repeated).toBe(false); + expect(failures[3].path).toEqual(second); + + // A failure of another kind on the same path is a new report. + filesystem.remove(second); + await jest.advanceTimersByTimeAsync(10); + expect(failures).toHaveLength(5); + expect(failures[4].repeated).toBe(false); + expect(failures[4].kind).toEqual('read'); + }); + + it('retries after a failure without further triggers', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.retryDelayMs = 20; + }); + await reloader.reloadNow(); + expect(applied).toHaveLength(1); + + filesystem.set(path, '{"flagValues"'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(failures).toHaveLength(1); + expect(applied).toHaveLength(1); + + // Fix the file without a trigger. Only the automatic retry can observe the fix. + filesystem.set(path, '{"flagValues": {"flag1": false}}'); + await jest.advanceTimersByTimeAsync(20); + expect(applied).toHaveLength(2); + expect(applied[1].flags[0].item.variations).toEqual([false]); + }); + + it('stops retrying after a success', async () => { + filesystem.set(path, '{"flagValues"'); + makeReloader((config) => { + config.retryDelayMs = 10; + config.skipUnchanged = true; + }); + await reloader.reloadNow(); + expect(failures).toHaveLength(1); + + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + await jest.advanceTimersByTimeAsync(10); + expect(applied).toHaveLength(1); + + // No reload happens without a trigger once the retry has succeeded. + filesystem.set(path, '{"flagValues": {"flag1": false}}'); + await jest.advanceTimersByTimeAsync(100); + expect(applied).toHaveLength(1); + }); + + it('does not retry when the retry delay is disabled', async () => { + filesystem.set(path, '{"flagValues"'); + makeReloader(); + await reloader.reloadNow(); + expect(failures).toHaveLength(1); + + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + await jest.advanceTimersByTimeAsync(5000); + expect(applied).toHaveLength(0); + expect(failures).toHaveLength(1); + }); + + it('skips an application when the content is unchanged', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.skipUnchanged = true; + }); + await reloader.reloadNow(); + expect(applied).toHaveLength(1); + + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(applied).toHaveLength(1); + + filesystem.set(path, '{"flagValues": {"flag1": false}}'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(applied).toHaveLength(2); + }); + + it('applies a recovery even when the content is unchanged', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.skipUnchanged = true; + }); + await reloader.reloadNow(); + expect(applied).toHaveLength(1); + + // A reload fails. The consumer hears about it through onFailure. + filesystem.remove(path); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(failures).toHaveLength(1); + + // The file comes back with identical content. Only apply tells the consumer that the + // interruption is over, so the success is applied. + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(applied).toHaveLength(2); + + // Once recovered, identical content skips again. + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(applied).toHaveLength(2); + }); + + it('applies every reload when skip unchanged is off', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader(); + await reloader.reloadNow(); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + expect(applied).toHaveLength(2); + }); + + it('does nothing after close', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + makeReloader((config) => { + config.retryDelayMs = 10; + }); + await reloader.reloadNow(); + expect(applied).toHaveLength(1); + + reloader.close(); + reloader.close(); + filesystem.set(path, '{"flagValues": {"flag1": false}}'); + reloader.trigger(); + await reloader.reloadNow(); + await jest.advanceTimersByTimeAsync(100); + expect(applied).toHaveLength(1); + expect(failures).toHaveLength(0); + }); + + it('does not deliver the result of a reload that completes after close', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.deferReads = true; + makeReloader(); + + const pending = reloader.reloadNow(); + await flushMicrotasks(); + expect(filesystem.pendingReads).toHaveLength(1); + + reloader.close(); + filesystem.completePendingReads(); + await pending; + + expect(applied).toHaveLength(0); + expect(failures).toHaveLength(0); + }); + + it('does not deliver a failure from a reload that completes after close', async () => { + filesystem.deferReads = true; + makeReloader((config) => { + config.retryDelayMs = 10; + }); + + const pending = reloader.reloadNow(); + await flushMicrotasks(); + expect(filesystem.pendingReads).toHaveLength(1); + reloader.close(); + filesystem.completePendingReads(); + await pending; + await jest.advanceTimersByTimeAsync(100); + + expect(failures).toHaveLength(0); + }); + + it('runs reloads one at a time', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.deferReads = true; + makeReloader(); + + const first = reloader.reloadNow(); + await flushMicrotasks(); + expect(filesystem.readCount).toEqual(1); + + // The second reload waits for the first to finish before it reads anything. + filesystem.deferReads = false; + const secondReload = reloader.reloadNow(); + await flushMicrotasks(); + expect(filesystem.readCount).toEqual(1); + expect(applied).toHaveLength(0); + + filesystem.completePendingReads(); + await first; + await secondReload; + expect(filesystem.readCount).toEqual(2); + expect(applied).toHaveLength(2); + }); + + it('coalesces triggers that arrive while a reload is in progress', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.deferReads = true; + makeReloader(); + + const first = reloader.reloadNow(); + await flushMicrotasks(); + expect(filesystem.readCount).toEqual(1); + reloader.trigger(); + reloader.trigger(); + reloader.trigger(); + await flushMicrotasks(); + expect(filesystem.readCount).toEqual(1); + + filesystem.deferReads = false; + filesystem.completePendingReads(); + await first; + await jest.advanceTimersByTimeAsync(0); + + // One reload for the initial load and one for the whole burst. + expect(filesystem.readCount).toEqual(2); + expect(applied).toHaveLength(2); + }); + + it('reads the files in the configured order on every reload', async () => { + filesystem.set(path, '{"flagValues": {"flag1": true}}'); + filesystem.set(second, '{"flagValues": {"flag2": true}}'); + const readFile = jest.spyOn(filesystem, 'readFile'); + makeReloader((config) => { + config.paths = [second, path]; + }); + + await reloader.reloadNow(); + reloader.trigger(); + await jest.advanceTimersByTimeAsync(0); + + expect(readFile.mock.calls.map((call) => call[0])).toEqual([second, path, second, path]); + expect(flagKeys(applied[1])).toEqual(['flag2', 'flag1']); + }); + + it('uses the policy parser', async () => { + const parseDocument = jest.fn(() => ({ flagValues: { flag1: true } })); + filesystem.set('/data/data.yaml', 'flagValues:\n flag1: true\n'); + makeReloader((config) => { + config.paths = ['/data/data.yaml']; + config.policy = testPolicy({ missingFile: 'fail', parseDocument }); + }); + + await reloader.reloadNow(); + + expect(parseDocument).toHaveBeenCalledWith('/data/data.yaml', 'flagValues:\n flag1: true\n'); + expect(applied).toHaveLength(1); + expect(flagKeys(applied[0])).toEqual(['flag1']); + }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/filedata/MockFilesystem.ts b/packages/shared/sdk-server/__tests__/data_sources/filedata/MockFilesystem.ts new file mode 100644 index 0000000000..3ff493315a --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/MockFilesystem.ts @@ -0,0 +1,198 @@ +import { FileStats, Filesystem, WatchHandle } from '@launchdarkly/js-sdk-common'; + +import { directoryOf } from '../../../src/data_sources/filedata'; + +interface MockFile { + data: string; + timestamp: number; + size?: number; +} + +export interface MockWatch { + path: string; + callback: (eventType: string, filename: string, changedName?: string) => void; + closed: boolean; +} + +interface PendingRead { + path: string; + resolve: (data: string) => void; + reject: (err: any) => void; +} + +interface PendingStats { + path: string; + resolve: (stats: FileStats | undefined) => void; +} + +function notFound(path: string): Error { + const err = new Error(`ENOENT: no such file or directory, open '${path}'`) as Error & { + code: string; + }; + err.code = 'ENOENT'; + return err; +} + +/** + * An in-memory filesystem for the file data tests. Reads and metadata lookups can be deferred + * so that a test can observe what happens while an operation is in progress. + */ +export default class MockFilesystem implements Filesystem { + public files: Record = {}; + + public watches: MockWatch[] = []; + + public failingDirectories = new Set(); + + // Directories that were removed. Watching one fails, and metadata lookups report it absent. A + // watch that already exists on it stays open, as the platform watchers do. + public removedDirectories = new Set(); + + public deferReads = false; + + public pendingReads: PendingRead[] = []; + + public deferStats = false; + + public pendingStats: PendingStats[] = []; + + public readCount = 0; + + public statsCount = 0; + + constructor(supportsStats: boolean = true) { + if (!supportsStats) { + // A platform without metadata support does not have the method at all. + (this as any).getFileStats = undefined; + } + } + + set(path: string, data: string, timestamp: number = 0, size?: number) { + this.files[path] = { data, timestamp, size }; + this.removedDirectories.delete(directoryOf(path)); + } + + remove(path: string) { + delete this.files[path]; + } + + /** + * Removes a directory and the files in it. Existing watches on it stay open until closed. + */ + removeDirectory(directory: string) { + Object.keys(this.files).forEach((path) => { + if (directoryOf(path) === directory) { + delete this.files[path]; + } + }); + this.removedDirectories.add(directory); + } + + /** + * Creates a removed directory again, empty. + */ + restoreDirectory(directory: string) { + this.removedDirectories.delete(directory); + } + + emit(path: string, eventType: string = 'change', changedName?: string) { + this.watches + .filter((watch) => watch.path === path && !watch.closed) + .forEach((watch) => watch.callback(eventType, path, changedName)); + } + + activeWatches(path?: string): MockWatch[] { + return this.watches.filter((watch) => !watch.closed && (!path || watch.path === path)); + } + + /** + * Completes every deferred read with the current file contents. + */ + completePendingReads() { + const pending = this.pendingReads; + this.pendingReads = []; + pending.forEach((read) => { + const file = this.files[read.path]; + if (file) { + read.resolve(file.data); + } else { + read.reject(notFound(read.path)); + } + }); + } + + /** + * Completes every deferred metadata lookup with the current file state. + */ + completePendingStats() { + const pending = this.pendingStats; + this.pendingStats = []; + pending.forEach((lookup) => lookup.resolve(this._statsOf(lookup.path))); + } + + async getFileTimestamp(path: string): Promise { + const file = this.files[path]; + if (!file) { + throw notFound(path); + } + return file.timestamp; + } + + async readFile(path: string): Promise { + this.readCount += 1; + if (this.deferReads) { + return new Promise((resolve, reject) => { + this.pendingReads.push({ path, resolve, reject }); + }); + } + const file = this.files[path]; + if (!file) { + throw notFound(path); + } + return file.data; + } + + async getFileStats(path: string): Promise { + this.statsCount += 1; + if (this.deferStats) { + return new Promise((resolve) => { + this.pendingStats.push({ path, resolve }); + }); + } + return this._statsOf(path); + } + + watch( + path: string, + callback: (eventType: string, filename: string, changedName?: string) => void, + ): WatchHandle { + if (this.failingDirectories.has(path)) { + throw new Error(`cannot watch ${path}`); + } + if (this.removedDirectories.has(path)) { + throw notFound(path); + } + const watch: MockWatch = { path, callback, closed: false }; + this.watches.push(watch); + return { + close: () => { + watch.closed = true; + }, + }; + } + + private _statsOf(path: string): FileStats | undefined { + const file = this.files[path]; + if (file) { + return { timestamp: file.timestamp, size: file.size ?? file.data.length }; + } + // A directory exists when it was not removed and a file is in it or it was watched. + if (this.removedDirectories.has(path)) { + return undefined; + } + const isDirectory = + Object.keys(this.files).some((filePath) => directoryOf(filePath) === path) || + this.activeWatches(path).length > 0; + return isDirectory ? { timestamp: 0, size: 0 } : undefined; + } +} diff --git a/packages/shared/sdk-server/__tests__/data_sources/filedata/document.test.ts b/packages/shared/sdk-server/__tests__/data_sources/filedata/document.test.ts new file mode 100644 index 0000000000..01031edf90 --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/document.test.ts @@ -0,0 +1,82 @@ +import { + isFileNotFoundError, + isYamlPath, + parseDocumentByExtension, +} from '../../../src/data_sources/filedata'; + +const flag1 = { + key: 'flag1', + on: true, + fallthrough: { variation: 0 }, + variations: [true], + version: 2, +}; +const segment1 = { key: 'segment1', included: ['user1'], version: 3 }; +const jsonDocument = JSON.stringify({ + flags: { flag1 }, + flagValues: { flag2: 'value' }, + segments: { segment1 }, +}); +const yamlDocument = 'flagValues:\n flag1: true\n'; + +describe('given the extension-based parser', () => { + it('parses a file without a YAML extension as JSON and does not consult the YAML parser', () => { + const yamlParser = jest.fn(); + const document = parseDocumentByExtension(yamlParser)('data.txt', jsonDocument); + expect(document.flags).toEqual({ flag1 }); + expect(document.flagValues).toEqual({ flag2: 'value' }); + expect(document.segments).toEqual({ segment1 }); + expect(yamlParser).not.toHaveBeenCalled(); + }); + + it('reports a JSON error for a file without a YAML extension whose content is YAML', () => { + const yamlParser = jest.fn(); + expect(() => parseDocumentByExtension(yamlParser)('data.json', yamlDocument)).toThrow(/json/i); + expect(yamlParser).not.toHaveBeenCalled(); + }); + + it.each(['data.yml', 'data.yaml'])( + 'parses %s with the YAML parser even when the content looks like JSON', + (path) => { + const yamlParser = jest.fn(() => ({ flagValues: { flag1: true } })); + const document = parseDocumentByExtension(yamlParser)(path, jsonDocument); + expect(yamlParser).toHaveBeenCalledWith(jsonDocument); + expect(document.flagValues).toEqual({ flag1: true }); + }, + ); + + it.each(['data.yml', 'data.yaml'])('reports a missing YAML parser for %s', (path) => { + expect(() => parseDocumentByExtension()(path, '')).toThrow( + `Attempted to parse yaml file (${path}) without parser.`, + ); + }); + + it('returns the parsed value as it is, without validation', () => { + const parse = parseDocumentByExtension(); + expect(parse('data.json', '{"flags": [], "flagValues": null, "segments": 0}')).toEqual({ + flags: [], + flagValues: null, + segments: 0, + }); + expect(parse('data.json', '{"flags": {"flag-a": {"version": 1}}}')).toEqual({ + flags: { 'flag-a': { version: 1 } }, + }); + expect(parse('data.json', 'null')).toBeNull(); + }); +}); + +it('recognizes YAML paths', () => { + expect(isYamlPath('a.yml')).toBe(true); + expect(isYamlPath('a.yaml')).toBe(true); + expect(isYamlPath('a.json')).toBe(false); + expect(isYamlPath('a.yaml.json')).toBe(false); +}); + +it('recognizes a file not found error by its code', () => { + const err = new Error('missing') as Error & { code: string }; + err.code = 'ENOENT'; + expect(isFileNotFoundError(err)).toBe(true); + expect(isFileNotFoundError(new Error('other'))).toBe(false); + expect(isFileNotFoundError(undefined)).toBe(false); + expect(isFileNotFoundError({ code: 'EACCES' })).toBe(false); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/filedata/merge.test.ts b/packages/shared/sdk-server/__tests__/data_sources/filedata/merge.test.ts new file mode 100644 index 0000000000..c64dd8e5a4 --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/merge.test.ts @@ -0,0 +1,141 @@ +import { mergeDocuments } from '../../../src/data_sources/filedata'; +import { Flag } from '../../../src/evaluation/data/Flag'; +import { Segment } from '../../../src/evaluation/data/Segment'; +import testPolicy from './testPolicy'; + +function flag(key: string, version: number): Flag { + return { key, version, on: false, fallthrough: { variation: 0 }, variations: [true] }; +} + +function segment(key: string, version: number): Segment { + return { key, version }; +} + +const keysOf = (items: { key: string }[]) => items.map((item) => item.key); + +it('combines the items of several documents in order', () => { + const result = mergeDocuments(testPolicy(), [ + { flags: { flag1: flag('flag1', 2) } }, + { flagValues: { flag2: true } }, + { segments: { segment1: segment('segment1', 4) } }, + ]); + + expect(keysOf(result.flags)).toEqual(['flag1', 'flag2']); + expect(result.flags[0].item.version).toEqual(2); + expect(result.flags[1].item).toEqual(testPolicy().makeFlagWithValue('flag2', true, undefined)); + expect(keysOf(result.segments)).toEqual(['segment1']); + expect(result.segments[0].item.version).toEqual(4); + expect(result.documents).toEqual([ + { flags: 1, segments: 0 }, + { flags: 1, segments: 0 }, + { flags: 0, segments: 1 }, + ]); +}); + +it('expands flag values with the policy and hands it the previous flag for the key', () => { + const makeFlagWithValue = jest.fn((key: string, value: any, previous?: Flag) => + flag(key, (previous?.version ?? 0) + 1), + ); + const previous = { flag1: flag('flag1', 5) }; + + const result = mergeDocuments( + testPolicy({ makeFlagWithValue }), + [{ flagValues: { flag1: 'a', flag2: 'b' } }], + previous, + ); + + expect(makeFlagWithValue).toHaveBeenCalledWith('flag1', 'a', previous.flag1); + expect(makeFlagWithValue).toHaveBeenCalledWith('flag2', 'b', undefined); + expect(result.flags.map((item) => item.item.version)).toEqual([6, 1]); +}); + +it('fails the merge when the policy throws for a duplicate key', () => { + expect(() => + mergeDocuments(testPolicy(), [ + { flags: { flag1: flag('flag1', 1) } }, + { flags: { flag1: flag('flag1', 2) } }, + ]), + ).toThrow('duplicate flag flag1'); +}); + +it('keeps the first entry when the policy resolves a duplicate key to keepFirst', () => { + const result = mergeDocuments(testPolicy({ resolveDuplicateKey: () => 'keepFirst' }), [ + { flags: { flag1: flag('flag1', 1) } }, + { flags: { flag1: flag('flag1', 2) } }, + ]); + expect(result.flags).toHaveLength(1); + expect(result.flags[0].item.version).toEqual(1); + expect(result.documents).toEqual([ + { flags: 1, segments: 0 }, + { flags: 0, segments: 0 }, + ]); +}); + +it('keeps the last entry in the first position when the policy resolves to keepLast', () => { + const result = mergeDocuments(testPolicy({ resolveDuplicateKey: () => 'keepLast' }), [ + { flags: { flag1: flag('flag1', 1), other: flag('other', 1) } }, + { flags: { flag1: flag('flag1', 2) } }, + ]); + expect(keysOf(result.flags)).toEqual(['flag1', 'other']); + expect(result.flags[0].item.version).toEqual(2); + // The kept entry counts for the document it came from. + expect(result.documents).toEqual([ + { flags: 1, segments: 0 }, + { flags: 1, segments: 0 }, + ]); +}); + +it('treats a full flag and a flag value with the same key as duplicates', () => { + const resolveDuplicateKey = jest.fn(() => 'keepLast' as const); + const result = mergeDocuments(testPolicy({ resolveDuplicateKey }), [ + { flags: { flag1: flag('flag1', 1) }, flagValues: { flag1: true } }, + ]); + expect(resolveDuplicateKey).toHaveBeenCalledWith('flag', 'flag1'); + expect(result.flags).toHaveLength(1); + expect(result.flags[0].item.variations).toEqual([true]); +}); + +it('resolves duplicate segment keys through the policy', () => { + const resolveDuplicateKey = jest.fn(() => 'keepFirst' as const); + mergeDocuments(testPolicy({ resolveDuplicateKey }), [ + { segments: { segment1: segment('segment1', 1) } }, + { segments: { segment1: segment('segment1', 2) } }, + ]); + expect(resolveDuplicateKey).toHaveBeenCalledWith('segment', 'segment1'); +}); + +it('does not treat a flag and a segment with the same key as duplicates', () => { + const result = mergeDocuments(testPolicy(), [ + { flags: { same: flag('same', 1) }, segments: { same: segment('same', 1) } }, + ]); + expect(result.flags).toHaveLength(1); + expect(result.segments).toHaveLength(1); +}); + +it('preserves document order', () => { + const keys = ['flag-a', 'flag-b', 'flag-c', 'flag-d', 'flag-e']; + const result = mergeDocuments( + testPolicy(), + keys.map((key) => ({ flags: { [key]: flag(key, 1) } })), + ); + expect(keysOf(result.flags)).toEqual(keys); +}); + +it('keeps the key an entry was listed under, separate from the entry', () => { + const result = mergeDocuments(testPolicy(), [ + { flags: { 'map-key': { version: 1 } as unknown as Flag } }, + ]); + expect(result.flags[0].key).toEqual('map-key'); + expect(result.flags[0].item).toEqual({ version: 1 }); +}); + +it('contributes nothing for members that are not objects', () => { + const result = mergeDocuments(testPolicy(), [ + { flags: [] as unknown as Record, flagValues: null as any, segments: 0 as any }, + ]); + expect(result).toEqual({ flags: [], segments: [], documents: [{ flags: 0, segments: 0 }] }); +}); + +it('produces an empty result for no documents', () => { + expect(mergeDocuments(testPolicy(), [])).toEqual({ flags: [], segments: [], documents: [] }); +}); diff --git a/packages/shared/sdk-server/__tests__/data_sources/filedata/testPolicy.ts b/packages/shared/sdk-server/__tests__/data_sources/filedata/testPolicy.ts new file mode 100644 index 0000000000..981f72472d --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/testPolicy.ts @@ -0,0 +1,27 @@ +import { FileDataPolicy, parseDocumentByExtension } from '../../../src/data_sources/filedata'; +import { Flag } from '../../../src/evaluation/data/Flag'; + +/** + * A policy for the engine tests: content-based parsing, a flag value expands into an off flag + * with version 1, a duplicate key fails the load, and a missing file contributes nothing. Tests + * override the parts they exercise. + */ +export default function testPolicy(overrides: Partial = {}): FileDataPolicy { + return { + parseDocument: parseDocumentByExtension(), + entryKey: (mapKey) => mapKey, + makeFlagWithValue: (key: string, value: any): Flag => ({ + key, + version: 1, + on: false, + offVariation: 0, + fallthrough: { variation: 0 }, + variations: [value], + }), + resolveDuplicateKey: (category, key) => { + throw new Error(`duplicate ${category} ${key}`); + }, + missingFile: 'skip', + ...overrides, + }; +} diff --git a/packages/shared/sdk-server/src/api/integrations/FileDataSourceOptions.ts b/packages/shared/sdk-server/src/api/integrations/FileDataSourceOptions.ts index 39ba3e2a5b..f3322d2858 100644 --- a/packages/shared/sdk-server/src/api/integrations/FileDataSourceOptions.ts +++ b/packages/shared/sdk-server/src/api/integrations/FileDataSourceOptions.ts @@ -11,8 +11,22 @@ export interface FileDataSourceOptions { /** * True if FileDataSource should reload flags whenever one of the data files is modified. - * This feature uses Node's `fs.watch()` API, so it is subject to - * the limitations described [here](https://nodejs.org/docs/latest/api/fs.html#fs_fs_watch_filename_options_listener). + * This feature uses Node's `fs.watch()` API on the directories that contain the files, so it + * is subject to the limitations described + * [here](https://nodejs.org/docs/latest/api/fs.html#fs_fs_watch_filename_options_listener). + * A file that is replaced by a rename, or deleted and created again, is detected. A load + * that fails, for example because a file was read while it was written, is retried after a + * short delay and the previously loaded data stays in effect until it succeeds. + * A change to another entry in a directory causes a reload only when the metadata of a + * configured file changed, read through any symbolic link, which is how a mounted ConfigMap + * or Secret is updated. + * Each configured file is also watched directly, so a path that is a symbolic link to a file + * in another directory is followed, and that watch is set up again after each change so it + * survives the file being replaced. + * A directory that cannot be watched, because it does not exist yet or because of its + * permissions, is logged at error level and watched once it can be, with an attempt every + * second. Its files are still read on every load; a file that cannot be read fails the load, + * which is reported through the error handler and retried. */ autoUpdate?: boolean; diff --git a/packages/shared/sdk-server/src/data_sources/FileDataSource.ts b/packages/shared/sdk-server/src/data_sources/FileDataSource.ts index 6c50672bae..e222b897e3 100644 --- a/packages/shared/sdk-server/src/data_sources/FileDataSource.ts +++ b/packages/shared/sdk-server/src/data_sources/FileDataSource.ts @@ -12,7 +12,17 @@ import { LDDataSourceUpdates } from '../api/subsystems'; import { Flag } from '../evaluation/data/Flag'; import { processFlag, processSegment } from '../store/serialization'; import VersionedDataKinds from '../store/VersionedDataKinds'; -import FileLoader from './FileLoader'; +import { + DEFAULT_DEBOUNCE_DELAY_MS, + DEFAULT_RETRY_DELAY_MS, + FileDataPolicy, + FileDirectoryWatcher, + FileReloader, + LoadFailure, + parseDocumentByExtension, + ReloadResult, + YamlParser, +} from './filedata'; export type FileDataSourceErrorHandler = (err: LDFileDataSourceError) => void; @@ -26,14 +36,44 @@ export function makeFlagWithValue(key: string, value: any, version: number): Fla }; } +/** + * How the file data source translates its documents into data. This is the one place where its + * rules differ from the other file-based sources. + * + * - The parser is chosen by file extension only. + * - A `flagValues` entry becomes a flag that is on and serves the value by fallthrough. Its + * version starts at 1 and increments only when the value changes between loads. + * - A flag or segment entry is keyed by its own `key` property, as the source has always stored + * it, so two entries with the same `key` property are duplicates whatever their map keys. + * - A key that appears more than once fails the load. + * - A configured file that does not exist fails the load. + * + * @internal + */ +export function fileDataSourcePolicy(yamlParser?: YamlParser): FileDataPolicy { + return { + parseDocument: parseDocumentByExtension(yamlParser), + makeFlagWithValue: (key, value, previous) => { + let version = previous ? previous.version : 1; + if (previous && JSON.stringify(value) !== JSON.stringify(previous.variations?.[0])) { + version += 1; + } + return makeFlagWithValue(key, value, version); + }, + resolveDuplicateKey: (_category, key) => { + throw new Error(`found duplicate key: "${key}"`); + }, + missingFile: 'fail', + entryKey: (_mapKey, entry) => String(entry.key), + }; +} + export default class FileDataSource implements subsystem.LDStreamProcessor { private _logger?: LDLogger; - private _yamlParser?: (data: string) => any; + private _reloader: FileReloader; - private _fileLoader: FileLoader; - - private _allData: LDFeatureStoreDataStorage = {}; + private _watcher?: FileDirectoryWatcher; /** * This is internal because we want instances to only be created with the @@ -47,82 +87,68 @@ export default class FileDataSource implements subsystem.LDStreamProcessor { private _initSuccessHandler: VoidFunction = () => {}, private readonly _errorHandler?: FileDataSourceErrorHandler, ) { - this._fileLoader = new FileLoader( - filesystem, - options.paths, - options.autoUpdate ?? false, - (results: { path: string; data: string }[]) => { - // Whenever changes are detected we re-process all of the data. - // The FileLoader will have handled debouncing for us. - try { - this._processFileData(results); - } catch (err) { - // If this was during start, then the initCallback will be present. - this._errorHandler?.(err as LDFileDataSourceError); - this._logger?.error(`Error processing files: ${err}`); - } - }, - ); - this._logger = options.logger; - this._yamlParser = options.yamlParser; + const autoUpdate = options.autoUpdate ?? false; + this._reloader = new FileReloader({ + paths: options.paths, + policy: fileDataSourcePolicy(options.yamlParser), + filesystem, + logger: options.logger, + apply: (result) => this._applyData(result), + onFailure: (failure) => this._handleFailure(failure), + // Debouncing and automatic retries only matter when something can trigger further + // reloads. A source configured without automatic updates loads exactly once. + debounceDelayMs: autoUpdate ? DEFAULT_DEBOUNCE_DELAY_MS : 0, + retryDelayMs: autoUpdate ? DEFAULT_RETRY_DELAY_MS : 0, + skipUnchanged: true, + }); + if (autoUpdate) { + // The directories that contain the files are watched, so a file replaced by a rename or + // deleted and recreated is detected, and a file that is missing is picked up when it appears. + this._watcher = new FileDirectoryWatcher( + filesystem, + options.paths, + () => this._reloader.trigger(), + options.logger, + ); + } } start(): void { - // Use an immediately invoked function expression to allow handling of the - // async loading without making start async itself. - (async () => { - try { - await this._fileLoader.loadAndWatch(); - } catch (err) { - // There was an issue loading/watching the files. - // Report back to the caller. - this._errorHandler?.(err as LDFileDataSourceError); - } - })(); + // The watch is in place before the first load, so a change made between the two is not + // missed. The load never rejects: every filesystem and parsing failure is reported through + // the error handler. + this._watcher?.start(); + this._reloader.reloadNow(); } stop(): void { - this._fileLoader.close(); + this._watcher?.close(); + this._reloader.close(); } close(): void { this.stop(); } - private _addItem(kind: DataKind, item: any) { - if (!this._allData[kind.namespace]) { - this._allData[kind.namespace] = {}; - } - if (this._allData[kind.namespace][item.key]) { - throw new Error(`found duplicate key: "${item.key}"`); - } else { - this._allData[kind.namespace][item.key] = item; - } - } - - private _processFileData(fileData: { path: string; data: string }[]) { - // Clear any existing data before re-populating it. - const oldData = this._allData; - this._allData = {}; - - // We let the parsers throw, and the caller can handle the rejection. - fileData.forEach((fd) => { - let parsed: any; - if (fd.path.endsWith('.yml') || fd.path.endsWith('.yaml')) { - if (this._yamlParser) { - parsed = this._yamlParser(fd.data); - } else { - throw new Error(`Attempted to parse yaml file (${fd.path}) without parser.`); - } - } else { - parsed = JSON.parse(fd.data); + private _applyData(result: ReloadResult): void { + const allData: LDFeatureStoreDataStorage = {}; + const addItem = (kind: DataKind, item: any) => { + if (!allData[kind.namespace]) { + allData[kind.namespace] = {}; } - - this._processParsedData(parsed, oldData); + allData[kind.namespace][item.key] = item; + }; + result.flags.forEach(({ item }) => { + processFlag(item); + addItem(VersionedDataKinds.Features, item); + }); + result.segments.forEach(({ item }) => { + processSegment(item); + addItem(VersionedDataKinds.Segments, item); }); - this._featureStore.init(this._allData, () => { + this._featureStore.init(allData, () => { // Call the init callback if present. // Then clear the callback so we cannot call it again. this._initSuccessHandler(); @@ -130,28 +156,20 @@ export default class FileDataSource implements subsystem.LDStreamProcessor { }); } - private _processParsedData(parsed: any, oldData: LDFeatureStoreDataStorage) { - Object.keys(parsed.flags || {}).forEach((key) => { - processFlag(parsed.flags[key]); - this._addItem(VersionedDataKinds.Features, parsed.flags[key]); - }); - Object.keys(parsed.flagValues || {}).forEach((key) => { - const previousInstance = oldData[VersionedDataKinds.Features.namespace]?.[key]; - let { version } = previousInstance ?? { version: 1 }; - // If the data is different, then we want to increment the version. - if ( - previousInstance && - JSON.stringify(parsed.flagValues[key]) !== JSON.stringify(previousInstance?.variations?.[0]) - ) { - version += 1; - } - const flag = makeFlagWithValue(key, parsed.flagValues[key], version); - processFlag(flag); - this._addItem(VersionedDataKinds.Features, flag); - }); - Object.keys(parsed.segments || {}).forEach((key) => { - processSegment(parsed.segments[key]); - this._addItem(VersionedDataKinds.Segments, parsed.segments[key]); - }); + private _handleFailure(failure: LoadFailure): void { + // A load that fails to read a file can mean that the file's directory is gone, and a watch on + // a deleted directory cannot report anything useful. Let the watcher check its directories. + this._watcher?.verify(); + const message = + failure.kind === 'read' + ? `Error loading files: ${failure.error}` + : `Error processing files: ${failure.error}`; + if (failure.repeated) { + // An automatic retry hit the same failure again. Report it once at error level. + this._logger?.debug(message); + return; + } + this._errorHandler?.(failure.error as LDFileDataSourceError); + this._logger?.error(message); } } diff --git a/packages/shared/sdk-server/src/data_sources/FileLoader.ts b/packages/shared/sdk-server/src/data_sources/FileLoader.ts deleted file mode 100644 index fe2168c43f..0000000000 --- a/packages/shared/sdk-server/src/data_sources/FileLoader.ts +++ /dev/null @@ -1,92 +0,0 @@ -import { Filesystem, WatchHandle } from '@launchdarkly/js-sdk-common'; - -/** - * A debounced file load/watcher for use with the {@link FileDataSource}. - * - * The file loader will load all files specified and keep the string representations in memory. - * Whenever a change is made to any of the files, then that file will be reloaded and the in - * memory version updated. - * - * Updates to many files, which occur within 10ms of each other, will be coalesced into - * a single callback. - * - * @internal - */ -export default class FileLoader { - private _watchers: WatchHandle[] = []; - - private _fileData: Record = {}; - - private _fileTimestamps: Record = {}; - - private _debounceHandle: any; - - constructor( - private readonly _filesystem: Filesystem, - private readonly _paths: string[], - private readonly _watch: boolean, - private readonly _callback: (results: { path: string; data: string }[]) => void, - ) {} - - /** - * Load all the files and start watching them if watching is enabled. - */ - async loadAndWatch() { - const promises = this._paths.map(async (path) => { - const data = await this._filesystem.readFile(path); - const timeStamp = await this._filesystem.getFileTimestamp(path); - return { data, path, timeStamp }; - }); - // This promise could be rejected, let the caller handle it. - const results = await Promise.all(promises); - results.forEach((res) => { - this._fileData[res.path] = res.data; - this._fileTimestamps[res.path] = res.timeStamp; - }); - this._callback(results); - - // If we are watching, then setup watchers and notify of any changes. - if (this._watch) { - this._paths.forEach((path) => { - const watcher = this._filesystem.watch(path, async (_: string, updatePath: string) => { - const timeStamp = await this._filesystem.getFileTimestamp(updatePath); - // The modification time is the same, so we are going to ignore this update. - // In some implementations watch might be triggered multiple times for a single update. - if (timeStamp === this._fileTimestamps[updatePath]) { - return; - } - this._fileTimestamps[updatePath] = timeStamp; - const data = await this._filesystem.readFile(updatePath); - this._fileData[updatePath] = data; - this._debounceCallback(); - }); - this._watchers.push(watcher); - }); - } - } - - close() { - this._watchers.forEach((watcher) => watcher.close()); - } - - private _debounceCallback() { - // If there is a handle, then we have already started the debounce process. - if (!this._debounceHandle) { - this._debounceHandle = setTimeout(() => { - this._debounceHandle = undefined; - this._callback( - Object.entries(this._fileData).reduce( - (acc: { path: string; data: string }[], [path, data]) => { - acc.push({ path, data }); - return acc; - }, - [], - ), - ); - }, 10); - // The 10ms delay above is arbitrary - we just don't want to have the number be zero, - // because in a case where multiple watch events are fired off one after another, - // we want the reload to happen only after all of the event handlers have executed. - } - } -} diff --git a/packages/shared/sdk-server/src/data_sources/fileDataInitilizerFDv2.ts b/packages/shared/sdk-server/src/data_sources/fileDataInitilizerFDv2.ts index 515c76f56b..84c0d320ad 100644 --- a/packages/shared/sdk-server/src/data_sources/fileDataInitilizerFDv2.ts +++ b/packages/shared/sdk-server/src/data_sources/fileDataInitilizerFDv2.ts @@ -12,8 +12,48 @@ import { FileSystemDataSourceConfiguration } from '../api'; import { Flag } from '../evaluation/data/Flag'; import { Segment } from '../evaluation/data/Segment'; import { processFlag, processSegment } from '../store/serialization'; +import { + FileDataPolicy, + FileReloader, + KeyedItem, + LoadFailure, + parseDocumentByExtension, + ReloadResult, + YamlParser, +} from './filedata'; import { makeFlagWithValue } from './FileDataSource'; -import FileLoader from './FileLoader'; + +/** + * How the FDv2 file data initializer translates its documents into data. This is the one place + * where its rules differ from the other file-based sources. + * + * - The parser is chosen by file extension only. + * - A `flagValues` entry becomes a flag that is on and serves the value by fallthrough, with + * version 1. The initializer runs once and has no previous load to compare against. + * - A flag or segment entry is keyed by its map key, as the initializer has always merged it. + * - The last definition of a key wins, across files and between the `flags` and `flagValues` + * members of one file. + * - A configured file that does not exist fails the load. + * + * @internal + */ +export function fileDataInitializerPolicy(yamlParser?: YamlParser): FileDataPolicy { + return { + parseDocument: parseDocumentByExtension(yamlParser), + makeFlagWithValue: (key, value) => makeFlagWithValue(key, value, 1), + resolveDuplicateKey: () => 'keepLast', + missingFile: 'fail', + entryKey: (mapKey) => mapKey, + }; +} + +function byKey(items: KeyedItem[]): { [key: string]: T } { + const result: { [key: string]: T } = {}; + items.forEach(({ key, item }) => { + result[key] = item; + }); + return result; +} /** * Loads flag/segment data from one or more files. Each file may contain `flags` @@ -36,7 +76,7 @@ export default class FileDataInitializerFDv2 implements subsystemCommon.DataSour private _logger: LDLogger | undefined; private _filesystem: Filesystem; private _yamlParser?: (data: string) => any; - private _fileLoader?: FileLoader; + private _reloader?: FileReloader; constructor(options: FileSystemDataSourceConfiguration, platform: Platform, logger?: LDLogger) { this._validateInputs(options, platform); @@ -86,92 +126,57 @@ export default class FileDataInitializerFDv2 implements subsystemCommon.DataSour const adaptor = internal.FDv1PayloadAdaptor(payloadProcessor); - this._fileLoader = new FileLoader( - this._filesystem, - this._paths, - false, // autoupdate is always false for initializer - (results: { path: string; data: string }[]) => { - try { - const parsedData = this._processFileData(results); - - payloadProcessor.addPayloadListener((payload) => { - // NOTE: file data initializer will never have a valid basis, so we always pass false - dataCallback(false, { initMetadata, payload }); - }); - - statusCallback(subsystemCommon.DataSourceState.Valid); - - adaptor.processFullTransfer(parsedData); - - statusCallback(subsystemCommon.DataSourceState.Closed); - } catch (err) { - this._logger?.error('File contained invalid data', err); - statusCallback( - subsystemCommon.DataSourceState.Closed, - new LDPollingError(DataSourceErrorKind.InvalidData, 'Malformed data in file response'), - ); - } - }, - ); + const apply = (result: ReloadResult) => { + payloadProcessor.addPayloadListener((payload) => { + // NOTE: file data initializer will never have a valid basis, so we always pass false + dataCallback(false, { initMetadata, payload }); + }); + + statusCallback(subsystemCommon.DataSourceState.Valid); - this._fileLoader.loadAndWatch().catch((err) => { - this._logger?.error('Error loading files', err); + adaptor.processFullTransfer({ + segments: byKey(result.segments), + flags: byKey(result.flags), + }); + + statusCallback(subsystemCommon.DataSourceState.Closed); + }; + + const onFailure = (failure: LoadFailure) => { + if (failure.kind === 'read') { + this._logger?.error('Error loading files', failure.error); + statusCallback( + subsystemCommon.DataSourceState.Closed, + new LDPollingError( + DataSourceErrorKind.NetworkError, + `Failed to load files: ${failure.error.message}`, + ), + ); + return; + } + this._logger?.error('File contained invalid data', failure.error); statusCallback( subsystemCommon.DataSourceState.Closed, - new LDPollingError( - DataSourceErrorKind.NetworkError, - `Failed to load files: ${err instanceof Error ? err.message : String(err)}`, - ), + new LDPollingError(DataSourceErrorKind.InvalidData, 'Malformed data in file response'), ); + }; + + // The initializer loads once. It does not watch, debounce, or retry. + this._reloader = new FileReloader({ + paths: this._paths, + policy: fileDataInitializerPolicy(this._yamlParser), + filesystem: this._filesystem, + logger: this._logger, + apply, + onFailure, + debounceDelayMs: 0, + retryDelayMs: 0, + skipUnchanged: false, }); - } - - private _processFileData(results: { path: string; data: string }[]) { - const combined: any = results.reduce( - (acc, curr) => { - let parsed: any; - if (curr.path.endsWith('.yml') || curr.path.endsWith('.yaml')) { - if (this._yamlParser) { - parsed = this._yamlParser(curr.data); - } else { - throw new Error(`Attempted to parse yaml file (${curr.path}) without parser.`); - } - } else { - parsed = JSON.parse(curr.data); - } - - // flagValues has no previous-state to diff against, so each entry always - // gets version 1. Convert to full Flag objects here so they merge with - // flags below on equal footing - const flagsFromValues: { [key: string]: Flag } = {}; - Object.entries(parsed.flagValues ?? {}).forEach(([key, value]) => { - flagsFromValues[key] = makeFlagWithValue(key, value, 1); - }); - - return { - segments: { - ...acc.segments, - ...(parsed.segments ?? {}), - }, - flags: { - ...acc.flags, - ...(parsed.flags ?? {}), - ...flagsFromValues, - }, - }; - }, - { - segments: {}, - flags: {}, - }, - ); - - return combined; + this._reloader.reloadNow(); } stop() { - if (this._fileLoader) { - this._fileLoader.close(); - } + this._reloader?.close(); } } diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileDataPolicy.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileDataPolicy.ts new file mode 100644 index 0000000000..5d228c7743 --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileDataPolicy.ts @@ -0,0 +1,70 @@ +import { Flag } from '../../evaluation/data/Flag'; +import { Segment } from '../../evaluation/data/Segment'; +import { FileDataDocument } from './document'; + +/** + * The kind of item a key identifies. + * + * @internal + */ +export type ItemCategory = 'flag' | 'segment'; + +/** + * Which entry to keep when a key appears more than once. + * + * @internal + */ +export type DuplicateKeyResolution = 'keepFirst' | 'keepLast'; + +/** + * What a configured file that does not exist means. + * + * - `fail`: the load fails and is reported like any other read failure. + * - `skip`: the file contributes no data and the load succeeds with the other files. + * + * @internal + */ +export type MissingFileHandling = 'fail' | 'skip'; + +/** + * How a source translates file documents into flag and segment data. + * + * The shared file data engine reads, reloads, and merges files the same way for every source. + * Each source supplies one policy that decides how its documents become data, so the differences + * between sources are visible in one place per source. + * + * @internal + */ +export interface FileDataPolicy { + /** + * Parses the text of one file into a document. Throws when the file cannot be parsed. + */ + parseDocument(path: string, data: string): FileDataDocument; + + /** + * Expands a `flagValues` entry into a full flag definition. `previous` is the flag that the last + * successful load produced for the same key, if any, so that a policy can carry a version + * forward. + */ + makeFlagWithValue(key: string, value: any, previous: Flag | undefined): Flag; + + /** + * Decides what happens when a key appears more than once: across the configured files, or + * between the `flags` and `flagValues` members of one file. Returns which entry to keep, or + * throws to fail the load. + */ + resolveDuplicateKey(category: ItemCategory, key: string): DuplicateKeyResolution; + + /** + * What a configured file that does not exist means. + */ + missingFile: MissingFileHandling; + + /** + * Returns the key under which a `flags` or `segments` entry is stored and checked for + * duplicates. `mapKey` is the property name of the entry in the document, and `entry` is its + * definition. The file data source keys an entry by its own `key` property, the FDv2 + * initializer and the override source by the map key. + */ + entryKey(mapKey: string, entry: Flag | Segment): string; +} diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts new file mode 100644 index 0000000000..4e9b5fdebc --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts @@ -0,0 +1,426 @@ +import { Filesystem, LDLogger, WatchHandle } from '@launchdarkly/js-sdk-common'; + +/** + * The delay before a failed watch setup is attempted again. + */ +const WATCH_RETRY_DELAY_MS = 1000; + +/** + * Returns the directory part of a path. The shared code has no platform path module, so this + * handles both separators. + */ +export function directoryOf(path: string): string { + const index = Math.max(path.lastIndexOf('/'), path.lastIndexOf('\\')); + if (index < 0) { + return '.'; + } + if (index === 0) { + return path.substring(0, 1); + } + const directory = path.substring(0, index); + // A Windows drive root keeps its separator. + return /^[a-zA-Z]:$/.test(directory) ? `${directory}\\` : directory; +} + +/** + * Returns the file name part of a path, handling both separators. + */ +export function basenameOf(path: string): string { + const index = Math.max(path.lastIndexOf('/'), path.lastIndexOf('\\')); + return index < 0 ? path : path.substring(index + 1); +} + +async function directoryExists(filesystem: Filesystem, directory: string): Promise { + try { + if (filesystem.getFileStats) { + return (await filesystem.getFileStats(directory)) !== undefined; + } + await filesystem.getFileTimestamp(directory); + return true; + } catch { + return false; + } +} + +/** + * One watch on a directory. The flag marks a watch that the owner has closed, so that events the + * platform still delivers for it are ignored. + */ +class DirectoryWatch { + public closed = false; + + public readonly handle: WatchHandle; + + constructor( + filesystem: Filesystem, + directory: string, + onEvent: (watch: DirectoryWatch, eventType: string, changedName?: string) => void, + ) { + this.handle = filesystem.watch(directory, (eventType, _watched, changedName) => + onEvent(this, eventType, changedName), + ); + } +} + +/** + * Watches the directories that contain the given files, and invokes the change callback on + * any change in them. Watching the directory rather than the file means that a file which + * does not exist yet is detected when it appears, and that a file replaced by a rename does + * not lose its watch. + * + * An event that names an entry which is not one of the configured files runs the callback only + * when the metadata of a configured file in that directory changed. The metadata is read through + * any symbolic link, so a link swap that replaces a configured file's target without naming the + * link, which is how a mounted ConfigMap is updated, is detected, while a busy sibling file costs + * no reload. When the platform does not report which entry changed, every event in the directory + * runs the callback. Feed it into a {@link FileReloader}, whose debouncing and skip-unchanged + * handling absorb the excess. + * + * Each configured file is also watched directly. A watch on a file follows a symbolic link to its + * target, so a configured path that links to a file in another directory is detected when that + * file changes, as a watch on the path alone always did. A watch on a file ends when the file is + * replaced, so after each event the watch is set up again, on whatever the path names now. + * + * A directory that cannot be watched is logged and attempted again after a delay. A watch on a + * directory that is deleted goes dead on some platforms and floods events on others, so after + * each event, after a watch error, and on request after a failed load, the watcher checks that + * the directory still exists. A directory that no longer exists has its watch closed and set up + * again once it exists, through the same retry. When a retry succeeds, the callback runs once so + * that changes made in the meantime are picked up. + * + * @internal + */ +export default class FileDirectoryWatcher { + private readonly _directories: string[]; + + private readonly _watches: Record = {}; + + // The configured file names in each directory. An event for one of them always counts. + private readonly _names: Record> = {}; + + // The configured files in each directory, and the last observed metadata of each, read through + // any symbolic link. An event for another entry counts when this metadata changed. + private readonly _files: Record = {}; + + private readonly _signatures: Record = {}; + + // The directories whose files' metadata is being read. The flag records that a check was + // requested during the read, so that it runs again afterwards. + private readonly _fileChecks: Record = {}; + + // The configured paths, and the direct watch on each that currently exists. + private readonly _paths: string[]; + + private readonly _fileWatches: Record = {}; + + // The directories whose existence is being checked. The flag records that an event arrived + // during the check, so that the check runs again and observes the state after that event. + private readonly _checks: Record = {}; + + private _retryTimer?: ReturnType; + + private _closed = false; + + constructor( + private readonly _filesystem: Filesystem, + paths: string[], + private readonly _onChange: () => void, + private readonly _logger?: LDLogger, + ) { + this._paths = Array.from(new Set(paths)); + this._directories = Array.from(new Set(paths.map(directoryOf))); + paths.forEach((path) => { + const directory = directoryOf(path); + this._names[directory] = this._names[directory] ?? new Set(); + this._names[directory].add(basenameOf(path)); + this._files[directory] = this._files[directory] ?? []; + this._files[directory].push(path); + }); + } + + /** + * Sets up the watches. Returns after the first attempt. Failures are retried in the + * background. + */ + start(): void { + this._setupWatches(false); + this._armFileWatches(); + // Record the files' metadata, so that a later event for another entry can be compared to it. + this._directories.forEach((directory) => this._checkFiles(directory, false)); + } + + close(): void { + this._closed = true; + if (this._retryTimer) { + clearTimeout(this._retryTimer); + this._retryTimer = undefined; + } + Object.keys(this._watches).forEach((directory) => this._dropWatch(directory)); + Object.keys(this._fileWatches).forEach((path) => this._dropFileWatch(path)); + } + + /** + * Checks that every watched directory still exists. A directory that does not exist has its + * watch closed and set up again once it exists. Call it after a failed load, because a load + * that fails to read a file can mean that the file's directory is gone. + */ + verify(): void { + this._directories.forEach((directory) => this._checkDirectory(directory)); + // A file that did not exist when its watch was last attempted may exist now. + this._armFileWatches(); + } + + /** + * Sets up a direct watch on each configured file that has none. A file that cannot be watched, + * usually because it does not exist yet, is attempted again at the next event or verify call. + */ + private _armFileWatches(): void { + if (this._closed) { + return; + } + this._paths.forEach((path) => this._armFileWatch(path)); + } + + private _armFileWatch(path: string): void { + if (this._closed || this._fileWatches[path]) { + return; + } + try { + this._fileWatches[path] = this._filesystem.watch(path, () => this._handleFileEvent(path)); + } catch { + // The directory watch reports the file when it appears, and the watch is attempted again. + } + } + + private _dropFileWatch(path: string): void { + const handle = this._fileWatches[path]; + if (!handle) { + return; + } + delete this._fileWatches[path]; + handle.close(); + } + + /** + * Sets the direct watch on a configured file up again. A direct watch follows the path to the + * inode it names when the watch is set up. After the path was replaced or its link retargeted, + * the watch is on an inode that the path no longer names, and edits to the new target are + * invisible to it and to the directory watch alike. + */ + private _renewFileWatch(path: string): void { + this._dropFileWatch(path); + this._armFileWatch(path); + } + + /** + * An event from the direct watch on a configured file. The file or the target of its link + * changed, so the callback runs. The watch may now be on an inode that the path no longer + * names, so it is set up again. An error from the watch is treated the same way: the watch + * cannot be trusted anymore, and the file can have changed while it was failing, which the + * directory watch does not see when the path is a link to a file elsewhere. + */ + private _handleFileEvent(path: string): void { + if (this._closed || !this._fileWatches[path]) { + return; + } + this._renewFileWatch(path); + this._onChange(); + this._checkFiles(directoryOf(path), false); + } + + private _setupWatches(isRetry: boolean): void { + if (this._closed) { + return; + } + const added: string[] = []; + let failed = false; + this._directories.forEach((directory) => { + if (this._watches[directory]) { + return; + } + try { + this._watches[directory] = new DirectoryWatch( + this._filesystem, + directory, + (watch, eventType, changedName) => + this._handleEvent(watch, directory, eventType, changedName), + ); + added.push(directory); + } catch (err) { + failed = true; + this._logger?.error( + `Unable to watch directory "${directory}": ${err instanceof Error ? err.message : err}`, + ); + } + }); + if (failed) { + this._scheduleSetup(); + } + if (isRetry && added.length > 0) { + // Changes could have happened in a directory while its watch was not in place. The + // catch-up runs for every directory watched in this pass, whether or not another + // directory still cannot be watched: that one gets its own catch-up when it can be. + this._onChange(); + added.forEach((directory) => this._checkFiles(directory, false)); + // The direct watch on a file in the directory ended with it, or never existed when the + // directory was missing at start. It is set up again with the directory watch. + this._armFileWatches(); + } + } + + private _scheduleSetup(): void { + if (this._closed || this._retryTimer) { + return; + } + this._retryTimer = setTimeout(() => { + this._retryTimer = undefined; + this._setupWatches(true); + }, WATCH_RETRY_DELAY_MS); + } + + private _handleEvent( + watch: DirectoryWatch, + directory: string, + eventType: string, + changedName?: string, + ): void { + if (this._closed || watch.closed) { + return; + } + if (eventType === 'error') { + // The platform reported that the watch itself failed. It cannot be trusted anymore. + this._logger?.warn(`The watch on directory "${directory}" failed. Setting it up again.`); + this._replaceWatch(directory); + return; + } + if (changedName === undefined || this._names[directory].has(changedName)) { + this._onChange(); + // Keep the metadata current, so that a later event for another entry is compared to the + // state after this change. + this._checkFiles(directory, false); + // The file may have appeared, been replaced, or had its link retargeted, so its direct + // watch is set up again; the others are set up if they are missing. + this._files[directory] + .filter((path) => changedName === undefined || basenameOf(path) === changedName) + .forEach((path) => this._renewFileWatch(path)); + this._armFileWatches(); + } else { + // Another entry changed. A configured file that is a symbolic link can have changed with + // it. Its metadata decides. + this._checkFiles(directory, true); + } + // The directory itself may be gone. Check on every event, whatever entry it names. + this._checkDirectory(directory); + } + + /** + * Reads the metadata of the configured files in a directory and remembers it. When `notify` is + * set and any file's metadata differs from the last observation, the change callback runs, and + * the direct watch on each changed file is set up again, because its link can point elsewhere + * now. Reads for one directory do not overlap. A read requested during a read runs afterwards. + * A request to notify that arrives during a read applies to that read's result as well, since + * the read may already have observed the change the request is about; when the read is the + * first observation, which has nothing to compare to, such a request counts as a change. + */ + private _checkFiles(directory: string, notify: boolean): void { + if (this._closed || !this._filesystem.getFileStats) { + return; + } + const pending = this._fileChecks[directory]; + if (pending) { + pending.again = true; + pending.notify = pending.notify || notify; + return; + } + const check = { again: false, notify: false }; + this._fileChecks[directory] = check; + const { getFileStats } = this._filesystem; + const files = this._files[directory]; + Promise.all( + files.map(async (path) => { + try { + const stats = await getFileStats.call(this._filesystem, path); + return stats ? `${stats.timestamp}:${stats.size}` : 'absent'; + } catch { + return 'error'; + } + }), + ).then((signatures) => { + delete this._fileChecks[directory]; + if (this._closed) { + return; + } + let changed = false; + let firstObservation = false; + files.forEach((path, index) => { + if (this._signatures[path] !== signatures[index]) { + if (this._signatures[path] !== undefined) { + changed = true; + this._renewFileWatch(path); + } else { + firstObservation = true; + } + this._signatures[path] = signatures[index]; + } + }); + if ((notify || check.notify) && (changed || (check.notify && firstObservation))) { + this._onChange(); + } + if (check.again) { + this._checkFiles(directory, check.notify); + } + }); + } + + /** + * Closes and drops the watch on a directory that no longer exists, and sets it up again once + * the directory exists. Closing the watch is what ends the flood of events that some + * platforms deliver for a deleted directory. + */ + private _checkDirectory(directory: string): void { + if (this._closed || !this._watches[directory]) { + return; + } + const pending = this._checks[directory]; + if (pending) { + pending.again = true; + return; + } + const check = { again: false }; + this._checks[directory] = check; + directoryExists(this._filesystem, directory).then((exists) => { + delete this._checks[directory]; + if (this._closed || !this._watches[directory]) { + return; + } + if (exists) { + if (check.again) { + // The deletion of a file and of its directory arrive as separate events. The check + // for the first can complete before the second happens. + this._checkDirectory(directory); + } + return; + } + this._logger?.warn( + `Directory "${directory}" no longer exists. Its watch is set up again when it appears.`, + ); + this._replaceWatch(directory); + }); + } + + private _replaceWatch(directory: string): void { + this._dropWatch(directory); + // The setup fails until the directory exists again, and the retry keeps trying. + this._setupWatches(true); + } + + private _dropWatch(directory: string): void { + const watch = this._watches[directory]; + if (!watch) { + return; + } + watch.closed = true; + delete this._watches[directory]; + watch.handle.close(); + } +} diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts new file mode 100644 index 0000000000..254686f481 --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts @@ -0,0 +1,386 @@ +import { Filesystem, LDLogger, monotonicNow } from '@launchdarkly/js-sdk-common'; + +import { Flag } from '../../evaluation/data/Flag'; +import { FileDataDocument, isFileNotFoundError } from './document'; +import { FileDataPolicy } from './FileDataPolicy'; +import { FileSummary, mergeDocuments, MergeResult } from './merge'; + +/** + * A settle window long enough to coalesce the burst of change notifications that one file + * edit produces, and short enough to stay responsive. + * + * @internal + */ +export const DEFAULT_DEBOUNCE_DELAY_MS = 100; + +/** + * Bounds how long a failed reload can go uncorrected when no further change notification + * arrives, for example when the failure came from reading a file while it was written. + * Reading a local file is cheap, so this can be short. + * + * @internal + */ +export const DEFAULT_RETRY_DELAY_MS = 1000; + +/** + * The longest a burst of change notifications can delay a reload. A settle window restarts with + * each notification, so a sustained stream of notifications could postpone a reload without end. + * A reload runs no later than this long after the first notification of a burst. + * + * @internal + */ +export const MAX_DEBOUNCE_DELAY_MS = 1000; + +/** + * The result of one successful reload: the merged items plus a summary of each configured + * file. + * + * @internal + */ +export interface ReloadResult extends MergeResult { + /** + * One summary for each configured file, in order. + */ + files: FileSummary[]; +} + +/** + * Why a load failed. + * + * - `read`: a file could not be read. `path` names it. `error` is the filesystem error. + * - `parse`: a file could not be parsed. `path` names it. `error` is the parser's error. + * - `merge`: the documents could not be combined, for example because of a duplicate key. + * - `apply`: the consumer's `apply` threw while translating or storing the merged result, for + * example because an entry is not an object. The result is not remembered as the last good one. + * + * @internal + */ +export type LoadFailureKind = 'read' | 'parse' | 'merge' | 'apply'; + +/** + * A failed load. `repeated` is true when an automatic retry hit a failure of the same kind and + * path as the previous report, with no success in between. A failure after a change + * notification is never a repeat: the files changed, so their failure is news. The message is + * not compared: a file that is being written fails to parse with a different message on each + * attempt, and that is one problem. Consumers use it to report a persistent failure once. + * + * @internal + */ +export interface LoadFailure { + kind: LoadFailureKind; + error: Error; + path?: string; + repeated: boolean; +} + +/** + * @internal + */ +export interface FileReloaderConfig { + /** + * The files to load. The order is significant: it decides which file wins under the policy's + * duplicate key resolution. + */ + paths: string[]; + /** + * How documents become data. + */ + policy: FileDataPolicy; + filesystem: Filesystem; + logger?: LDLogger; + /** + * Receives each successfully merged result. Calls are serialized: one reload completes before + * the next starts. + */ + apply: (result: ReloadResult) => void; + /** + * Receives each failed load. The reloader does not log failures itself. The consumer decides + * how to report them, and can use `repeated` to report a persistent failure once. + */ + onFailure: (failure: LoadFailure) => void; + /** + * How long to wait after a trigger for further triggers to settle before reloading. Zero or + * negative reloads immediately on each trigger. + */ + debounceDelayMs: number; + /** + * How long to wait after a failed reload before retrying it without a trigger. Zero or + * negative disables the automatic retry. + */ + retryDelayMs: number; + /** + * When true, a reload whose file contents are identical to the last applied contents does not + * call `apply`. + */ + skipUnchanged: boolean; +} + +type ReloadReason = 'initial' | 'change' | 'retry'; + +function asError(err: unknown): Error { + return err instanceof Error ? err : new Error(String(err)); +} + +/** + * Owns the reload cycle for a set of data files. It serializes reloads, debounces change + * signals, keeps the last good result on failure (by not calling `apply`), retries after + * failures, and skips no-op applications. + * + * @internal + */ +export default class FileReloader { + private _closed = false; + + // Every reload is chained onto this promise, so reloads never overlap. + private _chain: Promise = Promise.resolve(); + + // The reload that waits for the chain, if any. Further requests join it. + private _queued?: Promise; + + private _debounceTimer?: ReturnType; + + // When the current burst of triggers began. A reload runs no later than MAX_DEBOUNCE_DELAY_MS + // after it. + private _burstStartedAt = 0; + + private _retryTimer?: ReturnType; + + private _lastGoodContent?: string; + + // The flags of the last applied result, keyed by their key property, for the policy's version + // handling. A failed reload leaves them as they were. + private _lastGoodFlags: Record = {}; + + private _lastFailureKey?: string; + + constructor(private readonly _config: FileReloaderConfig) {} + + /** + * Loads the files and applies the result, or reports the failure. Use it for the initial + * load. A failure schedules the same automatic retry as a failed triggered reload. + * + * @returns A promise that resolves when this reload has completed. It never rejects. + */ + reloadNow(): Promise { + return this._enqueue('initial'); + } + + /** + * Signals that the files may have changed. The reload happens after the debounce delay, or no + * later than the maximum debounce delay after the first trigger of a burst. Triggers that + * arrive while a reload is pending are coalesced into it. + */ + trigger(): void { + if (this._closed) { + return; + } + if (this._config.debounceDelayMs <= 0) { + this._enqueue('change'); + return; + } + // A monotonic clock, so that a wall-clock adjustment cannot lengthen or cut short the bound. + const now = monotonicNow(); + if (this._debounceTimer) { + clearTimeout(this._debounceTimer); + this._debounceTimer = undefined; + } else { + this._burstStartedAt = now; + } + // Each trigger moves the deadline out again, so the reload runs after activity settles. A + // burst that does not settle cannot postpone the reload past the maximum delay. + const remaining = this._burstStartedAt + MAX_DEBOUNCE_DELAY_MS - now; + if (remaining <= 0) { + this._enqueue('change'); + return; + } + this._debounceTimer = setTimeout( + () => { + this._debounceTimer = undefined; + this._enqueue('change'); + }, + Math.min(this._config.debounceDelayMs, remaining), + ); + } + + /** + * Stops the reloader. A reload that is in progress finishes its file reads, but it does not + * call `apply` or `onFailure`. + */ + close(): void { + this._closed = true; + if (this._debounceTimer) { + clearTimeout(this._debounceTimer); + this._debounceTimer = undefined; + } + this._clearRetry(); + } + + private _enqueue(reason: ReloadReason): Promise { + if (this._queued) { + return this._queued; + } + const run = this._chain.then(() => { + this._queued = undefined; + return this._reload(reason).catch((err) => { + // The reload reports every failure it can foresee. This is the last line of defense, so + // that one unexpected exception cannot reject a promise the caller dropped. + this._config.logger?.error(`Unexpected error while reloading flag data: ${err}`); + }); + }); + this._queued = run; + // The chain must never be a rejected promise. A later then on it would skip its callback, + // leave the queued reload set forever, and no reload would ever run again. + this._chain = run.catch(() => undefined); + return run; + } + + private _clearRetry(): void { + if (this._retryTimer) { + clearTimeout(this._retryTimer); + this._retryTimer = undefined; + } + } + + private _armRetry(): void { + if (this._closed || this._config.retryDelayMs <= 0) { + return; + } + this._clearRetry(); + this._retryTimer = setTimeout(() => { + this._retryTimer = undefined; + this._enqueue('retry'); + }, this._config.retryDelayMs); + } + + /** + * Performs one full load of all configured files. The whole set is read on every reload. + * Entries are combined across files in order, so a change to one file can alter which file + * wins for a key. Every filesystem and parsing failure is caught and reported, so the returned + * promise never rejects. + */ + private async _reload(reason: ReloadReason): Promise { + if (this._closed) { + return; + } + const { logger, paths, policy } = this._config; + if (reason === 'change') { + logger?.info('Reloading flag data after detecting a change'); + } else if (reason === 'retry') { + logger?.debug('Retrying flag data load after earlier failure'); + } + // A pending retry is superseded by this reload. It either succeeds, or it fails and arms a + // fresh retry. + this._clearRetry(); + + const documents: FileDataDocument[] = []; + const files: FileSummary[] = []; + const contents: string[] = []; + let failure: LoadFailure | undefined; + + const readAll = paths.reduce( + (previous, path) => + previous.then(async () => { + if (failure) { + return; + } + let data: string; + try { + data = await this._config.filesystem.readFile(path); + } catch (err) { + if (policy.missingFile === 'skip' && isFileNotFoundError(err)) { + logger?.debug(`File ${path} does not exist. It contributes no data.`); + files.push({ path, present: false, flags: 0, segments: 0 }); + return; + } + failure = { kind: 'read', error: asError(err), path, repeated: false }; + return; + } + contents.push(data); + try { + documents.push(policy.parseDocument(path, data)); + } catch (err) { + failure = { kind: 'parse', error: asError(err), path, repeated: false }; + return; + } + files.push({ path, present: true, flags: 0, segments: 0 }); + }), + Promise.resolve(), + ); + await readAll; + + let merged: MergeResult | undefined; + if (!failure) { + try { + merged = mergeDocuments(policy, documents, this._lastGoodFlags); + } catch (err) { + failure = { kind: 'merge', error: asError(err), repeated: false }; + } + } + if (failure || !merged) { + this._fail(failure!, reason); + return; + } + + // The documents are the present files in order. Copy their counts onto the summaries. + let next = 0; + files.forEach((file) => { + if (file.present) { + file.flags = merged!.documents[next].flags; + file.segments = merged!.documents[next].segments; + next += 1; + } + }); + + // The reloader may have been closed while the files were read. Deliver nothing then. + if (this._closed) { + return; + } + + // A success right after a failure applies even when the content is unchanged since the + // last success. The consumer heard about the failure and only `apply` tells it that things + // are good again. + const recovering = this._lastFailureKey !== undefined; + const content = JSON.stringify(contents); + if (this._config.skipUnchanged && !recovering && content === this._lastGoodContent) { + this._lastFailureKey = undefined; + return; + } + // Nothing is remembered until the consumer has accepted the result. A result the consumer + // rejects must not become the baseline that skip-unchanged compares against, and must not + // count as a recovery. + const nextFlags: Record = {}; + try { + merged.flags.forEach(({ item }) => { + nextFlags[String(item.key)] = item; + }); + this._config.apply({ ...merged, files }); + } catch (err) { + this._fail({ kind: 'apply', error: asError(err), repeated: false }, reason); + return; + } + this._lastFailureKey = undefined; + this._lastGoodContent = content; + this._lastGoodFlags = nextFlags; + } + + private _fail(failure: LoadFailure, reason: ReloadReason): void { + if (this._closed) { + return; + } + // With automatic retries, a persistent failure would repeat the same report on every + // attempt. The consumer learns that the failure is a repeat and can demote it. Only a retry + // can be a repeat: a failure after a change notification is news. The key leaves out the + // message, because a file that is being written fails with a different parse message on + // each attempt and that is one problem, not many. + const key = `${failure.kind}:${failure.path ?? ''}`; + const repeated = reason === 'retry' && key === this._lastFailureKey; + this._lastFailureKey = key; + try { + this._config.onFailure({ ...failure, repeated }); + } catch (err) { + // A consumer that throws while it reports a failure must not stop the retry. + this._config.logger?.error(`Error while reporting a flag data load failure: ${err}`); + } + this._armRetry(); + } +} diff --git a/packages/shared/sdk-server/src/data_sources/filedata/document.ts b/packages/shared/sdk-server/src/data_sources/filedata/document.ts new file mode 100644 index 0000000000..d535c9e86b --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/document.ts @@ -0,0 +1,69 @@ +import { Flag } from '../../evaluation/data/Flag'; +import { Segment } from '../../evaluation/data/Segment'; + +/** + * A function that parses YAML text into the same structure that `JSON.parse` produces. + * + * @internal + */ +export type YamlParser = (data: string) => any; + +/** + * The parsed form of one data file. A document can hold full flag definitions, flag key to + * value entries, and segment definitions. Every member is optional. + * + * @internal + */ +export interface FileDataDocument { + flags?: Record; + flagValues?: Record; + segments?: Record; +} + +/** + * Parses the text of one file into a document. + * + * @internal + */ +export type DocumentParser = (path: string, data: string) => FileDataDocument; + +/** + * Reports whether a path uses a YAML file extension. + * + * @internal + */ +export function isYamlPath(path: string): boolean { + return path.endsWith('.yml') || path.endsWith('.yaml'); +} + +function parseYaml(path: string, data: string, yamlParser?: YamlParser): any { + if (!yamlParser) { + throw new Error(`Attempted to parse yaml file (${path}) without parser.`); + } + return yamlParser(data); +} + +/** + * A parser that chooses the format by file extension only. A `.yml` or `.yaml` file uses the + * YAML parser, and every other file is JSON. The parsed value is returned as it is. This is the + * rule of the file data sources. + * + * @internal + */ +export function parseDocumentByExtension(yamlParser?: YamlParser): DocumentParser { + return (path, data) => { + if (isYamlPath(path)) { + return parseYaml(path, data, yamlParser); + } + return JSON.parse(data); + }; +} + +/** + * Reports whether a filesystem error means that the file does not exist. + * + * @internal + */ +export function isFileNotFoundError(err: unknown): boolean { + return (err as { code?: string } | undefined)?.code === 'ENOENT'; +} diff --git a/packages/shared/sdk-server/src/data_sources/filedata/index.ts b/packages/shared/sdk-server/src/data_sources/filedata/index.ts new file mode 100644 index 0000000000..f89f99c7bc --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/index.ts @@ -0,0 +1,51 @@ +import { + DocumentParser, + FileDataDocument, + isFileNotFoundError, + isYamlPath, + parseDocumentByExtension, + YamlParser, +} from './document'; +import { + DuplicateKeyResolution, + FileDataPolicy, + ItemCategory, + MissingFileHandling, +} from './FileDataPolicy'; +import FileDirectoryWatcher, { directoryOf } from './FileDirectoryWatcher'; +import FileReloader, { + DEFAULT_DEBOUNCE_DELAY_MS, + DEFAULT_RETRY_DELAY_MS, + FileReloaderConfig, + LoadFailure, + LoadFailureKind, + ReloadResult, +} from './FileReloader'; +import { DocumentSummary, FileSummary, KeyedItem, mergeDocuments, MergeResult } from './merge'; + +export { + DEFAULT_DEBOUNCE_DELAY_MS, + DEFAULT_RETRY_DELAY_MS, + directoryOf, + DocumentParser, + DocumentSummary, + DuplicateKeyResolution, + FileDataDocument, + FileDataPolicy, + FileDirectoryWatcher, + FileReloader, + FileReloaderConfig, + FileSummary, + isFileNotFoundError, + isYamlPath, + ItemCategory, + KeyedItem, + LoadFailure, + LoadFailureKind, + mergeDocuments, + MergeResult, + MissingFileHandling, + parseDocumentByExtension, + ReloadResult, + YamlParser, +}; diff --git a/packages/shared/sdk-server/src/data_sources/filedata/merge.ts b/packages/shared/sdk-server/src/data_sources/filedata/merge.ts new file mode 100644 index 0000000000..34722c35c6 --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/merge.ts @@ -0,0 +1,133 @@ +import { Flag } from '../../evaluation/data/Flag'; +import { Segment } from '../../evaluation/data/Segment'; +import { FileDataDocument } from './document'; +import { FileDataPolicy, ItemCategory } from './FileDataPolicy'; + +/** + * One merged item with the key it was listed under in its document. + * + * @internal + */ +export interface KeyedItem { + key: string; + item: T; +} + +/** + * Counts the entries the merge kept from one document. An entry that the duplicate key + * resolution dropped, or that a later document replaced, is not counted. + * + * @internal + */ +export interface DocumentSummary { + flags: number; + segments: number; +} + +/** + * Describes one configured file after a reload. + * + * @internal + */ +export interface FileSummary { + path: string; + /** + * False when the file does not exist and missing files contribute nothing. + */ + present: boolean; + flags: number; + segments: number; +} + +/** + * The merged items from one or more documents. + * + * Ordering is deterministic at document granularity only. All of one document's items precede + * the next document's items, in the order the documents were given. The order of the items + * within one document is unspecified. Consumers key items by their `key`. + * + * @internal + */ +export interface MergeResult { + flags: KeyedItem[]; + segments: KeyedItem[]; + /** + * One summary for each input document, in order. + */ + documents: DocumentSummary[]; +} + +/** + * Combines the items of the given documents in order. `flags` and `segments` entries are keyed + * by the policy's `entryKey`, and `flagValues` entries are expanded by the policy and keyed by + * their map key. The policy decides what happens when a key appears more than once: keep the + * first entry, keep the last entry, or fail the load. Members that are not objects contribute + * no entries. + * + * @param policy The source's translation policy. + * @param documents The parsed documents, in the configured order. + * @param previousFlags The flags the last successful load produced, keyed by flag key. The + * policy can use them to carry versions forward. + * + * @internal + */ +export function mergeDocuments( + policy: FileDataPolicy, + documents: FileDataDocument[], + previousFlags: Record = {}, +): MergeResult { + const flags: KeyedItem[] = []; + const segments: KeyedItem[] = []; + const summaries: DocumentSummary[] = documents.map(() => ({ flags: 0, segments: 0 })); + // For each kept key: its position in the items array and the document it came from. + const positions: Record> = { + flag: new Map(), + segment: new Map(), + }; + + const insert = ( + items: KeyedItem[], + category: ItemCategory, + key: string, + item: T, + documentIndex: number, + ): void => { + const counter = category === 'flag' ? 'flags' : 'segments'; + const existing = positions[category].get(key); + if (existing) { + if (policy.resolveDuplicateKey(category, key) === 'keepFirst') { + return; + } + // The later entry replaces the earlier one in place, so the earlier document's count + // moves to the later document. + items[existing.index] = { key, item }; + summaries[existing.document][counter] -= 1; + summaries[documentIndex][counter] += 1; + positions[category].set(key, { index: existing.index, document: documentIndex }); + return; + } + items.push({ key, item }); + positions[category].set(key, { index: items.length - 1, document: documentIndex }); + summaries[documentIndex][counter] += 1; + }; + + documents.forEach((document, documentIndex) => { + Object.entries(document.flags ?? {}).forEach(([key, flag]) => { + insert(flags, 'flag', policy.entryKey(key, flag), flag, documentIndex); + }); + Object.entries(document.flagValues ?? {}).forEach(([key, value]) => { + insert( + flags, + 'flag', + key, + policy.makeFlagWithValue(key, value, previousFlags[key]), + documentIndex, + ); + }); + Object.entries(document.segments ?? {}).forEach(([key, segment]) => { + insert(segments, 'segment', policy.entryKey(key, segment), segment, documentIndex); + }); + }); + + return { flags, segments, documents: summaries }; +}