fix: hanging grpc handles after shutdown (#683)

Signed-off-by: Todd Baert <[email protected]>

Author: toddbaert

libs/providers/flagd/jest.config.ts

@@ -13,5 +13,5 @@ export default {
   moduleFileExtensions: ['ts', 'js', 'html'],
   // ignore e2e path
   testPathIgnorePatterns: ["/e2e/"],
-  coverageDirectory: '../../../coverage/libs/providers/flagd',
+  coverageDirectory: '../../../coverage/libs/providers/flagd'
 };

libs/providers/flagd/project.json

@@ -78,8 +78,8 @@
       "executor": "nx:run-commands",
       "options": {
         "commands": [
-          "npx jest --setupFiles './setup-rpc-provider.ts'",
-          "npx jest --setupFiles './setup-in-process-provider.ts'"
+          "npx jest --setupFiles './setup-rpc-provider.ts' --runInBand --detectOpenHandles",
+          "npx jest --setupFiles './setup-in-process-provider.ts' --runInBand --detectOpenHandles"
         ],
         "cwd": "libs/providers/flagd/src/e2e",
         "parallel": false

libs/providers/flagd/src/e2e/jest.config.ts

@@ -7,5 +7,4 @@ export default {
   moduleNameMapper: {
     '@openfeature/flagd-core': ['<rootDir>/../../../../shared/flagd-core/src'],
   },
-  globalTeardown: './tear-down.ts',
 };

libs/providers/flagd/src/e2e/step-definitions/evaluation.spec.ts

@@ -29,6 +29,10 @@ defineFeature(feature, (test) => {
     });
   });
 
+  afterAll(async () => {
+    await OpenFeature.close();
+  });
+
   test('Resolves boolean value', ({ given, when, then }) => {
     let value: boolean;
     let flagKey: string;

libs/providers/flagd/src/e2e/step-definitions/flagd-json-evaluator.spec.ts

@@ -39,6 +39,10 @@ defineFeature(feature, (test) => {
     });
   });
 
+  afterAll(async () => {
+    await OpenFeature.close();
+  });
+
   test('Evaluator reuse', evaluateStringFlagWithContext);
 
   test('Fractional operator', ({ given, when, and, then }) => {

libs/providers/flagd/src/e2e/step-definitions/flagd-reconnect.unstable.spec.ts

@@ -20,6 +20,10 @@ defineFeature(feature, (test) => {
     });
   });
 
+  afterAll(async () => {
+    await OpenFeature.close();
+  });
+
   test('Provider reconnection', ({ given, when, then, and }) => {
     given('a flagd provider is set', () => {
       // handled in beforeAll

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

@@ -18,6 +18,10 @@ defineFeature(feature, (test) => {
     });
   });
 
+  afterAll(async () => {
+    await OpenFeature.close();
+  });
+
   test('Provider ready event', ({ given, when, then }) => {
     let ran = false;
 

libs/providers/flagd/src/e2e/tear-down.ts

@@ -81,6 +81,8 @@ describe(FlagdProvider.name, () => {
             }
           }),
           cancel: jest.fn(),
+          removeAllListeners: jest.fn(),
+          destroy: jest.fn(),
         };
       }),
       close: jest.fn(),
@@ -260,6 +262,8 @@ describe(FlagdProvider.name, () => {
         }
       }),
       cancel: jest.fn(),
+      removeAllListeners: jest.fn(),
+      destroy: jest.fn(),
     };
 
     const mockChannel = {
@@ -529,6 +533,8 @@ describe(FlagdProvider.name, () => {
             }
           }),
           cancel: jest.fn(),
+          removeAllListeners: jest.fn(),
+          destroy: jest.fn(),
         };
       }),
       close: jest.fn(),
@@ -635,6 +641,8 @@ describe(FlagdProvider.name, () => {
   });
 
   describe('shutdown', () => {
+    const removeAllListenersMock = jest.fn();
+    const destroyMock = jest.fn();
     const cancelMock = jest.fn();
     const closeMock = jest.fn();
 
@@ -648,6 +656,8 @@ describe(FlagdProvider.name, () => {
             }
           }),
           cancel: cancelMock,
+          removeAllListeners: removeAllListenersMock,
+          destroy: destroyMock,
         };
       }),
       close: closeMock,
@@ -668,7 +678,8 @@ describe(FlagdProvider.name, () => {
       it('should call service disconnect', async () => {
         await OpenFeature.close();
         expect(cancelMock).toHaveBeenCalled();
-        expect(closeMock).toHaveBeenCalled();
+        expect(removeAllListenersMock).toHaveBeenCalled();
+        expect(destroyMock).toHaveBeenCalled();
       });
     });
   });

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

@@ -0,0 +1,16 @@
+import { ClientReadableStream } from '@grpc/grpc-js';
+
+export const closeStreamIfDefined = (stream: ClientReadableStream<unknown> | undefined) => {
+  /**
+   * cancel() is necessary to prevent calls from hanging the process, so we need to we need to remove all the
+   * handlers, and add a no-op for 'error' so we can cancel without bubbling up an exception
+   */
+  if (stream) {
+    stream.removeAllListeners();
+    stream.on('error', () => {
+      // swallow errors after closed
+    });
+    stream.cancel();
+    stream.destroy();
+  }
+};

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

@@ -0,0 +1 @@
+export * from './grpc-util';

libs/providers/flagd/src/lib/service/common/index.ts

@@ -32,6 +32,7 @@ import { Config } from '../../configuration';
 import { DEFAULT_MAX_CACHE_SIZE, EVENT_CONFIGURATION_CHANGE, EVENT_PROVIDER_READY } from '../../constants';
 import { FlagdProvider } from '../../flagd-provider';
 import { Service } from '../service';
+import { closeStreamIfDefined } from '../common';
 
 type AnyResponse =
   | ResolveBooleanResponse
@@ -104,8 +105,7 @@ export class GRPCService implements Service {
   }
 
   async disconnect(): Promise<void> {
-    // cancel the stream and close the connection
-    this._eventStream?.cancel();
+    closeStreamIfDefined(this._eventStream);
     this._client.close();
   }
 
@@ -153,6 +153,10 @@ export class GRPCService implements Service {
     rejectConnect?: (reason: Error) => void,
   ) {
     this.logger?.debug(`${FlagdProvider.name}: connecting stream...`);
+
+    // close the previous stream if we're reconnecting
+    closeStreamIfDefined(this._eventStream);
+
     const stream = this._client.eventStream({}, {});
     stream.on('error', (err: Error) => {
       rejectConnect?.(err);

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

@@ -4,6 +4,7 @@ import { GeneralError } from '@openfeature/server-sdk';
 import { FlagSyncServiceClient, SyncFlagsRequest, SyncFlagsResponse } from '../../../../proto/ts/sync/v1/sync_service';
 import { Config } from '../../../configuration';
 import { DataFetch } from '../data-fetch';
+import { closeStreamIfDefined } from '../../common';
 
 /**
  * Implements the gRPC sync contract to fetch flag data.
@@ -34,15 +35,14 @@ export class GrpcFetch implements DataFetch {
     changedCallback: (flagsChanged: string[]) => void,
     disconnectCallback: () => void,
   ): Promise<void> {
-    // note that we never reject the promise as sync is a long-running operation
     return new Promise((resolve, reject) =>
       this.listen(dataFillCallback, reconnectCallback, changedCallback, disconnectCallback, resolve, reject),
     );
   }
 
   disconnect() {
     this._logger?.debug('Disconnecting gRPC sync connection');
-    this._syncStream?.destroy();
+    closeStreamIfDefined(this._syncStream);
     this._syncClient.close();
   }
 
@@ -54,6 +54,7 @@ export class GrpcFetch implements DataFetch {
     resolveConnect?: () => void,
     rejectConnect?: (reason: Error) => void,
   ) {
+    closeStreamIfDefined(this._syncStream);
     this._syncStream = this._syncClient.syncFlags(this._request);
 
     this._syncStream.on('data', (data: SyncFlagsResponse) => {