fix(analytics): guard polling start so concurrent callers don't leak intervals (SDK-81)

packages/mixpanel/lib/flags/local_flags.js

@@ -55,14 +55,32 @@ class LocalFeatureFlagsProvider extends FeatureFlagsProvider {
     this.flagDefinitions = new Map();
     this.pollingInterval = null;
     this._initialFetchPromise = null;
+    // Cached in-flight start so concurrent startPollingForDefinitions() calls
+    // share the same async work. In JS's single-threaded model the
+    // !this.pollingInterval check + setInterval assignment are atomic within a
+    // microtask, so setInterval itself isn't racy. What the guard prevents is
+    // N concurrent starts each awaiting their own _fetchFlagDefinitions() and
+    // clobbering _initialFetchPromise — N redundant HTTP calls where one
+    // would do.
+    this._startPromise = null;
   }
 
   /**
    * Start polling for flag definitions.
-   * Fetches immediately and then at regular intervals if polling is enabled
+   * Fetches immediately and then at regular intervals if polling is enabled.
+   * Concurrent calls share the same in-flight promise; subsequent calls
+   * after start has completed are a no-op until stopPollingForDefinitions().
    * @returns {Promise<void>}
    */
   async startPollingForDefinitions() {
+    if (this._startPromise) {
+      return this._startPromise;
+    }
+    this._startPromise = this._doStartPolling();
+    return this._startPromise;
+  }
+
+  async _doStartPolling() {
     try {
       this._initialFetchPromise = this._fetchFlagDefinitions();
       await this._initialFetchPromise;
@@ -72,13 +90,25 @@ class LocalFeatureFlagsProvider extends FeatureFlagsProvider {
           try {
             await this._fetchFlagDefinitions();
           } catch (err) {
+            // Stop polling on the first refresh failure and reset state
+            // (clears the interval and _startPromise) so the caller can
+            // decide to restart. Continuing to poll a failing endpoint
+            // just spams the log without recovering — if it's transient
+            // (network blip), a follow-up startPollingForDefinitions()
+            // gets us back; if it's permanent (bad token, revoked
+            // project), silent retries hide the real failure.
             this.logger?.error(
-              `Error polling for flag definition: ${err.message}`,
+              `Error polling for flag definitions, stopping polling: ${err.message}`,
             );
+            this.stopPollingForDefinitions();
           }
         }, this.config.polling_interval_in_seconds * 1000);
       }
     } catch (err) {
+      // Drop the cached start so the caller can retry via
+      // startPollingForDefinitions(). Otherwise a failed initial
+      // fetch would permanently short-circuit subsequent calls.
+      this._startPromise = null;
       this.logger?.error(
         `Initial flag definitions fetch failed: ${err.message}`,
       );
@@ -92,18 +122,29 @@ class LocalFeatureFlagsProvider extends FeatureFlagsProvider {
     if (this.pollingInterval) {
       clearInterval(this.pollingInterval);
       this.pollingInterval = null;
-    } else {
+    } else if (!this._startPromise) {
+      // Only warn when the caller stopped something that was truly never
+      // started. The enable_polling=false lifecycle (start → stop) and the
+      // mid-start case (stop before setInterval fires) both leave
+      // pollingInterval null with a legitimate _startPromise present —
+      // warning in those cases is noise.
       this.logger?.warn(
         "stopPollingForDefinitions called but polling was not active",
       );
     }
+    // Always clear the cached start so a subsequent
+    // startPollingForDefinitions() can re-start cleanly, including
+    // the enable_polling=false case where pollingInterval was never
+    // set. Matches shutdown().
+    this._startPromise = null;
   }
 
   shutdown() {
     if (this.pollingInterval) {
       clearInterval(this.pollingInterval);
       this.pollingInterval = null;
     }
+    this._startPromise = null;
   }
 
   areFlagsReady() {

packages/mixpanel/test/flags/local_flags.js

@@ -886,4 +886,177 @@ describe("LocalFeatureFlagsProvider", () => {
       // This behavior is tested in the FeatureFlagsProvider base class tests.
     });
   });
+
+  describe("polling lifecycle", () => {
+    it("dedups initial fetch under concurrent startPollingForDefinitions calls", async () => {
+      // Mock setInterval so the spy records the call without scheduling a real
+      // timer. A real timer would keep the test process alive if an assertion
+      // failed before stopPollingForDefinitions() ran.
+      const setIntervalSpy = vi
+        .spyOn(global, "setInterval")
+        .mockImplementation(() => 999);
+
+      const provider = new LocalFeatureFlagsProvider(
+        TEST_TOKEN,
+        {
+          api_host: "localhost",
+          enable_polling: true,
+          polling_interval_in_seconds: 3600,
+        },
+        mockTracker,
+        mockLogger,
+      );
+
+      nock("https://localhost")
+        .persist()
+        .get("/flags/definitions")
+        .query(true)
+        .reply(200, { code: 200, flags: [] });
+
+      // Count fetch invocations — the guard should coalesce concurrent
+      // starts into a single _fetchFlagDefinitions() call. Asserting on
+      // setInterval alone wouldn't catch a regression: JS's single-threaded
+      // microtask model makes the !pollingInterval check-then-set atomic
+      // even without the guard, so setInterval would be called once
+      // regardless.
+      const originalFetch = provider._fetchFlagDefinitions.bind(provider);
+      const fetchSpy = vi
+        .spyOn(provider, "_fetchFlagDefinitions")
+        .mockImplementation(() => originalFetch());
+
+      const N = 8;
+      const starts = [];
+      for (let i = 0; i < N; i++) {
+        starts.push(provider.startPollingForDefinitions());
+      }
+      await Promise.all(starts);
+
+      expect(fetchSpy).toHaveBeenCalledTimes(1);
+      expect(setIntervalSpy).toHaveBeenCalledTimes(1);
+      expect(provider.pollingInterval).not.toBeNull();
+
+      provider.stopPollingForDefinitions();
+      fetchSpy.mockRestore();
+      setIntervalSpy.mockRestore();
+    });
+
+    it("stopPollingForDefinitions clears the cached start so start can re-fetch (enable_polling=false)", async () => {
+      const provider = new LocalFeatureFlagsProvider(
+        TEST_TOKEN,
+        {
+          api_host: "localhost",
+          enable_polling: false,
+        },
+        mockTracker,
+        mockLogger,
+      );
+
+      nock("https://localhost")
+        .persist()
+        .get("/flags/definitions")
+        .query(true)
+        .reply(200, { code: 200, flags: [] });
+
+      const originalFetch = provider._fetchFlagDefinitions.bind(provider);
+      const fetchSpy = vi
+        .spyOn(provider, "_fetchFlagDefinitions")
+        .mockImplementation(() => originalFetch());
+
+      await provider.startPollingForDefinitions();
+      provider.stopPollingForDefinitions();
+      await provider.startPollingForDefinitions();
+
+      expect(fetchSpy).toHaveBeenCalledTimes(2);
+
+      fetchSpy.mockRestore();
+    });
+
+    it("failed initial fetch clears the cached start so subsequent start retries", async () => {
+      const provider = new LocalFeatureFlagsProvider(
+        TEST_TOKEN,
+        {
+          api_host: "localhost",
+          enable_polling: false,
+        },
+        mockTracker,
+        mockLogger,
+      );
+
+      // First attempt fails, second succeeds.
+      nock("https://localhost")
+        .get("/flags/definitions")
+        .query(true)
+        .reply(500);
+      nock("https://localhost")
+        .get("/flags/definitions")
+        .query(true)
+        .reply(200, { code: 200, flags: [] });
+
+      const originalFetch = provider._fetchFlagDefinitions.bind(provider);
+      const fetchSpy = vi
+        .spyOn(provider, "_fetchFlagDefinitions")
+        .mockImplementation(() => originalFetch());
+
+      await provider.startPollingForDefinitions();
+      await provider.startPollingForDefinitions();
+
+      expect(fetchSpy).toHaveBeenCalledTimes(2);
+
+      fetchSpy.mockRestore();
+    });
+
+    it("stops polling and resets state when a refresh fails", async () => {
+      // Capture the interval callback synchronously so we can drive it
+      // directly without letting the real timer fire (and keep the test
+      // deterministic).
+      let intervalCallback = null;
+      const setIntervalSpy = vi
+        .spyOn(global, "setInterval")
+        .mockImplementation((cb) => {
+          intervalCallback = cb;
+          return 999;
+        });
+      const clearIntervalSpy = vi.spyOn(global, "clearInterval");
+
+      const provider = new LocalFeatureFlagsProvider(
+        TEST_TOKEN,
+        {
+          api_host: "localhost",
+          enable_polling: true,
+          polling_interval_in_seconds: 3600,
+        },
+        mockTracker,
+        mockLogger,
+      );
+
+      // Initial fetch succeeds so we get past _doStartPolling and set
+      // up the interval; second fetch (driven by the interval) fails.
+      nock("https://localhost")
+        .get("/flags/definitions")
+        .query(true)
+        .reply(200, { code: 200, flags: [] });
+      nock("https://localhost")
+        .get("/flags/definitions")
+        .query(true)
+        .reply(500);
+
+      await provider.startPollingForDefinitions();
+      expect(provider.pollingInterval).not.toBeNull();
+      expect(intervalCallback).not.toBeNull();
+
+      // Fire the poller manually and wait for the awaited fetch inside
+      // to reject and hit the catch.
+      await intervalCallback();
+
+      expect(clearIntervalSpy).toHaveBeenCalledWith(999);
+      expect(provider.pollingInterval).toBeNull();
+      expect(provider._startPromise).toBeNull();
+      expect(mockLogger.error).toHaveBeenCalledWith(
+        expect.stringContaining("stopping polling"),
+      );
+
+      clearIntervalSpy.mockRestore();
+      setIntervalSpy.mockRestore();
+    });
+  });
 });