chore: bot comments

fixing up the race conditions that could happen when connection mode is
changed during identify

Author: joker23

packages/sdk/node-client/__tests__/NodeClient.dataSource.test.ts

@@ -407,6 +407,142 @@ it('does not read cached flags when bootstrap is provided', async () => {
   await client.close();
 });
 
+it('rejects an in-flight identify immediately when close() is called while the processor awaits its first event', async () => {
+  // The streaming processor never fires a 'put' event, simulating a slow or stalled stream.
+  // close() must reject _pendingIdentifyReject promptly so the caller is not left waiting for
+  // the 5-second identify timeout.
+  const createEventSource = jest.fn(() => ({
+    addEventListener: jest.fn(),
+    close: jest.fn(),
+    onclose: jest.fn(),
+    onerror: jest.fn(),
+    onopen: jest.fn(),
+    onretrying: jest.fn(),
+  }));
+  const fakePlatform = makeMockPlatform({
+    requests: {
+      fetch: jest.fn(),
+      createEventSource: createEventSource as any,
+      getEventSourceCapabilities: () => ({ readTimeout: true, headers: true, customMethod: true }),
+    },
+  });
+  NodePlatformMock.mockImplementationOnce(() => fakePlatform);
+
+  const client = createClient(
+    'client-side-id',
+    { kind: 'user', key: 'bob' },
+    {
+      initialConnectionMode: 'streaming',
+      sendEvents: false,
+      diagnosticOptOut: true,
+      logger,
+    },
+  );
+
+  await client.start({ bootstrap: bootstrapData });
+
+  // Identify without bootstrap: cache miss -> _setupConnection -> processor running, no 'put' yet.
+  const identifyPromise = client.identify({ kind: 'user', key: 'alice' }, { timeout: 5 });
+  // Allow the identify to reach the streaming processor before closing.
+  await new Promise((resolve) => setTimeout(resolve, 10));
+
+  await client.close();
+
+  const result = await identifyPromise;
+  expect(result.status).toBe('error');
+});
+
+it('rejects an in-flight identify when setConnectionMode offline runs after the processor started', async () => {
+  // The processor phase: identify has passed loadCached and _setupConnection has registered
+  // _pendingIdentifyReject. The stream never delivers a 'put', so the only way out is the
+  // reject path.
+  const createEventSource = jest.fn(() => ({
+    addEventListener: jest.fn(),
+    close: jest.fn(),
+    onclose: jest.fn(),
+    onerror: jest.fn(),
+    onopen: jest.fn(),
+    onretrying: jest.fn(),
+  }));
+  const fakePlatform = makeMockPlatform({
+    requests: {
+      fetch: jest.fn(),
+      createEventSource: createEventSource as any,
+      getEventSourceCapabilities: () => ({ readTimeout: true, headers: true, customMethod: true }),
+    },
+  });
+  NodePlatformMock.mockImplementationOnce(() => fakePlatform);
+
+  const client = createClient(
+    'client-side-id',
+    { kind: 'user', key: 'bob' },
+    {
+      initialConnectionMode: 'streaming',
+      sendEvents: false,
+      diagnosticOptOut: true,
+      logger,
+    },
+  );
+
+  await client.start({ bootstrap: bootstrapData });
+
+  const identifyPromise = client.identify({ kind: 'user', key: 'alice' }, { timeout: 5 });
+  await new Promise((resolve) => setTimeout(resolve, 10));
+
+  // Go offline while the streaming processor is running and waiting for its first event.
+  await client.setConnectionMode('offline');
+
+  const result = await identifyPromise;
+  expect(result.status).toBe('error');
+
+  await client.close();
+});
+
+it('rejects an in-flight identify when setConnectionMode switches modes while the processor is running', async () => {
+  // Switching from streaming to polling replaces the active processor. The pending identify
+  // on the old processor must be rejected rather than silently abandoned.
+  const createEventSource = jest.fn(() => ({
+    addEventListener: jest.fn(),
+    close: jest.fn(),
+    onclose: jest.fn(),
+    onerror: jest.fn(),
+    onopen: jest.fn(),
+    onretrying: jest.fn(),
+  }));
+  const fakePlatform = makeMockPlatform({
+    requests: {
+      fetch: jest.fn(),
+      createEventSource: createEventSource as any,
+      getEventSourceCapabilities: () => ({ readTimeout: true, headers: true, customMethod: true }),
+    },
+  });
+  NodePlatformMock.mockImplementationOnce(() => fakePlatform);
+
+  const client = createClient(
+    'client-side-id',
+    { kind: 'user', key: 'bob' },
+    {
+      initialConnectionMode: 'streaming',
+      sendEvents: false,
+      diagnosticOptOut: true,
+      logger,
+    },
+  );
+
+  await client.start({ bootstrap: bootstrapData });
+
+  const identifyPromise = client.identify({ kind: 'user', key: 'alice' }, { timeout: 5 });
+  await new Promise((resolve) => setTimeout(resolve, 10));
+
+  // Switch to polling -- the old streaming processor is replaced; the pending identify rejects.
+  await client.setConnectionMode('polling');
+
+  const result = await identifyPromise;
+  expect(result.status).toBe('error');
+
+  await client.close();
+});
+
 it('rejects identify rather than hanging when the mode flips to offline mid-identify', async () => {
   // Gate the cached-flag read so we can flip the connection mode while identify is parked on
   // the await -- reproducing the race where _setupConnection later sees connectionMode==='offline'.

packages/sdk/node-client/src/NodeDataManager.ts

@@ -21,6 +21,7 @@ const logTag = '[NodeDataManager]';
 
 export default class NodeDataManager extends BaseDataManager {
   protected connectionMode: ConnectionMode = 'streaming';
+  private _pendingIdentifyReject?: (err: Error) => void;
 
   constructor(
     platform: Platform,
@@ -65,7 +66,9 @@ export default class NodeDataManager extends BaseDataManager {
     }
     this.context = context;
 
-    const offline = this.connectionMode === 'offline';
+    // Snapshot the mode before any await so the bootstrap path and the stale-snapshot
+    // detection below both see a consistent starting point.
+    const startedOffline = this.connectionMode === 'offline';
 
     // Bootstrap and cache are mutually exclusive: when bootstrap data is provided it
     // resolves identify immediately, so we must not also load (and potentially overwrite
@@ -77,17 +80,19 @@ export default class NodeDataManager extends BaseDataManager {
         );
       }
       this._finishIdentifyFromBootstrap(context, identifyOptions, identifyResolve);
-      if (!offline) {
+      if (!startedOffline) {
         // Open a connection for ongoing updates, but identify is already resolved so no
         // callbacks are forwarded.
         this._setupConnection(context);
       }
       return;
     }
 
-    const waitForNetworkResults = !!identifyOptions?.waitForNetworkResults && !offline;
-
     const loadedFromCache = await this.flagManager.loadCached(context);
+    // Re-read connectionMode after the await: a concurrent setConnectionMode call may have
+    // changed it while loadCached was in flight.
+    const offline = this.connectionMode === 'offline';
+    const waitForNetworkResults = !!identifyOptions?.waitForNetworkResults && !offline;
     let identifyResolved = false;
     if (loadedFromCache && !waitForNetworkResults) {
       this._debugLog('Identify completing with cached flags');
@@ -96,6 +101,15 @@ export default class NodeDataManager extends BaseDataManager {
     }
 
     if (offline) {
+      if (!startedOffline) {
+        // The connection mode changed to offline while we were awaiting the cache. Reject
+        // rather than silently resolve so the caller knows the identify did not complete
+        // in the originally-requested mode.
+        if (!identifyResolved) {
+          identifyReject(new Error("Connection mode changed to 'offline' during identify."));
+        }
+        return;
+      }
       if (loadedFromCache) {
         this._debugLog('Offline identify - using cached flags.');
       } else {
@@ -142,6 +156,22 @@ export default class NodeDataManager extends BaseDataManager {
       return;
     }
 
+    // Wrap callbacks so _pendingIdentifyReject is cleared as soon as the identify settles,
+    // preventing a stale reject from firing if setConnectionMode runs after resolution.
+    const wrappedResolve = identifyResolve
+      ? () => {
+          this._pendingIdentifyReject = undefined;
+          identifyResolve();
+        }
+      : undefined;
+    const wrappedReject = identifyReject
+      ? (err: Error) => {
+          this._pendingIdentifyReject = undefined;
+          identifyReject(err);
+        }
+      : undefined;
+    this._pendingIdentifyReject = wrappedReject;
+
     const plainContextString = JSON.stringify(rawContext);
     const requestor = makeRequestor(
       plainContextString,
@@ -163,27 +193,27 @@ export default class NodeDataManager extends BaseDataManager {
           rawContext,
           context,
           requestor,
-          identifyResolve,
-          identifyReject,
+          wrappedResolve,
+          wrappedReject,
         );
         break;
       case 'polling':
         this.createPollingProcessor(
           rawContext,
           context,
           requestor,
-          identifyResolve,
-          identifyReject,
+          wrappedResolve,
+          wrappedReject,
         );
         break;
       default:
         this.logger.warn(
           `${logTag} _setupConnection called with unsupported connectionMode '${this.connectionMode}'.`,
         );
         this.updateProcessor = undefined;
-        // The mode may have changed to 'offline' while identify was awaiting the cache; reject
-        // rather than leave the identify promise to hang until its timeout.
-        identifyReject?.(
+        // connectionMode is an unsupported value; reject the in-flight identify so the
+        // promise does not hang until its timeout.
+        wrappedReject?.(
           new Error(`Connection mode changed to '${this.connectionMode}' during identify.`),
         );
         return;
@@ -207,12 +237,26 @@ export default class NodeDataManager extends BaseDataManager {
 
     switch (mode) {
       case 'offline':
+        // Reject any in-flight identify before closing the processor -- the processor's
+        // close() does not invoke pending callbacks, so without this the identify promise
+        // would hang until its timeout.
+        if (this._pendingIdentifyReject) {
+          const reject = this._pendingIdentifyReject;
+          this._pendingIdentifyReject = undefined;
+          reject(new Error("Connection mode changed to 'offline' during identify."));
+        }
         this.updateProcessor?.close();
         this.updateProcessor = undefined;
         break;
       case 'polling':
       case 'streaming':
         if (this.context) {
+          // Reject any in-flight identify from the previous processor before replacing it.
+          if (this._pendingIdentifyReject) {
+            const reject = this._pendingIdentifyReject;
+            this._pendingIdentifyReject = undefined;
+            reject(new Error(`Connection mode changed to '${mode}' during identify.`));
+          }
           this._setupConnection(this.context);
         }
         break;
@@ -227,4 +271,13 @@ export default class NodeDataManager extends BaseDataManager {
   getConnectionMode(): ConnectionMode {
     return this.connectionMode;
   }
+
+  override close(): void {
+    if (this._pendingIdentifyReject) {
+      const reject = this._pendingIdentifyReject;
+      this._pendingIdentifyReject = undefined;
+      reject(new Error('Client has been closed.'));
+    }
+    super.close();
+  }
 }