feat: Report Redis store errors and implement the availability check

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();
+  });
+});

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();
+  });
+});

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);
     },

packages/store/node-server-sdk-redis/src/RedisCore.ts

@@ -35,7 +35,7 @@ export default class RedisCore implements interfaces.PersistentDataStore {
 
   init(
     allData: interfaces.KindKeyedStore<interfaces.PersistentStoreDataKind>,
-    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,36 +173,41 @@ 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,
           });
           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();
   }