You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by ar...@apache.org on 2026/07/14 10:57:51 UTC
(superset) branch master updated: fix(async-queries): back off polling when async event requests keep failing (#42012)
This is an automated email from the ASF dual-hosted git repository.
Antonio-RiveroMartnez pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new 42523f8cc4c fix(async-queries): back off polling when async event requests keep failing (#42012)
42523f8cc4c is described below
commit 42523f8cc4c841622b71e9c3ceb08edc6aaf768f
Author: Antonio Rivero <38...@users.noreply.github.com>
AuthorDate: Tue Jul 14 12:57:45 2026 +0200
fix(async-queries): back off polling when async event requests keep failing (#42012)
---
.../src/middleware/asyncEvent.test.ts | 231 +++++++++++++++++++++
superset-frontend/src/middleware/asyncEvent.ts | 29 ++-
2 files changed, 258 insertions(+), 2 deletions(-)
diff --git a/superset-frontend/src/middleware/asyncEvent.test.ts b/superset-frontend/src/middleware/asyncEvent.test.ts
index 0ec0031514e..e78e95769b7 100644
--- a/superset-frontend/src/middleware/asyncEvent.test.ts
+++ b/superset-frontend/src/middleware/asyncEvent.test.ts
@@ -173,6 +173,237 @@ describe('asyncEvent middleware', () => {
expect(fetchMock.callHistory.calls(CACHED_DATA_ENDPOINT)).toHaveLength(1);
});
+ test('backs off exponentially when polling requests keep failing', async () => {
+ // stop the real-timer polling loop started by beforeEach before
+ // switching to fake timers, so all polls run on the fake clock
+ mockedIsFeatureEnabled.mockReturnValueOnce(false);
+ asyncEvent.init(config);
+ jest.useFakeTimers();
+ try {
+ fetchMock.clearHistory().removeRoutes();
+ fetchMock.get(EVENTS_ENDPOINT, { status: 403 });
+ asyncEvent.init(config);
+ asyncEvent.waitForAsyncData(asyncPendingEvent).catch(() => {});
+
+ // first poll fires after the configured delay and fails
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ // next poll is delayed by 2x the configured delay, so nothing yet
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(2);
+
+ // after the second failure the delay grows to 4x
+ await jest.advanceTimersByTimeAsync(
+ 3 * config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(2);
+
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(3);
+ } finally {
+ jest.useRealTimers();
+ }
+ });
+
+ test('resumes the configured polling delay after a successful poll', async () => {
+ // stop the real-timer polling loop started by beforeEach before
+ // switching to fake timers, so all polls run on the fake clock
+ mockedIsFeatureEnabled.mockReturnValueOnce(false);
+ asyncEvent.init(config);
+ jest.useFakeTimers();
+ try {
+ fetchMock.clearHistory().removeRoutes();
+ fetchMock.get(EVENTS_ENDPOINT, { status: 403 });
+ asyncEvent.init(config);
+ asyncEvent.waitForAsyncData(asyncPendingEvent).catch(() => {});
+
+ // two failed polls: 1x delay, then 2x delay
+ await jest.advanceTimersByTimeAsync(
+ 3 * config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(2);
+
+ // subsequent polls succeed, resetting the backoff
+ fetchMock.clearHistory().removeRoutes();
+ fetchMock.get(EVENTS_ENDPOINT, {
+ status: 200,
+ body: { result: [] },
+ });
+
+ // third poll fires 4x delay after the second failure and succeeds
+ await jest.advanceTimersByTimeAsync(
+ 4 * config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ // polling is back to the configured delay
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(2);
+ } finally {
+ jest.useRealTimers();
+ }
+ });
+
+ test('caps the polling backoff delay at 60 seconds', async () => {
+ const MAX_ERROR_POLLING_DELAY_MS = 60000;
+ // stop the real-timer polling loop started by beforeEach before
+ // switching to fake timers, so all polls run on the fake clock
+ mockedIsFeatureEnabled.mockReturnValueOnce(false);
+ asyncEvent.init(config);
+ jest.useFakeTimers();
+ try {
+ fetchMock.clearHistory().removeRoutes();
+ fetchMock.get(EVENTS_ENDPOINT, { status: 403 });
+ asyncEvent.init(config);
+ asyncEvent.waitForAsyncData(asyncPendingEvent).catch(() => {});
+
+ // first poll fires after the configured delay and fails
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ // walk the uncapped backoff: after failure N the next delay is
+ // 2^N times the configured delay, which stays below the cap through
+ // failure 10 (50ms * 2^10 = 51.2s)
+ for (let failures = 1; failures <= 10; failures += 1) {
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY * 2 ** failures,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(
+ failures + 1,
+ );
+ }
+
+ // after failure 11 the uncapped delay would be 102.4s, so the cap
+ // takes over: no poll just before the 60s mark...
+ await jest.advanceTimersByTimeAsync(MAX_ERROR_POLLING_DELAY_MS - 1);
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(11);
+
+ // ...and the next poll fires exactly at 60s
+ await jest.advanceTimersByTimeAsync(1);
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(12);
+
+ // additional failures remain capped at 60s
+ await jest.advanceTimersByTimeAsync(MAX_ERROR_POLLING_DELAY_MS);
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(13);
+ } finally {
+ jest.useRealTimers();
+ }
+ });
+
+ test('does not start a second loop when re-initialized during an in-flight poll', async () => {
+ // stop the real-timer polling loop started by beforeEach before
+ // switching to fake timers, so all polls run on the fake clock
+ mockedIsFeatureEnabled.mockReturnValueOnce(false);
+ asyncEvent.init(config);
+ jest.useFakeTimers();
+ try {
+ fetchMock.clearHistory().removeRoutes();
+ let resolveFetch: (response: any) => void = () => {};
+ fetchMock.get(
+ EVENTS_ENDPOINT,
+ new Promise(resolve => {
+ resolveFetch = resolve;
+ }),
+ );
+ asyncEvent.init(config);
+ asyncEvent.waitForAsyncData(asyncPendingEvent).catch(() => {});
+
+ // first poll fires and stays in-flight
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ // re-init while that fetch is pending, then let it resolve; the
+ // stale invocation must not schedule a second loop
+ asyncEvent.init(config);
+ asyncEvent.waitForAsyncData(asyncPendingEvent).catch(() => {});
+ resolveFetch({ status: 200, body: { result: [] } });
+ await jest.advanceTimersByTimeAsync(0);
+
+ fetchMock.clearHistory().removeRoutes();
+ fetchMock.get(EVENTS_ENDPOINT, {
+ status: 200,
+ body: { result: [] },
+ });
+
+ // exactly one poll per delay from here on
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(2);
+ } finally {
+ jest.useRealTimers();
+ }
+ });
+
+ test('does not resume polling when re-initialized with the feature disabled during an in-flight poll', async () => {
+ // stop the real-timer polling loop started by beforeEach before
+ // switching to fake timers, so all polls run on the fake clock
+ mockedIsFeatureEnabled.mockReturnValueOnce(false);
+ asyncEvent.init(config);
+ jest.useFakeTimers();
+ try {
+ fetchMock.clearHistory().removeRoutes();
+ let resolveFetch: (response: any) => void = () => {};
+ fetchMock.get(
+ EVENTS_ENDPOINT,
+ new Promise(resolve => {
+ resolveFetch = resolve;
+ }),
+ );
+ asyncEvent.init(config);
+ asyncEvent.waitForAsyncData(asyncPendingEvent).catch(() => {});
+
+ // first poll fires and stays in-flight
+ await jest.advanceTimersByTimeAsync(
+ config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(1);
+
+ // disable the feature and re-init while the fetch is pending; the
+ // stale invocation must not restart the stopped loop when it resumes
+ mockedIsFeatureEnabled.mockReturnValueOnce(false);
+ asyncEvent.init(config);
+ resolveFetch({ status: 200, body: { result: [] } });
+ await jest.advanceTimersByTimeAsync(0);
+
+ fetchMock.clearHistory().removeRoutes();
+ fetchMock.get(EVENTS_ENDPOINT, {
+ status: 200,
+ body: { result: [] },
+ });
+
+ await jest.advanceTimersByTimeAsync(
+ 10 * config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY,
+ );
+ expect(fetchMock.callHistory.calls(EVENTS_ENDPOINT)).toHaveLength(0);
+ } finally {
+ jest.useRealTimers();
+ }
+ });
+
// Regression guard for the motivating CodeQL case: a job_id that collides
// with a built-in Object property (e.g. "__proto__"/"constructor") must be
// routed through the Map-based registries without triggering prototype
diff --git a/superset-frontend/src/middleware/asyncEvent.ts b/superset-frontend/src/middleware/asyncEvent.ts
index be68ff9f211..b22a34d3f1b 100644
--- a/superset-frontend/src/middleware/asyncEvent.ts
+++ b/superset-frontend/src/middleware/asyncEvent.ts
@@ -58,6 +58,9 @@ const LOCALSTORAGE_KEY = 'last_async_event_id';
const POLLING_URL = '/api/v1/async_event/';
const MAX_RETRIES = 6;
const RETRY_DELAY = 100;
+// Cap for the exponential backoff applied when polling requests fail
+// repeatedly (e.g. expired session, server or network errors)
+const MAX_ERROR_POLLING_DELAY_MS = 60000;
let config: AppConfig;
let transport: string;
@@ -66,6 +69,11 @@ let pollingTimeoutId: number;
let listenersByJobId: Map<string, ListenerFn>;
let retriesByJobId: Map<string, number>;
let lastReceivedEventId: string | null | undefined;
+let consecutivePollingErrorCount = 0;
+// Incremented on every init() so polling invocations that are already
+// awaiting a fetch when re-init happens can detect they are stale and
+// stop, instead of mutating fresh state or scheduling a second loop
+let pollingGeneration = 0;
const addListener = (id: string, fn: ListenerFn) => {
listenersByJobId.set(id, fn);
@@ -170,19 +178,34 @@ export const processEvents = async (events: AsyncEvent[]) => {
});
};
+const getPollingDelay = () => {
+ if (!consecutivePollingErrorCount) return pollingDelayMs;
+ const backoffDelayMs = pollingDelayMs * 2 ** consecutivePollingErrorCount;
+ return Math.max(
+ pollingDelayMs,
+ Math.min(backoffDelayMs, MAX_ERROR_POLLING_DELAY_MS),
+ );
+};
+
const loadEventsFromApi = async () => {
+ const generation = pollingGeneration;
const eventArgs = lastReceivedEventId ? { last_id: lastReceivedEventId } : {};
if (listenersByJobId.size) {
try {
const { result: events } = await fetchEvents(eventArgs);
+ if (generation !== pollingGeneration) return;
+ consecutivePollingErrorCount = 0;
if (events?.length) await processEvents(events);
} catch (err) {
+ if (generation !== pollingGeneration) return;
+ consecutivePollingErrorCount += 1;
logging.warn(err);
}
}
+ if (generation !== pollingGeneration) return;
if (transport === TRANSPORT_POLLING) {
- pollingTimeoutId = window.setTimeout(loadEventsFromApi, pollingDelayMs);
+ pollingTimeoutId = window.setTimeout(loadEventsFromApi, getPollingDelay());
}
};
@@ -232,12 +255,14 @@ const wsConnect = (): void => {
};
export const init = (appConfig?: AppConfig) => {
- if (!isFeatureEnabled(FeatureFlag.GlobalAsyncQueries)) return;
+ pollingGeneration += 1;
if (pollingTimeoutId) clearTimeout(pollingTimeoutId);
+ if (!isFeatureEnabled(FeatureFlag.GlobalAsyncQueries)) return;
listenersByJobId = new Map();
retriesByJobId = new Map();
lastReceivedEventId = null;
+ consecutivePollingErrorCount = 0;
config = appConfig || getBootstrapData().common.conf;
transport = config.GLOBAL_ASYNC_QUERIES_TRANSPORT || TRANSPORT_POLLING;