feat: conform FDv1 streaming and polling data sources to the RETRY spec

.github/workflows/server-node-nightly.yaml

@@ -0,0 +1,41 @@
+name: sdk/server-node nightly
+
+on:
+  schedule:
+    - cron: '0 3 * * *'
+  workflow_dispatch:
+
+jobs:
+  contract-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+      - uses: ./actions/setup-yarn
+        with:
+          node-version: 22
+          registry-url: 'https://registry.npmjs.org'
+      - id: shared
+        name: Shared CI Steps
+        uses: ./actions/ci
+        with:
+          workspace_name: '@launchdarkly/node-server-sdk'
+          workspace_path: packages/sdk/server-node
+      - name: Install contract test service dependencies
+        env:
+          ELECTRON_SKIP_BINARY_DOWNLOAD: '1'
+        run: yarn workspace @launchdarkly/node-server-sdk-contract-tests install --no-immutable
+      - name: Build shared contract test utils (server)
+        run: yarn workspace @launchdarkly/js-contract-test-utils build:server
+      - name: Build the test service
+        run: yarn workspace @launchdarkly/node-server-sdk-contract-tests build
+      - name: Launch the test service in the background
+        run: yarn workspace @launchdarkly/node-server-sdk-contract-tests start 2>&1 &
+      # The retry-conformance suite is long-running (real backoff waits), so it
+      # runs here nightly rather than on every PR.
+      - name: Run contract tests (long-running)
+        uses: launchdarkly/gh-actions/actions/contract-tests@contract-tests-v1
+        with:
+          test_service_port: 8000
+          token: ${{ secrets.GITHUB_TOKEN }}
+          stop_service: 'false'
+          extra_params: '--enable-long-running-tests --skip-from=./packages/sdk/server-node/contract-tests/testharness-suppressions.txt'

.yarnrc.yml

@@ -7,6 +7,12 @@ nodeLinker: node-modules
 
 npmMinimalAgeGate: 4320
 
+# Exempt this first-party package from the minimum-age gate so freshly published
+# versions install without waiting it out. Exact name (not a glob) so it cannot
+# match an attacker-squatted launchdarkly-* package.
+npmPreapprovedPackages:
+  - 'launchdarkly-eventsource'
+
 npmPublishAccess: public
 
 npmScopes:

packages/sdk/browser/src/platform/DefaultBrowserEventSource.ts

@@ -35,8 +35,8 @@ export default class DefaultBrowserEventSource implements LDEventSource {
     options: EventSourceInitDict,
   ) {
     this._backoff = new DefaultBackoff(
-      options.initialRetryDelayMillis,
-      options.retryResetIntervalMillis,
+      options.initialRetryDelayMillis ?? 1000,
+      options.retryResetIntervalMillis ?? 60 * 1000,
     );
     this._errorFilter = options.errorFilter;
     this._urlBuilder = options.urlBuilder;

packages/sdk/server-node/contract-tests/src/index.ts

@@ -24,6 +24,8 @@ app.get('/', (req: Request, res: Response) => {
     capabilities: [
       'server-side-polling',
       'server-side',
+      'retry-conformance-fdv1-streaming',
+      'retry-conformance-fdv1-polling',
       'all-flags-client-side-only',
       'all-flags-details-only-for-tracked-flags',
       'all-flags-with-reasons',

packages/sdk/server-node/package.json

@@ -49,7 +49,7 @@
   "dependencies": {
     "@launchdarkly/js-server-sdk-common": "2.21.6",
     "https-proxy-agent": "^7.0.6",
-    "launchdarkly-eventsource": "2.2.0"
+    "launchdarkly-eventsource": "2.3.0"
   },
   "devDependencies": {
     "@types/jest": "^29.4.0",

packages/shared/common/src/api/platform/EventSource.ts

@@ -17,14 +17,44 @@ export interface EventSource {
   close(): void;
 }
 
+/**
+ * The strategy an {@link EventSource} uses to decide how long to wait before
+ * each reconnection attempt. It is injected so a data source can drive
+ * reconnection timing from its own retry state rather than the transport's
+ * built-in backoff. The `nowMs` arguments are the transport's wall clock; an
+ * implementation that keeps its own clock may ignore them.
+ */
+export interface EventSourceRetryDelayStrategy {
+  /** Returns the delay, in milliseconds, before the next reconnection. */
+  nextRetryDelay(nowMs: number): number;
+  /** Records that the connection is currently healthy. */
+  setGoodSince(nowMs: number): void;
+  /** Applies a server-directed base delay, in milliseconds. */
+  setBaseDelay(baseDelayMs: number): void;
+}
+
 export interface EventSourceInitDict {
   method?: string;
   headers: { [key: string]: string | string[] };
   body?: string;
   errorFilter: (err: HttpErrorResponse) => boolean;
-  initialRetryDelayMillis: number;
   readTimeoutMillis: number;
-  retryResetIntervalMillis: number;
+
+  /**
+   * The initial delay and reset window for the built-in (default) retry-delay
+   * strategy. These configure the default behavior; they are ignored when a
+   * custom {@link retryDelayStrategy} is provided, since that strategy then
+   * owns all reconnection timing.
+   */
+  initialRetryDelayMillis?: number;
+  retryResetIntervalMillis?: number;
+
+  /**
+   * A custom strategy that replaces the built-in (default) one. When provided,
+   * the EventSource defers all reconnection timing to it and the built-in
+   * options (`initialRetryDelayMillis`, `retryResetIntervalMillis`) are ignored.
+   */
+  retryDelayStrategy?: EventSourceRetryDelayStrategy;
   /**
    * Optional callback that returns a fresh URL on each reconnection attempt.
    * When provided, the EventSource implementation should call this instead of

packages/shared/sdk-server/__tests__/data_sources/PollingProcessor.test.ts

@@ -164,7 +164,7 @@ describe('given a polling processor with a short poll duration', () => {
     processor.start();
 
     expect(initSuccessHandler).not.toBeCalled();
-    expect(errorHandler.mock.lastCall[0].message).toMatch(/malformed json/i);
+    expect(errorHandler).not.toBeCalled();
 
     setTimeout(() => {
       expect(requestor.requestAllData.mock.calls.length).toBeGreaterThanOrEqual(2);
@@ -173,26 +173,77 @@ describe('given a polling processor with a short poll duration', () => {
     }, 300);
   });
 
-  it.each<number | jest.DoneCallback>([401, 403])(
-    'does not continue after non-recoverable error',
-    (status, done) => {
-      requestor.requestAllData = jest.fn((cb) =>
-        cb(
-          {
-            status,
-          },
-          undefined,
-        ),
-      );
+  it('continues polling when deserialization throws on structurally invalid data', (done) => {
+    // Parses as JSON but throws during the revive step (here, a null flag
+    // entry). It must be handled like unparseable data, not escape and kill
+    // the poll loop.
+    requestor.requestAllData = jest.fn((cb) => cb(undefined, '{"flags":{"x":null},"segments":{}}'));
+
+    processor.start();
+
+    expect(initSuccessHandler).not.toBeCalled();
+    expect(errorHandler).not.toBeCalled();
+
+    setTimeout(() => {
+      expect(requestor.requestAllData.mock.calls.length).toBeGreaterThanOrEqual(2);
+      expect(testLogger.getCount(LogLevel.Error)).toBeGreaterThan(2);
+      (done as jest.DoneCallback)();
+    }, 300);
+  });
+
+  it('cancels the scheduled poll when stopped before it fires', () => {
+    jest.useFakeTimers();
+    try {
+      requestor.requestAllData = jest.fn((cb) => cb({ status: 500 }, undefined));
       processor.start();
-      expect(initSuccessHandler).not.toBeCalled();
-      expect(errorHandler.mock.lastCall[0].message).toMatch(new RegExp(`${status}.*permanently`));
+      expect(requestor.requestAllData).toHaveBeenCalledTimes(1);
 
-      setTimeout(() => {
-        expect(requestor.requestAllData.mock.calls.length).toBe(1);
+      // The failed poll armed the next-poll timer; stopping must cancel it.
+      processor.stop();
+      jest.advanceTimersByTime(5 * 60 * 1000);
+
+      expect(requestor.requestAllData).toHaveBeenCalledTimes(1);
+    } finally {
+      jest.useRealTimers();
+    }
+  });
+
+  it('treats a status-less transport error as a normal, retryable failure', (done) => {
+    requestor.requestAllData = jest.fn((cb) => cb({ message: 'socket hang up' } as any, undefined));
+
+    processor.start();
+
+    expect(errorHandler).not.toBeCalled();
+    setTimeout(() => {
+      expect(requestor.requestAllData.mock.calls.length).toBeGreaterThanOrEqual(2);
+      expect(testLogger.getCount(LogLevel.Error)).toBe(0);
+      expect(testLogger.getCount(LogLevel.Warn)).toBeGreaterThan(2);
+      (done as jest.DoneCallback)();
+    }, 300);
+  });
+
+  it.each([401, 403])(
+    'retries with extended backoff rather than stopping after error %p',
+    (status) => {
+      jest.useFakeTimers();
+      try {
+        requestor.requestAllData = jest.fn((cb) => cb({ status }, undefined));
+        processor.start();
+
+        expect(initSuccessHandler).not.toBeCalled();
+        expect(requestor.requestAllData).toHaveBeenCalledTimes(1);
+        // Previously terminal; now an 'error'-level *log* that keeps retrying —
+        // like any recoverable failure, it is not surfaced as an 'error' event.
+        expect(errorHandler).not.toBeCalled();
         expect(testLogger.getCount(LogLevel.Error)).toBe(1);
-        (done as jest.DoneCallback)();
-      }, 300);
+
+        // Not a permanent stop: an unexpected failure schedules the next poll in
+        // the extended regime (~5 minutes), so polling resumes after that wait.
+        jest.advanceTimersByTime(5 * 60 * 1000);
+        expect(requestor.requestAllData.mock.calls.length).toBeGreaterThanOrEqual(2);
+      } finally {
+        jest.useRealTimers();
+      }
     },
   );
 });