fix: race condition when flush() does not complete

Author: joker23

packages/sdk/node-client/__tests__/NodeClientFDv2.test.ts

@@ -296,13 +296,13 @@ it('duplicate-mode setConnectionMode awaits an in-flight transition to the same
     localStoragePath: tmpRoot,
   });
 
-  // Keeps the queue draining when p3 is issued, making the race deterministic.
+  // flush() must not reject during the offline leg so p1 does not poison the queue.
   const flushSpy = jest
     .spyOn(LDClientImpl.prototype, 'flush')
     .mockImplementation(() => Promise.resolve({ result: true }));
 
-  // p3 hits the idempotency guard because p2 sets _connectionMode synchronously,
-  // before p2's queued task runs. Without the await, p3 resolves too early.
+  // p3 queues behind p2; by the time p3's task runs, p2 has already committed
+  // _connectionMode to 'streaming', so the idempotency check fires and p3 no-ops.
   const p1 = client.setConnectionMode('offline');
   const p2 = client.setConnectionMode('streaming');
   const p3 = client.setConnectionMode('streaming');
@@ -348,35 +348,31 @@ it('setConnectionMode with an invalid mode logs a warning and does not delegate
   expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('invalid-mode'));
 });
 
-// ------ Mode rollback on failed transition ------
+// ------ Failed offline transition ------
 
-it('restores connection mode when an FDv2 offline transition task throws', async () => {
+it('setConnectionMode rejects and does not change connection mode when flush fails in FDv2', async () => {
   const client = createClient('client-side-id', DEFAULT_INITIAL_CONTEXT, {
     dataSystem: {},
     diagnosticOptOut: true,
     sendEvents: false,
     logger,
     localStoragePath: tmpRoot,
   });
-  // flush() is the first async step, so a reject here means setConnectionMode
-  // on the data manager is never reached.
+  // flush() throws before _connectionMode is written or the data manager is notified.
   const flushSpy = jest
     .spyOn(LDClientImpl.prototype, 'flush')
     .mockRejectedValueOnce(new Error('flush failed'));
 
   await expect(client.setConnectionMode('offline')).rejects.toThrow('flush failed');
 
-  // The synchronously-set 'offline' is rolled back: the data manager is still
-  // streaming, so isOffline() reflects actual state.
   expect(client.getConnectionMode()).toBe('streaming');
   expect(client.isOffline()).toBe(false);
-  // The data manager was never told to go offline.
   expect(mockSetConnectionMode).not.toHaveBeenCalled();
 
   flushSpy.mockRestore();
 });
 
-it('keeps the FDv2 transition queue alive for subsequent calls after a failed transition', async () => {
+it('setConnectionMode queue stays alive after a failed offline transition in FDv2', async () => {
   const client = createClient('client-side-id', DEFAULT_INITIAL_CONTEXT, {
     dataSystem: {},
     diagnosticOptOut: true,
@@ -391,7 +387,7 @@ it('keeps the FDv2 transition queue alive for subsequent calls after a failed tr
   await expect(client.setConnectionMode('offline')).rejects.toThrow('flush failed');
   flushSpy.mockRestore();
 
-  // A subsequent transition must still work; the queue was not poisoned.
+  // task.catch() keeps _fdv2ConnectionModeQueue alive; the next call must succeed.
   await client.setConnectionMode('polling');
   expect(client.getConnectionMode()).toBe('polling');
   expect(mockSetConnectionMode).toHaveBeenCalledWith('polling');

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

@@ -36,15 +36,13 @@ import NodePlatform from './platform/NodePlatform';
 
 export class NodeClient extends LDClientImpl {
   private readonly _plugins: LDPlugin[];
-  // Tracks the current connection mode for both FDv1 (NodeDataManager) and
-  // FDv2 paths. Updated synchronously so getConnectionMode()/isOffline()
-  // reflect the caller's intent without waiting for async data manager work.
+  // Connection mode for both paths. FDv2 updates this inside the
+  // serialized queue task; FDv1 reads it back from the data manager after each
+  // transition.
   private _connectionMode: ConnectionMode;
-  // Serializes FDv2 connection-mode transitions so concurrent calls cannot
-  // reorder flush/event-sending around the await in the offline branch.
+  // Serializes FDv2 mode transitions so concurrent callers cannot interleave
+  // flush() and setEventSendingEnabled() across the offline branch's await.
   private _fdv2ConnectionModeQueue: Promise<void> = Promise.resolve();
-  // Set to true when close() is called so post-close setConnectionMode calls
-  // are no-ops in the FDv2 path.
   private _closed: boolean = false;
 
   constructor(envKey: string, initialContext: LDContext, options: NodeOptions = {}) {
@@ -180,50 +178,30 @@ export class NodeClient extends LDClientImpl {
 
   async setConnectionMode(mode: ConnectionMode): Promise<void> {
     if (this.isFDv2) {
-      // Validate mode before forwarding to prevent a runtime crash inside the
-      // debounce timer when an invalid value is passed from JavaScript.
+      // JS callers can pass arbitrary strings; validate before the debounce timer fires.
       if (mode !== 'offline' && mode !== 'streaming' && mode !== 'polling') {
         this.logger.warn(`[NodeClient] Unknown connection mode "${mode}", ignoring.`);
         return;
       }
       if (this._closed) {
         return;
       }
-      if (mode === this._connectionMode) {
-        // Await without appending: an earlier call may still be transitioning to
-        // this same mode, and we must not resolve until it has finished.
-        await this._fdv2ConnectionModeQueue;
-        return;
-      }
-      // Without serialization, concurrent calls can interleave flush() and
-      // setEventSendingEnabled() across an offline await, corrupting order.
-      // Capture the prior mode so a failed transition can be rolled back:
-      // otherwise isOffline()/getConnectionMode() would report a state the
-      // data manager never actually entered.
-      const previousMode = this._connectionMode;
-      this._connectionMode = mode;
       const task = this._fdv2ConnectionModeQueue.then(async () => {
         // Re-check: close() may have been called while this task was queued.
         if (this._closed) {
           return;
         }
-        try {
-          if (mode === 'offline') {
-            await this.flush();
-            this.setEventSendingEnabled(false, false);
-          }
-          (this.dataManager as FDv2DataManagerControl).setConnectionMode(mode);
-          if (mode !== 'offline') {
-            this.setEventSendingEnabled(true, false);
-          }
-        } catch (e) {
-          // Only roll back if no later queued task has already moved past this
-          // mode; otherwise isOffline()/getConnectionMode() would drift from
-          // the state the data manager actually reached.
-          if (this._connectionMode === mode) {
-            this._connectionMode = previousMode;
-          }
-          throw e;
+        if (mode === this._connectionMode) {
+          return;
+        }
+        if (mode === 'offline') {
+          await this.flush();
+          this.setEventSendingEnabled(false, false);
+        }
+        (this.dataManager as FDv2DataManagerControl).setConnectionMode(mode);
+        this._connectionMode = mode;
+        if (mode !== 'offline') {
+          this.setEventSendingEnabled(true, false);
         }
       });
       this._fdv2ConnectionModeQueue = task.catch(() => {});