From 23faebce11567ec14cef94e581ceafdaa3ff8bfc Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Fri, 11 Sep 2026 15:17:08 -0400 Subject: [PATCH 1/2] feat: Report DynamoDB store errors and implement the availability check --- .../DynamoDBCoreAvailability.test.ts | 122 ++++++++++++++++++ .../src/DynamoDBCore.ts | 26 +++- .../src/DynamoDBFeatureStore.ts | 19 ++- 3 files changed, 163 insertions(+), 4 deletions(-) create mode 100644 packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts diff --git a/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts new file mode 100644 index 0000000000..549403a254 --- /dev/null +++ b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts @@ -0,0 +1,122 @@ +import { interfaces, PersistentDataStoreWrapper } from '@launchdarkly/node-server-sdk'; + +import DynamoDBCore from '../src/DynamoDBCore'; +import DynamoDBFeatureStore from '../src/DynamoDBFeatureStore'; + +jest.mock('@launchdarkly/node-server-sdk', () => { + const actual = jest.requireActual('@launchdarkly/node-server-sdk'); + return { + ...actual, + PersistentDataStoreWrapper: jest.fn(), + }; +}); + +beforeEach(() => { + jest.clearAllMocks(); +}); + +const featuresKind: interfaces.PersistentStoreDataKind = { + namespace: 'features', + deserialize: (data: string) => JSON.parse(data), +}; + +const allData: interfaces.KindKeyedStore = [ + { + key: featuresKind, + item: [{ key: 'flagA', item: { version: 1, deleted: false, serializedItem: '{"version":1}' } }], + }, +]; + +it('reports an init error through the callback when the batch write fails', (done) => { + const state = { + prefixedKey: (key: string) => key, + query: jest.fn().mockResolvedValue([]), + batchWrite: jest.fn().mockRejectedValue(new Error('write failed')), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + core.init(allData, (err) => { + expect(err).toEqual(new Error('write failed')); + done(); + }); +}); + +it('reports an init error through the callback when reading existing items fails', (done) => { + const state = { + prefixedKey: (key: string) => key, + query: jest.fn().mockRejectedValue(new Error('read failed')), + batchWrite: jest.fn(), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + core.init(allData, (err) => { + expect(err).toEqual(new Error('read failed')); + expect(state.batchWrite).not.toHaveBeenCalled(); + done(); + }); +}); + +it('calls back without an error when init succeeds', (done) => { + const state = { + prefixedKey: (key: string) => key, + query: jest.fn().mockResolvedValue([]), + batchWrite: jest.fn().mockResolvedValue(undefined), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + core.init(allData, (err) => { + expect(err).toBeUndefined(); + done(); + }); +}); + +it('calls back true from isStoreAvailable when the read succeeds', (done) => { + const state = { + prefixedKey: (key: string) => key, + get: jest.fn().mockResolvedValue(undefined), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + core.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(true); + done(); + }); +}); + +it('calls back false from isStoreAvailable when the read fails', (done) => { + const state = { + prefixedKey: (key: string) => key, + get: jest.fn().mockRejectedValue(new Error('connection failed')), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + core.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(false); + done(); + }); +}); + +it('does not reject its returned promise when the isStoreAvailable callback throws', async () => { + const state = { + prefixedKey: (key: string) => key, + get: jest.fn().mockResolvedValue(undefined), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + await expect( + core.isStoreAvailable(() => { + throw new Error('callback exploded'); + }), + ).resolves.toBeUndefined(); +}); + +it('forwards isStoreAvailable through the feature store facade', (done) => { + (PersistentDataStoreWrapper as unknown as jest.Mock).mockImplementation(() => ({ + isStoreAvailable: (callback: (isAvailable: boolean) => void) => callback(true), + })); + const store = new DynamoDBFeatureStore('test-table'); + store.isStoreAvailable((isAvailable) => { + expect(isAvailable).toBe(true); + done(); + }); +}); diff --git a/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts b/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts index 2d1123f623..ce9d0d52e9 100644 --- a/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts +++ b/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts @@ -124,14 +124,14 @@ export default class DynamoDBCore implements interfaces.PersistentDataStore { async init( allData: interfaces.KindKeyedStore, - callback: () => void, + callback: (err?: Error) => void, ) { let items: Record[]; try { items = await this._readExistingItems(allData); } catch (error) { this._logger?.error(`Error reading existing items from DynamoDB: ${error}`); - callback(); + callback(error as Error); return; } @@ -178,6 +178,8 @@ export default class DynamoDBCore implements interfaces.PersistentDataStore { await this._state.batchWrite(this._tableName, ops); } catch (error) { this._logger?.error(`Error writing to DynamoDB: ${error}`); + callback(error as Error); + return; } callback(); } @@ -267,6 +269,26 @@ export default class DynamoDBCore implements interfaces.PersistentDataStore { callback(initialized); } + async isStoreAvailable(callback: (isAvailable: boolean) => void) { + let isAvailable = false; + try { + // A cheap read. The store is available when the request succeeds. The result + // value does not matter. + await this._state.get(this._tableName, this._initializedToken()); + isAvailable = true; + } catch { + isAvailable = false; + } + // Callback outside the try for the read above, so a failed read is never + // mistaken for a callback error. It gets its own try/catch so a throw from the + // caller's callback cannot reject this method's returned promise. + try { + callback(isAvailable); + } catch { + // The caller's callback is responsible for handling its own errors. + } + } + close(): void { this._state.close(); } diff --git a/packages/store/node-server-sdk-dynamodb/src/DynamoDBFeatureStore.ts b/packages/store/node-server-sdk-dynamodb/src/DynamoDBFeatureStore.ts index b26c781cdf..fd74f12e7c 100644 --- a/packages/store/node-server-sdk-dynamodb/src/DynamoDBFeatureStore.ts +++ b/packages/store/node-server-sdk-dynamodb/src/DynamoDBFeatureStore.ts @@ -44,7 +44,7 @@ export default class DynamoDBFeatureStore 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); } @@ -52,7 +52,11 @@ export default class DynamoDBFeatureStore 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); } @@ -60,6 +64,17 @@ export default class DynamoDBFeatureStore 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(); } From 9a4c68ac95e62ce70811599a98a86431202aba56 Mon Sep 17 00:00:00 2001 From: Steven Zhang Date: Wed, 23 Sep 2026 17:21:03 -0400 Subject: [PATCH 2/2] fix: Retry unprocessed DynamoDB batch items and write the initialized token after the data --- .../DynamoDBClientStateBatchWrite.test.ts | 118 ++++++++++++++++++ .../__tests__/DynamoDBCore.test.ts | 79 ++++++++++++ .../DynamoDBCoreAvailability.test.ts | 20 +++ .../src/DynamoDBClientState.ts | 47 ++++++- .../src/DynamoDBCore.ts | 8 +- 5 files changed, 266 insertions(+), 6 deletions(-) create mode 100644 packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBClientStateBatchWrite.test.ts diff --git a/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBClientStateBatchWrite.test.ts b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBClientStateBatchWrite.test.ts new file mode 100644 index 0000000000..136acd53cd --- /dev/null +++ b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBClientStateBatchWrite.test.ts @@ -0,0 +1,118 @@ +import { DynamoDBClient, WriteRequest } from '@aws-sdk/client-dynamodb'; + +import DynamoDBClientState from '../src/DynamoDBClientState'; + +const TABLE_NAME = 'test-table'; + +function makeWriteRequest(key: string): WriteRequest { + return { + PutRequest: { + Item: { + namespace: { S: 'features' }, + key: { S: key }, + }, + }, + }; +} + +function makeState(send: jest.Mock): DynamoDBClientState { + // @ts-ignore Partial client mock for testing. + const client = { send, destroy: jest.fn() } as DynamoDBClient; + return new DynamoDBClientState({ dynamoDBClient: client }); +} + +afterEach(() => { + jest.useRealTimers(); +}); + +it('does not retry when the response contains no unprocessed items', async () => { + const send = jest.fn().mockResolvedValue({}); + const state = makeState(send); + + await state.batchWrite(TABLE_NAME, [makeWriteRequest('flagA'), makeWriteRequest('flagB')]); + + expect(send).toHaveBeenCalledTimes(1); +}); + +it('treats an empty unprocessed map as success', async () => { + const send = jest.fn().mockResolvedValue({ UnprocessedItems: {} }); + const state = makeState(send); + + await state.batchWrite(TABLE_NAME, [makeWriteRequest('flagA')]); + + expect(send).toHaveBeenCalledTimes(1); +}); + +it('retries only the unprocessed items until they succeed', async () => { + jest.useFakeTimers(); + const requestA = makeWriteRequest('flagA'); + const requestB = makeWriteRequest('flagB'); + const send = jest + .fn() + .mockResolvedValueOnce({ UnprocessedItems: { [TABLE_NAME]: [requestB] } }) + .mockResolvedValue({}); + const state = makeState(send); + + const pendingWrite = state.batchWrite(TABLE_NAME, [requestA, requestB]); + await jest.runAllTimersAsync(); + await pendingWrite; + + expect(send).toHaveBeenCalledTimes(2); + expect(send.mock.calls[1][0].input.RequestItems[TABLE_NAME]).toEqual([requestB]); +}); + +it('throws when items remain unprocessed after the retries are exhausted', async () => { + jest.useFakeTimers(); + const requestA = makeWriteRequest('flagA'); + const send = jest.fn().mockResolvedValue({ UnprocessedItems: { [TABLE_NAME]: [requestA] } }); + const state = makeState(send); + + const pendingWrite = state.batchWrite(TABLE_NAME, [requestA]); + // The assertion is awaited after the fake timers run. + // eslint-disable-next-line jest/valid-expect + const assertion = expect(pendingWrite).rejects.toThrow( + 'DynamoDB batch write returned 1 unprocessed item(s) after 3 retries', + ); + await jest.runAllTimersAsync(); + await assertion; + + expect(send).toHaveBeenCalledTimes(4); +}); + +it('waits with exponential backoff before each retry', async () => { + jest.useFakeTimers(); + const requestA = makeWriteRequest('flagA'); + const send = jest.fn().mockResolvedValue({ UnprocessedItems: { [TABLE_NAME]: [requestA] } }); + const state = makeState(send); + + const pendingWrite = state.batchWrite(TABLE_NAME, [requestA]); + // The assertion is awaited after the fake timers run. + // eslint-disable-next-line jest/valid-expect + const assertion = expect(pendingWrite).rejects.toThrow( + 'DynamoDB batch write returned 1 unprocessed item(s) after 3 retries', + ); + + // The first attempt does not wait. + await jest.advanceTimersByTimeAsync(0); + expect(send).toHaveBeenCalledTimes(1); + + // The first retry waits 100 milliseconds. + await jest.advanceTimersByTimeAsync(99); + expect(send).toHaveBeenCalledTimes(1); + await jest.advanceTimersByTimeAsync(1); + expect(send).toHaveBeenCalledTimes(2); + + // The second retry waits 200 milliseconds. + await jest.advanceTimersByTimeAsync(199); + expect(send).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(1); + expect(send).toHaveBeenCalledTimes(3); + + // The third retry waits 400 milliseconds. + await jest.advanceTimersByTimeAsync(399); + expect(send).toHaveBeenCalledTimes(3); + await jest.advanceTimersByTimeAsync(1); + expect(send).toHaveBeenCalledTimes(4); + + await assertion; +}); diff --git a/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCore.test.ts b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCore.test.ts index 0a7189db06..c6f2c0d091 100644 --- a/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCore.test.ts +++ b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCore.test.ts @@ -170,6 +170,85 @@ describe('given an empty store', () => { }, ]); }); + + it('does not write the initialized token when the batch write fails', async () => { + const mockLogger = { + error: jest.fn(), + warn: jest.fn(), + info: jest.fn(), + debug: jest.fn(), + }; + const state = new DynamoDBClientState(DEFAULT_CLIENT_OPTIONS); + const errorCore = new DynamoDBCore(DEFAULT_TABLE_NAME, state, mockLogger); + const errorFacade = new AsyncCoreFacade(errorCore); + + const putSpy = jest.spyOn(state, 'put'); + const error = new Error('write failed'); + jest.spyOn(state, 'batchWrite').mockRejectedValueOnce(error); + + await errorFacade.init([]); + + // The token write must not happen when the data batch fails. + expect(putSpy).not.toHaveBeenCalled(); + expect(mockLogger.error).toHaveBeenCalledWith(`Error writing to DynamoDB: ${error}`); + + errorCore.close(); + }); + + it('writes the initialized token only after the data batch succeeds', async () => { + const flags = [ + { key: 'first', item: { version: 1, serializedItem: `{"version":1}`, deleted: false } }, + ]; + + const state = new DynamoDBClientState(DEFAULT_CLIENT_OPTIONS); + const successCore = new DynamoDBCore(DEFAULT_TABLE_NAME, state, undefined); + const successFacade = new AsyncCoreFacade(successCore); + + const batchWriteSpy = jest.spyOn(state, 'batchWrite'); + const putSpy = jest.spyOn(state, 'put'); + + await successFacade.init([{ key: dataKind.features, item: flags }]); + + // The data batch must not include the token. + const [, ops] = batchWriteSpy.mock.calls[0]; + expect( + ops.some( + (op) => + op.PutRequest?.Item?.namespace.S === '$inited' && + op.PutRequest?.Item?.key.S === '$inited', + ), + ).toBe(false); + + // The token is written on its own, after the data batch succeeds. + expect(putSpy).toHaveBeenCalledWith({ + TableName: DEFAULT_TABLE_NAME, + Item: { namespace: { S: '$inited' }, key: { S: '$inited' } }, + }); + expect(batchWriteSpy.mock.invocationCallOrder[0]).toBeLessThan( + putSpy.mock.invocationCallOrder[0], + ); + + successCore.close(); + }); + + it('writes the initialized token even when there is no data to batch', async () => { + const state = new DynamoDBClientState(DEFAULT_CLIENT_OPTIONS); + const emptyCore = new DynamoDBCore(DEFAULT_TABLE_NAME, state, undefined); + const emptyFacade = new AsyncCoreFacade(emptyCore); + + const batchWriteSpy = jest.spyOn(state, 'batchWrite'); + const putSpy = jest.spyOn(state, 'put'); + + await emptyFacade.init([]); + + expect(batchWriteSpy).toHaveBeenCalledWith(DEFAULT_TABLE_NAME, []); + expect(putSpy).toHaveBeenCalledWith({ + TableName: DEFAULT_TABLE_NAME, + Item: { namespace: { S: '$inited' }, key: { S: '$inited' } }, + }); + + emptyCore.close(); + }); }); describe('given a store with basic data', () => { diff --git a/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts index 549403a254..b33610eb67 100644 --- a/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts +++ b/packages/store/node-server-sdk-dynamodb/__tests__/DynamoDBCoreAvailability.test.ts @@ -61,11 +61,31 @@ it('calls back without an error when init succeeds', (done) => { prefixedKey: (key: string) => key, query: jest.fn().mockResolvedValue([]), batchWrite: jest.fn().mockResolvedValue(undefined), + put: jest.fn().mockResolvedValue(undefined), }; // @ts-ignore Partial state mock for testing. const core = new DynamoDBCore('test-table', state); core.init(allData, (err) => { expect(err).toBeUndefined(); + expect(state.put).toHaveBeenCalledWith({ + TableName: 'test-table', + Item: { namespace: { S: '$inited' }, key: { S: '$inited' } }, + }); + done(); + }); +}); + +it('reports an init error through the callback when writing the initialized token fails', (done) => { + const state = { + prefixedKey: (key: string) => key, + query: jest.fn().mockResolvedValue([]), + batchWrite: jest.fn().mockResolvedValue(undefined), + put: jest.fn().mockRejectedValue(new Error('token write failed')), + }; + // @ts-ignore Partial state mock for testing. + const core = new DynamoDBCore('test-table', state); + core.init(allData, (err) => { + expect(err).toEqual(new Error('token write failed')); done(); }); }); diff --git a/packages/store/node-server-sdk-dynamodb/src/DynamoDBClientState.ts b/packages/store/node-server-sdk-dynamodb/src/DynamoDBClientState.ts index 33e7d40fc3..dba9a795e1 100644 --- a/packages/store/node-server-sdk-dynamodb/src/DynamoDBClientState.ts +++ b/packages/store/node-server-sdk-dynamodb/src/DynamoDBClientState.ts @@ -21,6 +21,18 @@ const DEFAULT_PREFIX = ''; // BatchWrite can only accept 25 items at a time, so split up the writes into batches of 25. const WRITE_BATCH_SIZE = 25; +// DynamoDB can return unprocessed items when it throttles a batch write. +// Retry the unprocessed items a limited number of times with exponential +// backoff. Report a failure if items remain unprocessed after the retries. +const MAX_UNPROCESSED_RETRIES = 3; +const UNPROCESSED_RETRY_BASE_DELAY_MS = 100; + +function sleep(delayMs: number): Promise { + return new Promise((resolve) => { + setTimeout(resolve, delayMs); + }); +} + /** * Class for managing the state of a dynamodb client. * @@ -73,14 +85,33 @@ export default class DynamoDBClientState { } async batchWrite(table: string, params: WriteRequest[]) { + let pending = params; + // The first attempt writes all the items. Each retry writes only the + // items that DynamoDB returned as unprocessed. + for (let attempt = 0; attempt <= MAX_UNPROCESSED_RETRIES && pending.length > 0; attempt += 1) { + if (attempt > 0) { + // eslint-disable-next-line no-await-in-loop + await sleep(UNPROCESSED_RETRY_BASE_DELAY_MS * 2 ** (attempt - 1)); + } + // eslint-disable-next-line no-await-in-loop + pending = await this._writeBatches(table, pending); + } + if (pending.length > 0) { + throw new Error( + `DynamoDB batch write returned ${pending.length} unprocessed item(s) after ${MAX_UNPROCESSED_RETRIES} retries`, + ); + } + } + + private async _writeBatches(table: string, requests: WriteRequest[]): Promise { const batches: WriteRequest[][] = []; // Split into batches of at most 25 commands. - for (let i = 0; i < params.length; i += WRITE_BATCH_SIZE) { - batches.push(params.slice(i, i + WRITE_BATCH_SIZE)); + for (let i = 0; i < requests.length; i += WRITE_BATCH_SIZE) { + batches.push(requests.slice(i, i + WRITE_BATCH_SIZE)); } // Execute all the batches and wait for them to complete. - await Promise.all( + const results = await Promise.all( batches.map((batch) => this._client.send( new BatchWriteItemCommand({ @@ -89,6 +120,16 @@ export default class DynamoDBClientState { ), ), ); + + // Collect the items that DynamoDB did not process. + const unprocessed: WriteRequest[] = []; + results.forEach((result) => { + const items = result.UnprocessedItems?.[table]; + if (items) { + unprocessed.push(...items); + } + }); + return unprocessed; } async get( diff --git a/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts b/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts index ce9d0d52e9..8bbd474553 100644 --- a/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts +++ b/packages/store/node-server-sdk-dynamodb/src/DynamoDBCore.ts @@ -171,11 +171,13 @@ export default class DynamoDBCore implements interfaces.PersistentDataStore { }); }); - // Always write the initialized token when we initialize. - ops.push({ PutRequest: { Item: this._initializedToken() } }); - try { await this._state.batchWrite(this._tableName, ops); + // Write the initialized token on its own, after the data batch + // succeeds. A batch write is not atomic, so writing the token as + // part of the batch could leave it durably set while data items + // are still unprocessed. + await this._state.put({ TableName: this._tableName, Item: this._initializedToken() }); } catch (error) { this._logger?.error(`Error writing to DynamoDB: ${error}`); callback(error as Error);