From 34cca69c031e7f12a993fc9297f21cf9e5ab964f Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Mon, 5 Oct 2026 22:32:28 +0000 Subject: [PATCH 1/5] feat: Add a shared file data engine Adds FileReloader, FileDirectoryWatcher, mergeDocuments, parseDocumentByExtension, and the FileDataPolicy interface under packages/shared/sdk-server/src/data_sources/filedata, with their tests. Nothing uses the engine yet. The file data source and the FDv2 file data initializer move onto it in the following change. --- .../filedata/FileDirectoryWatcher.test.ts | 381 ++++++++++ .../filedata/FileReloader.test.ts | 695 ++++++++++++++++++ .../data_sources/filedata/MockFilesystem.ts | 198 +++++ .../data_sources/filedata/document.test.ts | 82 +++ .../data_sources/filedata/merge.test.ts | 141 ++++ .../data_sources/filedata/testPolicy.ts | 27 + .../data_sources/filedata/FileDataPolicy.ts | 70 ++ .../filedata/FileDirectoryWatcher.ts | 404 ++++++++++ .../src/data_sources/filedata/FileReloader.ts | 385 ++++++++++ .../src/data_sources/filedata/document.ts | 69 ++ .../src/data_sources/filedata/index.ts | 51 ++ .../src/data_sources/filedata/merge.ts | 133 ++++ 12 files changed, 2636 insertions(+) create mode 100644 packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts create mode 100644 packages/shared/sdk-server/__tests__/data_sources/filedata/FileReloader.test.ts create mode 100644 packages/shared/sdk-server/__tests__/data_sources/filedata/MockFilesystem.ts create mode 100644 packages/shared/sdk-server/__tests__/data_sources/filedata/document.test.ts create mode 100644 packages/shared/sdk-server/__tests__/data_sources/filedata/merge.test.ts create mode 100644 packages/shared/sdk-server/__tests__/data_sources/filedata/testPolicy.ts create mode 100644 packages/shared/sdk-server/src/data_sources/filedata/FileDataPolicy.ts create mode 100644 packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts create mode 100644 packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts create mode 100644 packages/shared/sdk-server/src/data_sources/filedata/document.ts create mode 100644 packages/shared/sdk-server/src/data_sources/filedata/index.ts create mode 100644 packages/shared/sdk-server/src/data_sources/filedata/merge.ts 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..4acb21310b --- /dev/null +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts @@ -0,0 +1,381 @@ +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('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('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('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..c951940098 --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts @@ -0,0 +1,404 @@ +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, (eventType) => + this._handleFileEvent(path, eventType), + ); + } 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(); + } + + /** + * 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. + */ + private _handleFileEvent(path: string, eventType: string): void { + if (this._closed || !this._fileWatches[path]) { + return; + } + this._dropFileWatch(path); + this._armFileWatch(path); + if (eventType !== 'error') { + this._onChange(); + this._checkFiles(directoryOf(path), false); + } + } + + private _setupWatches(isRetry: boolean): void { + if (this._closed) { + return; + } + 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), + ); + } catch (err) { + failed = true; + this._logger?.error( + `Unable to watch directory "${directory}": ${err instanceof Error ? err.message : err}`, + ); + } + }); + if (failed) { + this._scheduleSetup(); + return; + } + if (isRetry) { + // Changes could have happened while the watch was not in place. + this._onChange(); + this._directories.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, or been replaced under a watch that ended with it. + 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. + * Reads for one directory do not overlap. A read requested during a read runs afterwards, and + * notifies when either request asked for it. + */ + 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; + files.forEach((path, index) => { + if (this._signatures[path] !== signatures[index]) { + if (this._signatures[path] !== undefined) { + changed = true; + } + this._signatures[path] = signatures[index]; + } + }); + if (notify && changed) { + 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..cf3e266dd7 --- /dev/null +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts @@ -0,0 +1,385 @@ +import { Filesystem, LDLogger } 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; + } + const now = Date.now(); + 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 }; +} From bcfbb0239e2f1d2782ba65ee2f642823ae839a7a Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Mon, 5 Oct 2026 22:44:55 +0000 Subject: [PATCH 2/5] fix: Reload when a direct file watch reports an error A direct watch on a configured file that reported an error was dropped and set up again, but no reload ran. 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. An error from a direct file watch is now handled like a change event: the watch is renewed and the callback runs, as a directory watch already does when it is replaced. --- .../filedata/FileDirectoryWatcher.test.ts | 15 +++++++++++++++ .../filedata/FileDirectoryWatcher.ts | 16 +++++++--------- 2 files changed, 22 insertions(+), 9 deletions(-) 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 index 4acb21310b..3eaa540ba2 100644 --- a/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts @@ -68,6 +68,21 @@ describe('given a directory watcher over a mock filesystem', () => { 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']); diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts index c951940098..d0d921f354 100644 --- a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts @@ -186,9 +186,7 @@ export default class FileDirectoryWatcher { return; } try { - this._fileWatches[path] = this._filesystem.watch(path, (eventType) => - this._handleFileEvent(path, eventType), - ); + 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. } @@ -206,18 +204,18 @@ export default class FileDirectoryWatcher { /** * 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. + * 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, eventType: string): void { + private _handleFileEvent(path: string): void { if (this._closed || !this._fileWatches[path]) { return; } this._dropFileWatch(path); this._armFileWatch(path); - if (eventType !== 'error') { - this._onChange(); - this._checkFiles(directoryOf(path), false); - } + this._onChange(); + this._checkFiles(directoryOf(path), false); } private _setupWatches(isRetry: boolean): void { From 60a16b3fa32a9c6d6caa2cc356fc3475da101bac Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:38:21 +0000 Subject: [PATCH 3/5] fix: Run the catch-up reload for each directory that becomes watchable When a watch could not be set up for every directory, the setup returned before the catch-up that reloads after a watch gap, so a directory whose watch was created in that pass got no reload while another directory still could not be watched, and the next retry skipped it because it was watched by then. A file written into the directory before its watch was in place was never loaded. A path in a directory that never exists is a valid configuration, so the catch-up now runs for the directories watched in each pass, and the retry continues for the rest. --- .../filedata/FileDirectoryWatcher.test.ts | 43 +++++++++++++++++++ .../filedata/FileDirectoryWatcher.ts | 11 +++-- 2 files changed, 50 insertions(+), 4 deletions(-) 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 index 3eaa540ba2..b2697ba109 100644 --- a/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts @@ -150,6 +150,49 @@ describe('given a directory watcher over a mock filesystem', () => { 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']); diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts index d0d921f354..4648cbfb67 100644 --- a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts @@ -222,6 +222,7 @@ export default class FileDirectoryWatcher { if (this._closed) { return; } + const added: string[] = []; let failed = false; this._directories.forEach((directory) => { if (this._watches[directory]) { @@ -234,6 +235,7 @@ export default class FileDirectoryWatcher { (watch, eventType, changedName) => this._handleEvent(watch, directory, eventType, changedName), ); + added.push(directory); } catch (err) { failed = true; this._logger?.error( @@ -243,12 +245,13 @@ export default class FileDirectoryWatcher { }); if (failed) { this._scheduleSetup(); - return; } - if (isRetry) { - // Changes could have happened while the watch was not in place. + 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(); - this._directories.forEach((directory) => this._checkFiles(directory, false)); + 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(); From 07851e03c2a40d631319ce26f1a0278c17ad3646 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Thu, 8 Oct 2026 22:19:17 +0000 Subject: [PATCH 4/5] fix: Measure the settle bound with the shared monotonic clock The bound on how long a stream of change notifications can postpone a reload was measured with the wall clock, which a clock adjustment can move. It now uses monotonicNow from the common package, which reads a monotonic clock where the runtime has one. --- .../sdk-server/src/data_sources/filedata/FileReloader.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts index cf3e266dd7..254686f481 100644 --- a/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileReloader.ts @@ -1,4 +1,4 @@ -import { Filesystem, LDLogger } from '@launchdarkly/js-sdk-common'; +import { Filesystem, LDLogger, monotonicNow } from '@launchdarkly/js-sdk-common'; import { Flag } from '../../evaluation/data/Flag'; import { FileDataDocument, isFileNotFoundError } from './document'; @@ -178,7 +178,8 @@ export default class FileReloader { this._enqueue('change'); return; } - const now = Date.now(); + // 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; From 1784d775e8aceee994ebe721b930d2eb427e55b1 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Thu, 8 Oct 2026 22:34:14 +0000 Subject: [PATCH 5/5] fix: Renew direct file watches on retargets and keep queued notify requests A direct watch on a configured file follows the path to an inode when it is set up. When a directory event named the file, or the file's metadata changed through another entry such as a link swap, the existing direct watch was kept, so after a retarget it stayed on the old target and edits to the new one were invisible to it and to the directory watch alike. The direct watch is now set up again in both cases. A request to notify that arrived while a metadata read was in flight was only applied to the read that ran afterwards, which compared against the metadata the first read had already recorded and so found no change. The request now applies to the read in flight as well, and during the first observation, which has nothing to compare to, it counts as a change. --- .../filedata/FileDirectoryWatcher.test.ts | 72 +++++++++++++++++++ .../filedata/FileDirectoryWatcher.ts | 35 +++++++-- 2 files changed, 100 insertions(+), 7 deletions(-) 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 index b2697ba109..3ba9dbca9d 100644 --- a/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts +++ b/packages/shared/sdk-server/__tests__/data_sources/filedata/FileDirectoryWatcher.test.ts @@ -121,6 +121,78 @@ describe('given a directory watcher over a mock filesystem', () => { 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']); diff --git a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts index 4648cbfb67..4e9b5fdebc 100644 --- a/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts +++ b/packages/shared/sdk-server/src/data_sources/filedata/FileDirectoryWatcher.ts @@ -201,6 +201,17 @@ export default class FileDirectoryWatcher { 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 @@ -212,8 +223,7 @@ export default class FileDirectoryWatcher { if (this._closed || !this._fileWatches[path]) { return; } - this._dropFileWatch(path); - this._armFileWatch(path); + this._renewFileWatch(path); this._onChange(); this._checkFiles(directoryOf(path), false); } @@ -288,7 +298,11 @@ export default class FileDirectoryWatcher { // 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, or been replaced under a watch that ended with it. + // 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 @@ -301,9 +315,12 @@ export default class FileDirectoryWatcher { /** * 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. - * Reads for one directory do not overlap. A read requested during a read runs afterwards, and - * notifies when either request asked for it. + * 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) { @@ -334,15 +351,19 @@ export default class FileDirectoryWatcher { 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 && changed) { + if ((notify || check.notify) && (changed || (check.notify && firstObservation))) { this._onChange(); } if (check.again) {