diff --git a/packages/store/node-server-sdk-redis/__tests__/RedisCoreAvailability.test.ts b/packages/store/node-server-sdk-redis/__tests__/RedisCoreAvailability.test.ts new file mode 100644 index 0000000000..f6f4f5e13f --- /dev/null +++ b/packages/store/node-server-sdk-redis/__tests__/RedisCoreAvailability.test.ts @@ -0,0 +1,124 @@ +import { PersistentDataStoreWrapper } from '@launchdarkly/node-server-sdk'; + +import RedisCore from '../src/RedisCore'; +import RedisFeatureStore from '../src/RedisFeatureStore'; + +jest.mock('@launchdarkly/node-server-sdk', () => { + const actual = jest.requireActual('@launchdarkly/node-server-sdk'); + return { + ...actual, + PersistentDataStoreWrapper: jest.fn(), + }; +}); + +beforeEach(() => { + jest.clearAllMocks(); +}); + +function makeState(overrides: object) { + return { + prefixedKey: (key: string) => key, + isConnected: () => true, + isInitialConnection: () => false, + ...overrides, + }; +} + +it('reports an init error through the callback when the transaction fails', (done) => { + const state = makeState({ + getClient: () => ({ + multi: () => ({ + del: jest.fn(), + hmset: jest.fn(), + set: jest.fn(), + exec: (cb: (err: Error | null) => void) => cb(new Error('connection refused')), + }), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + core.init([], (err) => { + expect(err).toEqual(new Error('connection refused')); + done(); + }); +}); + +it('calls back without an error when init succeeds', (done) => { + const state = makeState({ + getClient: () => ({ + multi: () => ({ + del: jest.fn(), + hmset: jest.fn(), + set: jest.fn(), + exec: (cb: (err: Error | null) => void) => cb(null), + }), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + core.init([], (err) => { + expect(err).toBeUndefined(); + done(); + }); +}); + +it('calls back true from isStoreAvailable when the check succeeds', (done) => { + const state = makeState({ + getClient: () => ({ + exists: (_key: string, cb: (err: Error | null, count: number) => void) => cb(null, 0), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + core.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(true); + done(); + }); +}); + +it('calls back false from isStoreAvailable when the check fails', (done) => { + const state = makeState({ + getClient: () => ({ + exists: (_key: string, cb: (err: Error | null, count: number) => void) => + cb(new Error('connection refused'), 0), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + core.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(false); + done(); + }); +}); + +it('calls back false from isStoreAvailable when the connection is down', (done) => { + const state = makeState({ + isConnected: () => false, + isInitialConnection: () => false, + getClient: () => { + throw new Error('should not create a client while disconnected'); + }, + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + core.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(false); + done(); + }); +}); + +it('forwards isStoreAvailable through the feature store facade', (done) => { + (PersistentDataStoreWrapper as unknown as jest.Mock).mockImplementation(() => ({ + isStoreAvailable: (callback: (isAvailable: boolean) => void) => callback(true), + })); + // Provide a fake client so no real Redis connection is made. + const fakeClient = { on: jest.fn() }; + const store = new RedisFeatureStore( + // @ts-ignore Partial client mock for testing. + { client: fakeClient }, + ); + store.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(true); + done(); + }); +}); diff --git a/packages/store/node-server-sdk-redis/__tests__/RedisCoreUpsert.test.ts b/packages/store/node-server-sdk-redis/__tests__/RedisCoreUpsert.test.ts new file mode 100644 index 0000000000..addc30ed9b --- /dev/null +++ b/packages/store/node-server-sdk-redis/__tests__/RedisCoreUpsert.test.ts @@ -0,0 +1,127 @@ +import { interfaces } from '@launchdarkly/node-server-sdk'; + +import RedisCore from '../src/RedisCore'; + +const featuresKind = { namespace: 'features', deserialize: (data: string) => JSON.parse(data) }; + +function makeState(overrides: object) { + return { + prefixedKey: (key: string) => key, + isConnected: () => true, + isInitialConnection: () => false, + ...overrides, + }; +} + +beforeEach(() => { + jest.clearAllMocks(); +}); + +it('reports an error through the callback when watch rejects, with no unhandled rejection', async () => { + const unhandledRejections: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => unhandledRejections.push(reason); + process.on('unhandledRejection', onUnhandledRejection); + + const watchError = new Error('connection is closed.'); + const state = makeState({ + getClient: () => ({ + watch: () => Promise.reject(watchError), + // The get() call reads via hget; leaving it uncalled isolates the + // watch-rejection path so the callback observed below only comes from it. + hget: jest.fn(), + multi: () => ({ + discard: jest.fn(), + hset: jest.fn(), + exec: jest.fn(), + }), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + + const result = await new Promise<{ + err?: Error; + updated?: interfaces.SerializedItemDescriptor; + }>((resolve) => { + core.upsert(featuresKind, 'flagA', { version: 1, serializedItem: '{}' }, (err, updated) => { + resolve({ err, updated }); + }); + }); + + // Flush the microtask queue so an unhandled rejection, if any, would surface. + await new Promise((resolve) => { + setImmediate(resolve); + }); + process.off('unhandledRejection', onUnhandledRejection); + + expect(result.err).toBe(watchError); + expect(result.updated).toBeUndefined(); + expect(unhandledRejections).toEqual([]); +}); + +it('settles the callback exactly once when watch rejects and exec also errors', async () => { + const unhandledRejections: unknown[] = []; + const onUnhandledRejection = (reason: unknown) => unhandledRejections.push(reason); + process.on('unhandledRejection', onUnhandledRejection); + + const watchError = new Error('watch: connection is closed.'); + const execError = new Error('exec: connection is closed.'); + const state = makeState({ + getClient: () => ({ + watch: () => Promise.reject(watchError), + hget: (_ns: string, _key: string, cb: (err: Error | null, val: string | null) => void) => { + cb(null, null); + }, + multi: () => ({ + hset: jest.fn(), + discard: jest.fn(), + exec: (cb: (err: Error | null, replies: unknown) => void) => { + cb(execError, undefined); + }, + }), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + + const callback = jest.fn(); + core.upsert(featuresKind, 'flagA', { version: 1, serializedItem: '{}' }, callback); + + // Flush the microtask queue so the watch rejection's handler runs after exec's + // synchronous callback has already settled the upsert. + await new Promise((resolve) => { + setImmediate(resolve); + }); + process.off('unhandledRejection', onUnhandledRejection); + + expect(callback).toHaveBeenCalledTimes(1); + expect(callback).toHaveBeenCalledWith(execError, { version: 1, serializedItem: '{}' }); + expect(unhandledRejections).toEqual([]); +}); + +it('stores the serializedItem verbatim for a deleted descriptor', (done) => { + const hset = jest.fn(); + const serializedItem = JSON.stringify({ key: 'flagA', version: 3, deleted: true }); + const state = makeState({ + getClient: () => ({ + watch: jest.fn().mockResolvedValue('OK'), + hget: (_ns: string, _key: string, cb: (err: Error | null, val: string | null) => void) => { + cb(null, null); + }, + multi: () => ({ + hset, + discard: jest.fn(), + exec: (cb: (err: Error | null, replies: unknown) => void) => { + cb(null, ['OK']); + }, + }), + }), + }); + // @ts-ignore Partial state mock for testing. + const core = new RedisCore(state); + + core.upsert(featuresKind, 'flagA', { version: 3, deleted: true, serializedItem }, () => { + expect(hset).toHaveBeenCalledWith('features', 'flagA', serializedItem); + done(); + }); +}); diff --git a/packages/store/node-server-sdk-redis/__tests__/RedisErrorLogging.test.ts b/packages/store/node-server-sdk-redis/__tests__/RedisErrorLogging.test.ts index 9ca9ba545f..016900eccf 100644 --- a/packages/store/node-server-sdk-redis/__tests__/RedisErrorLogging.test.ts +++ b/packages/store/node-server-sdk-redis/__tests__/RedisErrorLogging.test.ts @@ -52,7 +52,7 @@ it('logs through the wrapper when a write fails', (done) => { ); const logger = makeLogger(); const fakeClient = { - watch: jest.fn(), + watch: jest.fn().mockResolvedValue('OK'), hget: (_ns: string, _key: string, cb: (err: Error | null, val: string | null) => void) => { cb(null, null); }, diff --git a/packages/store/node-server-sdk-redis/src/RedisCore.ts b/packages/store/node-server-sdk-redis/src/RedisCore.ts index 9f651d63af..98f1925e63 100644 --- a/packages/store/node-server-sdk-redis/src/RedisCore.ts +++ b/packages/store/node-server-sdk-redis/src/RedisCore.ts @@ -35,7 +35,7 @@ export default class RedisCore implements interfaces.PersistentDataStore { init( allData: interfaces.KindKeyedStore, - callback: () => void, + callback: (err?: Error) => void, ): void { const multi = this._state.getClient().multi(); allData.forEach((keyedItems) => { @@ -66,7 +66,7 @@ export default class RedisCore implements interfaces.PersistentDataStore { if (err) { this._logger?.error(`Error initializing Redis store ${err}`); } - callback(); + callback(err ?? undefined); }); } @@ -138,9 +138,29 @@ export default class RedisCore implements interfaces.PersistentDataStore { updatedDescriptor?: interfaces.SerializedItemDescriptor | undefined, ) => void, ): void { + // The watch rejection and the exec completion race independently, so the callback + // must only ever fire once. A second fire would shift the persistent store wrapper's + // update queue twice and silently drop the next queued operation. + let settled = false; + const settleOnce = (err?: Error, updatedDescriptor?: interfaces.SerializedItemDescriptor) => { + if (settled) { + return; + } + settled = true; + callback(err, updatedDescriptor); + }; + // The persistent store wrapper manages interactions with a queue, so we can use watch like // this without concerns for overlapping transactions. - this._state.getClient().watch(this._state.prefixedKey(kind.namespace)); + this._state + .getClient() + .watch(this._state.prefixedKey(kind.namespace)) + .catch((err: unknown) => { + // Without this handler a rejected watch (for example during a store outage) + // becomes an unhandled promise rejection and can crash the process. + this._logger?.error(`Error watching '${kind.namespace}' in Redis ${err}`); + settleOnce(err as Error, undefined); + }); const multi = this._state.getClient().multi(); this.get(kind, key, (old) => { @@ -153,7 +173,7 @@ export default class RedisCore implements interfaces.PersistentDataStore { if ((deserializedOld?.version || 0) >= descriptor.version) { multi.discard(); - callback(undefined, { + settleOnce(undefined, { version: deserializedOld!.version, deleted: !deserializedOld?.item, // If there is no item, then it is deleted. serializedItem: old.serializedItem, @@ -161,28 +181,33 @@ export default class RedisCore implements interfaces.PersistentDataStore { return; } } - if (descriptor.deleted) { + if (descriptor.serializedItem) { + multi.hset(this._state.prefixedKey(kind.namespace), key, descriptor.serializedItem); + } else if (descriptor.deleted) { + // The SDK contract guarantees a serializedItem is always provided for writes, + // including deletes, so this only runs if that contract is violated. It keeps + // today's placeholder shape, but adds the key so the tombstone stays identifiable. multi.hset( this._state.prefixedKey(kind.namespace), key, - JSON.stringify({ version: descriptor.version, deleted: true }), + JSON.stringify({ key, version: descriptor.version, deleted: true }), ); - } else if (descriptor.serializedItem) { - multi.hset(this._state.prefixedKey(kind.namespace), key, descriptor.serializedItem); } else { // This call violates the contract. multi.discard(); this._logger?.error('Attempt to write a non-deleted item without data to Redis.'); - callback(undefined, undefined); + settleOnce(undefined, undefined); return; } multi.exec((err, replies) => { if (!err && (replies === null || replies === undefined)) { // This means the EXEC failed because someone modified the watched key this._logger?.debug('Concurrent modification detected, retrying'); + // This is a fresh attempt with its own watch/settle guard, not a + // completion of this one, so it gets the original callback, not settleOnce. this.upsert(kind, key, descriptor, callback); } else { - callback(err || undefined, descriptor); + settleOnce(err || undefined, descriptor); } }); }); @@ -199,6 +224,20 @@ export default class RedisCore implements interfaces.PersistentDataStore { }); } + isStoreAvailable(callback: (isAvailable: boolean) => void): void { + // During the initial connection ioredis queues the command and may still connect. + // Fail fast only once a prior connection has dropped. + if (!this._state.isConnected() && !this._state.isInitialConnection()) { + callback(false); + return; + } + // A cheap read. The store is available when the command round-trip succeeds. + // The value of the key does not matter. + this._state.getClient().exists(this._initedKey, (err) => { + callback(!err); + }); + } + close(): void { this._state.close(); } diff --git a/packages/store/node-server-sdk-redis/src/RedisFeatureStore.ts b/packages/store/node-server-sdk-redis/src/RedisFeatureStore.ts index 45920e798b..20030a03e3 100644 --- a/packages/store/node-server-sdk-redis/src/RedisFeatureStore.ts +++ b/packages/store/node-server-sdk-redis/src/RedisFeatureStore.ts @@ -40,7 +40,7 @@ export default class RedisFeatureStore implements LDFeatureStore { this._wrapper.all(kind, callback); } - init(allData: LDFeatureStoreDataStorage, callback: () => void): void { + init(allData: LDFeatureStoreDataStorage, callback: (err?: Error) => void): void { this._wrapper.init(allData, callback); } @@ -48,7 +48,11 @@ export default class RedisFeatureStore implements LDFeatureStore { this._wrapper.delete(kind, key, version, callback); } - upsert(kind: interfaces.DataKind, data: LDKeyedFeatureStoreItem, callback: () => void): void { + upsert( + kind: interfaces.DataKind, + data: LDKeyedFeatureStoreItem, + callback: (err?: Error) => void, + ): void { this._wrapper.upsert(kind, data, callback); } @@ -56,6 +60,17 @@ export default class RedisFeatureStore implements LDFeatureStore { this._wrapper.initialized(callback); } + isStoreAvailable(callback: (isAvailable: boolean) => void): void { + if (this._wrapper.isStoreAvailable) { + this._wrapper.isStoreAvailable(callback); + return; + } + // No availability check on the wrapped core. Report unavailable so recovery + // falls back to the next successful write instead of a probe that can never + // report true. + callback(false); + } + close(): void { this._wrapper.close(); }