fix: stream stability when connecting (wait4Ready) (#1458)

Signed-off-by: Todd Baert <todd.baert@dynatrace.com>

Author: toddbaert

libs/providers/flagd/src/e2e/step-definitions/providerSteps.ts

@@ -1,4 +1,4 @@
-import { OpenFeature, ProviderEvents, ProviderStatus } from '@openfeature/server-sdk';
+import { OpenFeature, ProviderStatus } from '@openfeature/server-sdk';
 import { FlagdComposeContainer } from '../tests/flagdComposeContainer';
 import type { State, Steps } from './state';
 import { FlagdProvider } from '../../lib/flagd-provider';
@@ -9,7 +9,7 @@ import { existsSync } from 'node:fs';
 
 export const providerSteps: Steps =
   (state: State) =>
-  ({ given, when, then, and }) => {
+  ({ given, when, then }) => {
     const container: FlagdComposeContainer = FlagdComposeContainer.build();
     beforeAll(async () => {
       console.log('Setting test harness...');
@@ -21,8 +21,8 @@ export const providerSteps: Steps =
       await container.stop();
     });
 
-    beforeEach(() => {
-      OpenFeature.clearProviders();
+    beforeEach(async () => {
+      await OpenFeature.clearProviders();
     });
 
     afterEach(async () => {
@@ -34,10 +34,11 @@ export const providerSteps: Steps =
       const flagdOptions: FlagdProviderOptions = {
         resolverType: state.resolverType,
         retryGracePeriod: 2, // retryGracePeriod is related to test expectations; this must be 2
-        // these 3 options are optimized for test speed and stability
-        deadlineMs: 4000,
+        // these options are optimized for test speed and stability
+        deadlineMs: 15000,
+        keepAliveTime: 200,
         retryBackoffMaxMs: 1000,
-        retryBackoffMs: 1000,
+        retryBackoffMs: 100,
 
         ...state.config,
         ...state.options,
@@ -77,7 +78,6 @@ export const providerSteps: Steps =
       }
 
       await fetch('http://' + container.getLaunchpadUrl() + '/start?config=' + type);
-      await new Promise((r) => setTimeout(r, 100));
       if (providerType == 'unavailable' || providerType == 'forbidden') {
         OpenFeature.setProvider(providerType, new FlagdProvider(flagdOptions));
       } else {

libs/providers/flagd/src/e2e/tests/in-process.spec.ts

@@ -10,6 +10,8 @@ import { contextSteps } from '../step-definitions/contextSteps';
 const steps = [providerSteps, configSteps, eventSteps, flagSteps, contextSteps];
 
 jest.setTimeout(50000);
+jest.retryTimes(3);
+
 describe('in-process', () => {
   const state: State = {
     resolverType: 'in-process',

libs/providers/flagd/src/e2e/tests/rpc.spec.ts

@@ -10,6 +10,7 @@ import { GHERKIN_FLAGD } from '../constants';
 const steps = [providerSteps, configSteps, eventSteps, flagSteps, contextSteps];
 
 jest.setTimeout(50000);
+jest.retryTimes(3);
 
 describe('rpc', () => {
   const state: State = {

libs/providers/flagd/src/lib/configuration.spec.ts

@@ -1,6 +1,6 @@
 import type { Config, FlagdProviderOptions } from './configuration';
 import { getConfig } from './configuration';
-import { DEFAULT_MAX_CACHE_SIZE, DEFAULT_RETRY_GRACE_PERIOD } from './constants';
+import { DEFAULT_MAX_BACKOFF_MS, DEFAULT_MAX_CACHE_SIZE, DEFAULT_RETRY_GRACE_PERIOD } from './constants';
 import type { EvaluationContext } from '@openfeature/server-sdk';
 import { configSteps } from '../e2e/step-definitions/configSteps';
 import type { State } from '../e2e/step-definitions/state';
@@ -23,7 +23,7 @@ describe('Configuration', () => {
       maxCacheSize: DEFAULT_MAX_CACHE_SIZE,
       cache: 'lru',
       resolverType: 'rpc',
-      retryBackoffMaxMs: 120000,
+      retryBackoffMaxMs: DEFAULT_MAX_BACKOFF_MS,
       retryBackoffMs: 1000,
       selector: '',
       deadlineMs: 500,
@@ -124,7 +124,7 @@ describe('Configuration', () => {
       maxCacheSize: 1000,
       cache: 'lru',
       resolverType: 'rpc',
-      retryBackoffMaxMs: 120000,
+      retryBackoffMaxMs: DEFAULT_MAX_BACKOFF_MS,
       retryBackoffMs: 1000,
       selector: '',
       defaultAuthority: '',

libs/providers/flagd/src/lib/configuration.ts

@@ -1,4 +1,4 @@
-import { DEFAULT_MAX_CACHE_SIZE, DEFAULT_RETRY_GRACE_PERIOD } from './constants';
+import { DEFAULT_MAX_BACKOFF_MS, DEFAULT_MAX_CACHE_SIZE, DEFAULT_RETRY_GRACE_PERIOD } from './constants';
 import type { EvaluationContext } from '@openfeature/server-sdk';
 
 export type CacheOption = 'lru' | 'disabled';
@@ -158,7 +158,7 @@ const DEFAULT_CONFIG: Omit<FlagdConfig & FlagdGrpcConfig, 'port' | 'resolverType
   maxCacheSize: DEFAULT_MAX_CACHE_SIZE,
   contextEnricher: (syncContext: EvaluationContext | null) => syncContext ?? {},
   retryBackoffMs: 1000,
-  retryBackoffMaxMs: 120000,
+  retryBackoffMaxMs: DEFAULT_MAX_BACKOFF_MS,
   keepAliveTime: 0,
   retryGracePeriod: DEFAULT_RETRY_GRACE_PERIOD,
 };

libs/providers/flagd/src/lib/constants.ts

@@ -4,3 +4,4 @@ export const EVENT_CONFIGURATION_CHANGE = 'configuration_change';
 export const EVENT_PROVIDER_READY = 'provider_ready';
 export const DEFAULT_MAX_CACHE_SIZE = 1000;
 export const DEFAULT_RETRY_GRACE_PERIOD = 5;
+export const DEFAULT_MAX_BACKOFF_MS = 12000;

libs/providers/flagd/src/lib/flagd-provider.spec.ts

@@ -67,6 +67,9 @@ describe(FlagdProvider.name, () => {
 
     // mock ServiceClient to inject
     const basicServiceClientMock: ServiceClient = {
+      waitForReady: jest.fn((_deadline: number, callback: (err?: Error) => void) => {
+        callback();
+      }),
       eventStream: jest.fn(() => {
         return {
           on: jest.fn((event: string, callback: (message: unknown) => void) => {
@@ -270,6 +273,9 @@ describe(FlagdProvider.name, () => {
 
     // mock ServiceClient to inject
     const streamingServiceClientMock = {
+      waitForReady: jest.fn((_deadline: number, callback: (err?: Error) => void) => {
+        callback();
+      }),
       eventStream: jest.fn(() => {
         return streamMock;
       }),
@@ -507,7 +513,15 @@ describe(FlagdProvider.name, () => {
     });
 
     describe('connection/re-connection', () => {
-      it('should watch channel for reconnect after error', () => {
+      beforeEach(() => {
+        jest.useFakeTimers();
+      });
+
+      afterEach(() => {
+        jest.useRealTimers();
+      });
+
+      it('should attempt to reconnect after error', () => {
         const provider = new FlagdProvider(
           undefined,
           undefined,
@@ -523,8 +537,11 @@ describe(FlagdProvider.name, () => {
         // fake some errors
         registeredOnErrorCallback();
 
-        expect(streamingServiceClientMock.getChannel().getConnectivityState).toHaveBeenCalledWith(true);
-        expect(streamingServiceClientMock.getChannel().watchConnectivityState).toHaveBeenCalled();
+        // verify reconnection is scheduled via setTimeout
+        jest.runAllTimers();
+
+        // eventStream should have been called twice (initial + reconnect attempt)
+        expect(streamingServiceClientMock.eventStream).toHaveBeenCalledTimes(2);
       });
     });
   });
@@ -535,6 +552,9 @@ describe(FlagdProvider.name, () => {
 
     // mock ServiceClient to inject
     const errServiceClientMock: ServiceClient = {
+      waitForReady: jest.fn((_deadline: number, callback: (err?: Error) => void) => {
+        callback();
+      }),
       eventStream: jest.fn(() => {
         return {
           on: jest.fn((event: string, callback: (message: unknown) => void) => {
@@ -661,6 +681,9 @@ describe(FlagdProvider.name, () => {
 
     // mock ServiceClient to inject
     const errServiceClientMock: ServiceClient = {
+      waitForReady: jest.fn((_deadline: number, callback: (err?: Error) => void) => {
+        callback();
+      }),
       eventStream: jest.fn(() => {
         return {
           on: jest.fn((event: string, callback: (message: unknown) => void) => {

libs/providers/flagd/src/lib/service/common/grpc-util.spec.ts

@@ -60,7 +60,29 @@ describe('buildClientOptions', () => {
     expect(buildClientOptions(config)).toEqual({
       'grpc.default_authority': 'my-authority',
       'grpc.keepalive_time_ms': 5000,
-      'grpc.service_config': buildRetryPolicy('flagd.service.v1.FlagService', 100, 200),
+      'grpc.service_config': buildRetryPolicy(
+        ['flagd.evaluation.v1.Service', 'flagd.sync.v1.FlagSyncService'],
+        100,
+        200,
+      ),
+    });
+  });
+});
+
+describe('buildRetryPolicy', () => {
+  it('should create a single methodConfig with multiple services sharing one retryPolicy', () => {
+    const result = JSON.parse(buildRetryPolicy(['service.A', 'service.B'], 2000, 60000));
+
+    expect(result.methodConfig).toHaveLength(1);
+    expect(result.methodConfig[0]).toEqual({
+      name: [{ service: 'service.A' }, { service: 'service.B' }],
+      retryPolicy: {
+        maxAttempts: 3,
+        initialBackoff: '2.00s',
+        maxBackoff: '60.00s',
+        backoffMultiplier: 2,
+        retryableStatusCodes: ['UNAVAILABLE', 'UNKNOWN'],
+      },
     });
   });
 });

libs/providers/flagd/src/lib/service/common/grpc-util.ts

@@ -58,7 +58,7 @@ const CONFIG_TO_GRPC_OPTIONS: {
 export function buildClientOptions(config: Config): ClientOptions {
   const options: Partial<ClientOptions> = {
     'grpc.service_config': buildRetryPolicy(
-      'flagd.service.v1.FlagService',
+      ['flagd.evaluation.v1.Service', 'flagd.sync.v1.FlagSyncService'],
       config.retryBackoffMs,
       config.retryBackoffMaxMs,
     ),
@@ -76,27 +76,33 @@ export function buildClientOptions(config: Config): ClientOptions {
 
 /**
  * Builds RetryPolicy for gRPC client options.
- * @param serviceName
+ * @param serviceNames Array of service names to configure retry policy for
  * @param retryBackoffMs Initial backoff duration in milliseconds
  * @param retryBackoffMaxMs Maximum backoff duration in milliseconds
  * @returns gRPC client options with retry policy
  */
-export const buildRetryPolicy = (serviceName: string, retryBackoffMs?: number, retryBackoffMaxMs?: number): string => {
+export const buildRetryPolicy = (
+  serviceNames: string[],
+  retryBackoffMs?: number,
+  retryBackoffMaxMs?: number,
+): string => {
   const initialBackoff = retryBackoffMs ?? 1000;
   const maxBackoff = retryBackoffMaxMs ?? 120000;
 
+  const retryPolicy = {
+    maxAttempts: 3,
+    initialBackoff: `${Math.round(initialBackoff / 1000).toFixed(2)}s`,
+    maxBackoff: `${Math.round(maxBackoff / 1000).toFixed(2)}s`,
+    backoffMultiplier: 2,
+    retryableStatusCodes: [statusName(status.UNAVAILABLE), statusName(status.UNKNOWN)],
+  };
+
   return JSON.stringify({
     loadBalancingConfig: [],
     methodConfig: [
       {
-        name: [{ service: serviceName }],
-        retryPolicy: {
-          maxAttempts: 3,
-          initialBackoff: `${Math.round(initialBackoff / 1000).toFixed(2)}s`,
-          maxBackoff: `${Math.round(maxBackoff / 1000).toFixed(2)}s`,
-          backoffMultiplier: 2,
-          retryableStatusCodes: [statusName(status.UNAVAILABLE), statusName(status.UNKNOWN)],
-        },
+        name: serviceNames.map((serviceName) => ({ service: serviceName })),
+        retryPolicy,
       },
     ],
   });

libs/providers/flagd/src/lib/service/grpc/grpc-service.ts

@@ -1,4 +1,4 @@
-import type { ClientReadableStream, ClientUnaryCall, ServiceError } from '@grpc/grpc-js';
+import { type ClientReadableStream, type ClientUnaryCall, type ServiceError } from '@grpc/grpc-js';
 import { status } from '@grpc/grpc-js';
 import { ConnectivityState } from '@grpc/grpc-js/build/src/connectivity-state';
 import type { EvaluationContext, FlagValue, JsonValue, Logger, ResolutionDetails } from '@openfeature/server-sdk';
@@ -27,7 +27,12 @@ import type {
 } from '../../../proto/ts/flagd/evaluation/v1/evaluation';
 import { ServiceClient } from '../../../proto/ts/flagd/evaluation/v1/evaluation';
 import type { FlagdGrpcConfig } from '../../configuration';
-import { DEFAULT_MAX_CACHE_SIZE, EVENT_CONFIGURATION_CHANGE, EVENT_PROVIDER_READY } from '../../constants';
+import {
+  DEFAULT_MAX_BACKOFF_MS,
+  DEFAULT_MAX_CACHE_SIZE,
+  EVENT_CONFIGURATION_CHANGE,
+  EVENT_PROVIDER_READY,
+} from '../../constants';
 import { FlagdProvider } from '../../flagd-provider';
 import type { Service } from '../service';
 import {
@@ -79,6 +84,8 @@ export class GRPCService implements Service {
   private readonly _fatalStatusCodes: Set<number>;
   private _initialized = false;
   private _streamDeadline: number;
+  private _maxBackoffMs: number;
+  private _errorThrottled = false;
 
   private get _cacheActive() {
     // the cache is "active" (able to be used) if the config enabled it, AND the gRPC stream is live
@@ -94,6 +101,7 @@ export class GRPCService implements Service {
     const clientOptions = buildClientOptions(config);
     const channelCredentials = createChannelCredentials(tls, certPath);
 
+    this._maxBackoffMs = config.retryBackoffMaxMs || DEFAULT_MAX_BACKOFF_MS;
     this._client = client
       ? client
       : new ServiceClient(socketPath ? `unix://${socketPath}` : `${host}:${port}`, channelCredentials, clientOptions);
@@ -175,32 +183,48 @@ export class GRPCService implements Service {
     // close the previous stream if we're reconnecting
     closeStreamIfDefined(this._eventStream);
 
-    const deadline = this._streamDeadline != 0 ? Date.now() + this._streamDeadline : undefined;
-    const stream = this._client.eventStream({ waitForReady: true }, { deadline });
-    stream.on('error', (err: Error) => {
-      // Check if error is a fatal status code on first connection only
-      if (isFatalStatusCodeError(err, this._initialized, this._fatalStatusCodes)) {
-        handleFatalStatusCodeError(err, this.logger, disconnectCallback, rejectConnect);
-        return;
-      }
-      rejectConnect?.(err);
-      this.handleError(reconnectCallback, changedCallback, disconnectCallback);
-    });
-    stream.on('data', (message) => {
-      if (message.type === EVENT_PROVIDER_READY) {
-        this.logger?.debug(`${FlagdProvider.name}: streaming connection established with flagd`);
-        this._initialized = true;
-        // if resolveConnect is undefined, this is a reconnection; we only want to fire the reconnect callback in that case
-        if (resolveConnect) {
-          resolveConnect();
-        } else {
-          reconnectCallback();
+    // wait for connection to be stable
+    this._client.waitForReady(Date.now() + this._deadline, (err) => {
+      if (err) {
+        // Check if error is a fatal status code on first connection only
+        if (isFatalStatusCodeError(err, this._initialized, this._fatalStatusCodes)) {
+          handleFatalStatusCodeError(err, this.logger, disconnectCallback, rejectConnect);
+          return;
         }
-      } else if (message.type === EVENT_CONFIGURATION_CHANGE) {
-        this.handleFlagsChanged(message, changedCallback);
+        rejectConnect?.(err);
+        this.handleError(reconnectCallback, changedCallback, disconnectCallback);
+      } else {
+        const streamDeadline = this._streamDeadline != 0 ? Date.now() + this._streamDeadline : undefined;
+        const stream = this._client.eventStream({}, { deadline: streamDeadline });
+        stream.on('error', (err: Error) => {
+          // In cases where we get an explicit error status, we add a delay.
+          // This prevents tight loops when errors are returned immediately, typically by intervening proxies like Envoy.
+          this._errorThrottled = true;
+          // Check if error is a fatal status code on first connection only
+          if (isFatalStatusCodeError(err, this._initialized, this._fatalStatusCodes)) {
+            handleFatalStatusCodeError(err, this.logger, disconnectCallback, rejectConnect);
+            return;
+          }
+          rejectConnect?.(err);
+          this.handleError(reconnectCallback, changedCallback, disconnectCallback);
+        });
+        stream.on('data', (message) => {
+          if (message.type === EVENT_PROVIDER_READY) {
+            this.logger?.debug(`${FlagdProvider.name}: streaming connection established with flagd`);
+            this._initialized = true;
+            // if resolveConnect is undefined, this is a reconnection; we only want to fire the reconnect callback in that case
+            if (resolveConnect) {
+              resolveConnect();
+            } else {
+              reconnectCallback();
+            }
+          } else if (message.type === EVENT_CONFIGURATION_CHANGE) {
+            this.handleFlagsChanged(message, changedCallback);
+          }
+        });
+        this._eventStream = stream;
       }
     });
-    this._eventStream = stream;
   }
 
   private handleFlagsChanged(message: EventStreamResponse, changedCallback: (flagsChanged: string[]) => void) {
@@ -228,10 +252,11 @@ export class GRPCService implements Service {
     changedCallback: (flagsChanged: string[]) => void,
     disconnectCallback: (message: string) => void,
   ) {
-    const channel = this._client.getChannel();
-    channel.watchConnectivityState(channel.getConnectivityState(true), Infinity, () => {
-      this.listen(reconnectCallback, changedCallback, disconnectCallback);
-    });
+    setTimeout(
+      () => this.listen(reconnectCallback, changedCallback, disconnectCallback),
+      this._errorThrottled ? this._maxBackoffMs : 0,
+    );
+    this._errorThrottled = false;
   }
 
   private handleError(