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/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 }; +}