diff --git a/packages/mixpanel/lib/flags/local_flags.js b/packages/mixpanel/lib/flags/local_flags.js index 712373a..4a5f986 100644 --- a/packages/mixpanel/lib/flags/local_flags.js +++ b/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} */ 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,11 +122,21 @@ 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() { @@ -104,6 +144,7 @@ class LocalFeatureFlagsProvider extends FeatureFlagsProvider { clearInterval(this.pollingInterval); this.pollingInterval = null; } + this._startPromise = null; } areFlagsReady() { diff --git a/packages/mixpanel/test/flags/local_flags.js b/packages/mixpanel/test/flags/local_flags.js index 2f06a8b..4223cdf 100644 --- a/packages/mixpanel/test/flags/local_flags.js +++ b/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(); + }); + }); });