sadpandajoe commented on code in PR #43407:
URL: https://github.com/apache/superset/pull/43407#discussion_r3886731097
##########
superset-frontend/src/middleware/asyncEvent.ts:
##########
@@ -16,327 +16,386 @@
* specific language governing permissions and limitations
* under the License.
*/
+/**
+ * Await completion of asynchronous chart-data queries (GLOBAL_ASYNC_QUERIES).
+ *
+ * A 202 from POST /chart/data carries the GTF tasks the query runs as. Their
+ * completion is learned two ways: the shared realtime socket
+ * (src/middleware/realtime.ts) delivers per-principal status events, and a
+ * single shared poll of /task/status_changes runs while anything is awaited.
The
+ * socket is best-effort and only accelerates things; the poll is the
correctness
+ * backstop.
+ */
import {
- ensureIsArray,
isFeatureEnabled,
FeatureFlag,
makeApi,
SupersetClient,
- getClientErrorObject,
- parseErrorJson,
- SupersetError,
} from '@superset-ui/core';
import { logging } from '@apache-superset/core/utils';
import getBootstrapData from 'src/utils/getBootstrapData';
-
-type AsyncEvent = {
- id?: string | null;
- channel_id: string;
- job_id: string;
- user_id?: string;
- status: string;
- errors?: SupersetError[];
- result_url: string | null;
+import {
+ connectRealtime,
+ subscribeRealtime,
+ type RealtimeMessage,
+} from 'src/middleware/realtime';
+
+// The GTF task type chart-data queries run under (see
+// superset/tasks/async_queries.py CHART_QUERY_TASK). Polling is filtered to
this
+// type so a dashboard only tracks its own chart-data work, not every task.
+const CHART_QUERY_TASK_TYPE = 'superset.query_object_v1';
+const STATUS_CHANGES_URL = '/api/v1/task/status_changes';
+
+// Terminal GTF task statuses (mirror superset_core.tasks.types.TaskStatus).
+const STATUS_SUCCESS = 'success';
+const TERMINAL_STATUSES = new Set([
+ STATUS_SUCCESS,
+ 'failure',
+ 'aborted',
+ 'timed_out',
+]);
+
+type TaskStatusChange = { status: string; progress: number | null };
+type StatusChangesResponse = {
+ statuses: Record<string, TaskStatusChange>;
+ cursor: string | null;
};
-type CachedDataResponse = {
- status: string;
- data: any;
+// The 202 body from POST /chart/data when async: the query tasks to await,
plus
+// a status-poll cursor captured server-side *before* the tasks were created,
so
+// polling from it can never skip a task's terminal transition.
+export type AsyncJob = { task_ids: string[]; cursor?: string | null };
+
+type AppConfig = {
+ WEBSOCKET_ENABLE?: boolean;
+ WEBSOCKET_URL?: string;
+ GLOBAL_ASYNC_QUERIES_POLLING_DELAY?: number;
+ GLOBAL_ASYNC_QUERIES_POLLING_MAX_DELAY?: number;
+ GLOBAL_ASYNC_QUERIES_POLLING_STALE_TIMEOUT?: number;
};
-type AppConfig = Record<string, any>;
-type ListenerFn = (asyncEvent: AsyncEvent) => Promise<any>;
-const TRANSPORT_POLLING = 'polling';
-const TRANSPORT_WS = 'ws';
-const JOB_STATUS = {
- PENDING: 'pending',
- RUNNING: 'running',
- ERROR: 'error',
- DONE: 'done',
+type Waiter = {
+ taskIds: string[];
+ pending: Set<string>;
+ failed: boolean;
+ // Re-issue the original chart-data request once every task has succeeded;
the
+ // per-query DATA cache is now warm, so it returns synchronously (200).
+ resolve: () => void;
+ reject: (error: unknown) => void;
+ signal?: AbortSignal;
+ onAbort?: () => void;
};
-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;
let pollingDelayMs: number;
+// Backoff state: the poll starts eager (`pollingDelayMs`), degrades —
doubling up
+// to `pollBackoffMaxMs` — while awaited tasks are quiet, and snaps back to
eager
+// the moment an awaited task changes or a new waiter registers.
+let currentPollDelayMs: number;
+let pollBackoffMaxMs: number;
+// Give-up guard: if no awaited task makes progress for `pollStaleTimeoutMs`,
the
+// poll abandons its waiters (rejecting them) so a stuck/orphaned task can't
spin a
+// chart — or the poll — forever. `lastProgressAt` is the epoch ms of the last
+// progress (or waiter registration); it resets on any awaited-task change.
+let pollStaleTimeoutMs: number;
+let lastProgressAt: number;
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
+// Whether the poll loop is running. It stops entirely when no waiters remain
(no
+// idle heartbeat) and is restarted by ``ensurePolling`` when a waiter
registers.
+let pollingActive = false;
+// Registry of in-flight waiters keyed by every task uuid they await, so a
single
+// shared poll loop fans status changes out to whichever requests are awaiting
them.
+// A SHARED task can be deduplicated across concurrent chart requests, so each
task
+// id maps to a *set* of waiters (never overwrite an earlier subscriber).
+// Initialized eagerly (not just in init()): the shared realtime socket
connects
+// whenever WEBSOCKET_ENABLE — independent of GLOBAL_ASYNC_QUERIES — so the
+// subscribed handler may run applyStatus even when async queries are off, and
+// must find a map rather than undefined.
+let waitersByTaskId: Map<string, Set<Waiter>> = new Map();
+// Server-issued watermark: seeded from a chart request's 202 pre-task cursor
+// and advanced by each poll. Always the server's own clock, never the
browser's.
+let cursor: string | null;
+// Incremented on every init() so an in-flight poll can detect it is stale and
+// stop instead of scheduling a second loop or mutating fresh state.
let pollingGeneration = 0;
-const addListener = (id: string, fn: ListenerFn) => {
- listenersByJobId.set(id, fn);
+// A poll from a superseded init() must abandon its tick: stop the loop and
leave
+// the fresh generation's state alone. Returns whether the tick was abandoned.
+const stopIfStale = (generation: number): boolean => {
+ if (generation === pollingGeneration) return false;
+ pollingActive = false;
+ return true;
};
-const removeListener = (id: string) => {
- if (!listenersByJobId.has(id)) return;
- listenersByJobId.delete(id);
-};
+// Browser channel prefix for per-principal messages emitted by
+// superset-websocket after it fans out backend task-status events. The browser
+// socket is JWT-bound to its own principal routing key, so it only ever
receives
+// its own realtime messages.
+const REALTIME_CHANNEL_PREFIX = 'realtime:';
-const fetchCachedData = async (
- asyncEvent: AsyncEvent,
- signal?: AbortSignal,
-): Promise<CachedDataResponse> => {
- let status = 'success';
- let data;
- try {
- const { json } = await SupersetClient.get({
- endpoint: String(asyncEvent.result_url),
- signal,
- });
- data = 'result' in json ? json.result : json;
- } catch (response) {
- status = 'error';
- data = await getClientErrorObject(response);
- }
-
- return { status, data };
-};
+const fetchStatusChanges = makeApi<
+ { cursor?: string | null; task_type: string },
+ StatusChangesResponse
+>({
+ method: 'GET',
+ endpoint: STATUS_CHANGES_URL,
+});
-const cancelAsyncJob = (jobId: string) => {
- // Best-effort server-side cancel; the request stops the running Celery task
- // so it no longer consumes warehouse resources. Failures are non-fatal: the
- // client has already stopped waiting on the job.
+const cancelTask = (taskId: string) => {
+ // Best-effort task abort/unsubscribe. This can prevent pending work from
+ // starting, but chart tasks do not cancel an underlying warehouse query
after
+ // execution starts. Failures are non-fatal: the client has stopped waiting.
SupersetClient.post({
- endpoint: `/api/v1/async_event/${jobId}/cancel`,
+ endpoint: `/api/v1/task/${taskId}/cancel`,
}).catch(error => {
- logging.warn('Failed to cancel async job', jobId, error);
+ logging.warn('Failed to cancel task', taskId, error);
});
};
-export const waitForAsyncData = async (
- asyncResponse: AsyncEvent,
- signal?: AbortSignal,
-) =>
- new Promise((resolve, reject) => {
- const jobId = asyncResponse.job_id;
-
- let onAbort: (() => void) | undefined;
- const cleanup = () => {
- removeListener(jobId);
- if (onAbort && signal) {
- signal.removeEventListener('abort', onAbort);
- }
- };
-
- // Bail immediately if the caller has already aborted (e.g. the chart was
- // unmounted before the job started), avoiding a leaked listener.
- if (signal?.aborted) {
- cancelAsyncJob(jobId);
- reject(new DOMException('Aborted', 'AbortError'));
- return;
- }
-
- const listener = async (asyncEvent: AsyncEvent) => {
- switch (asyncEvent.status) {
- case JOB_STATUS.DONE: {
- // Forward the signal so the cached-result download is cancelled too
if
- // the caller aborts mid-fetch, rather than wasting
network/processing.
- let { data, status } = await fetchCachedData(asyncEvent, signal); //
eslint-disable-line prefer-const
- data = ensureIsArray(data);
- if (status === 'success') {
- resolve(data);
- } else {
- reject(data);
- }
- // Terminal status: the promise is settled, so fully clean up.
- cleanup();
- break;
- }
- case JOB_STATUS.ERROR: {
- const err = parseErrorJson(asyncEvent);
- reject(err);
- // Terminal status: the promise is settled, so fully clean up.
- cleanup();
- break;
- }
- default: {
- // Non-terminal status (e.g., 'pending', 'running'): keep the
listener
- // registered so it can receive the eventual terminal event ('done',
'error').
- // Only cleanup happens on terminal states or abort.
- logging.info(
- 'received non-terminal event with status',
- asyncEvent.status,
- );
- }
- }
- };
-
- // When the caller aborts (Stop pressed, chart superseded/unmounted), stop
- // listening so the listener and its retained closure don't leak, and ask
the
- // server to cancel the job so it stops consuming warehouse resources.
- if (signal) {
- onAbort = () => {
- cleanup();
- cancelAsyncJob(jobId);
- reject(new DOMException('Aborted', 'AbortError'));
- };
- signal.addEventListener('abort', onAbort, { once: true });
- }
-
- addListener(jobId, listener);
+// Drop a waiter from the registry entry of every task it was awaiting, so a
+// settled/aborted waiter never leaks and completion of one task can't
re-touch it.
+const unregister = (waiter: Waiter) => {
+ waiter.taskIds.forEach(taskId => {
+ const waiters = waitersByTaskId.get(taskId);
+ if (!waiters) return;
+ waiters.delete(waiter);
+ if (waiters.size === 0) waitersByTaskId.delete(taskId);
});
+};
-const fetchEvents = makeApi<
- { last_id?: string | null },
- { result: AsyncEvent[] }
->({
- method: 'GET',
- endpoint: POLLING_URL,
-});
-
-const setLastId = (asyncEvent: AsyncEvent) => {
- lastReceivedEventId = asyncEvent.id;
- try {
- localStorage.setItem(LOCALSTORAGE_KEY, lastReceivedEventId as string);
- } catch (err) {
- logging.warn('Error saving event Id to localStorage', err);
+const settle = (waiter: Waiter, error?: unknown) => {
+ unregister(waiter);
+ if (waiter.signal && waiter.onAbort) {
+ waiter.signal.removeEventListener('abort', waiter.onAbort);
+ }
+ if (error !== undefined) {
+ waiter.reject(error);
+ } else if (waiter.failed) {
+ waiter.reject(
+ new Error('One or more chart-data queries failed'), // surfaced via
getClientErrorObject
+ );
+ } else {
+ waiter.resolve();
}
};
-export const processEvents = async (events: AsyncEvent[]) => {
- events.forEach((asyncEvent: AsyncEvent) => {
- const jobId = asyncEvent.job_id;
- const listener = listenersByJobId.get(jobId);
- // `jobId` originates from server/WebSocket payloads, so the listener is
- // resolved exclusively through a Map (never plain-object property access,
- // which would expose the prototype chain), and we confirm the retrieved
- // value is a registered function before dispatching the event to it.
- if (typeof listener === 'function') {
- listener(asyncEvent);
- retriesByJobId.delete(jobId);
- } else {
- // handle race condition where event is received
- // before listener is registered
- const retries = (retriesByJobId.get(jobId) ?? 0) + 1;
- retriesByJobId.set(jobId, retries);
-
- if (retries <= MAX_RETRIES) {
- setTimeout(() => {
- processEvents([asyncEvent]);
- }, RETRY_DELAY * retries);
- } else {
- retriesByJobId.delete(jobId);
- logging.warn('listener not found for job_id', asyncEvent.job_id);
- }
- }
- setLastId(asyncEvent);
- });
+// Give up on every still-pending waiter (rejecting them), so a stuck/orphaned
+// task surfaces an error instead of spinning forever. Collect first — settle()
+// mutates waitersByTaskId as it unregisters. Emptying the registry lets the
poll
+// loop stop on its next tick.
+const abandonPolling = () => {
+ const stranded = new Set<Waiter>();
+ waitersByTaskId.forEach(waiters => waiters.forEach(w => stranded.add(w)));
+ stranded.forEach(waiter =>
+ settle(waiter, new Error('Timed out waiting for chart-data query
results')),
+ );
};
-const getPollingDelay = () => {
- if (!consecutivePollingErrorCount) return pollingDelayMs;
- const backoffDelayMs = pollingDelayMs * 2 ** consecutivePollingErrorCount;
- return Math.max(
- pollingDelayMs,
- Math.min(backoffDelayMs, MAX_ERROR_POLLING_DELAY_MS),
- );
+const applyStatus = (taskId: string, status: string) => {
+ const waiters = waitersByTaskId.get(taskId);
+ if (!waiters || !TERMINAL_STATUSES.has(status)) return;
+ // Settle every request awaiting this task, not just the most recent one.
+ [...waiters].forEach(waiter => {
+ waiter.pending.delete(taskId);
+ if (status !== STATUS_SUCCESS) waiter.failed = true;
+ if (waiter.pending.size === 0) settle(waiter);
+ });
+ waitersByTaskId.delete(taskId);
};
-const loadEventsFromApi = async () => {
- const generation = pollingGeneration;
- const eventArgs = lastReceivedEventId ? { last_id: lastReceivedEventId } :
{};
- if (listenersByJobId.size) {
+const loadStatusChanges = async (generation: number) => {
+ if (stopIfStale(generation)) return;
+ if (waitersByTaskId.size) {
try {
- const { result: events } = await fetchEvents(eventArgs);
- if (generation !== pollingGeneration) return;
- consecutivePollingErrorCount = 0;
- if (events?.length) await processEvents(events);
+ const { statuses, cursor: next } = await fetchStatusChanges({
+ cursor,
+ task_type: CHART_QUERY_TASK_TYPE,
+ });
+ if (stopIfStale(generation)) return;
+ cursor = next;
+ // "Progress" = the batch carried a change for a task we're awaiting;
check
+ // membership before applyStatus, which deletes a settled task's waiters.
+ let progressed = false;
+ Object.entries(statuses).forEach(([taskId, { status }]) => {
+ if (waitersByTaskId.has(taskId)) progressed = true;
+ applyStatus(taskId, status);
+ });
+ if (progressed) {
+ // Reset to eager polling and restart the give-up clock on any change.
+ currentPollDelayMs = pollingDelayMs;
+ lastProgressAt = Date.now();
+ } else {
+ // No progress: give up if we've been stale too long (a stuck/orphaned
+ // task), else back off (bounded).
+ if (Date.now() - lastProgressAt >= pollStaleTimeoutMs) {
+ abandonPolling();
+ pollingActive = false;
+ return;
+ }
+ currentPollDelayMs = Math.min(currentPollDelayMs * 2,
pollBackoffMaxMs);
+ }
} catch (err) {
- if (generation !== pollingGeneration) return;
- consecutivePollingErrorCount += 1;
+ if (stopIfStale(generation)) return;
logging.warn(err);
}
}
+ // Reschedule from the tail so a slow request never overlaps the next tick.
+ // Nothing left to await → stop the loop entirely (no idle heartbeat); the
next
+ // waiter restarts it from the eager interval via ensurePolling.
+ if (!waitersByTaskId.size) {
+ pollingActive = false;
+ currentPollDelayMs = pollingDelayMs;
+ return;
+ }
+ pollingActive = true;
+ pollingTimeoutId = window.setTimeout(
+ () => loadStatusChanges(generation),
+ currentPollDelayMs,
+ );
+};
- if (generation !== pollingGeneration) return;
- if (transport === TRANSPORT_POLLING) {
- pollingTimeoutId = window.setTimeout(loadEventsFromApi, getPollingDelay());
+// Start (or wake) the poll loop for a freshly registered waiter: poll eagerly
+// again, and kick the loop if it had gone idle. Idempotent — a no-op while the
+// loop is already running or when async queries are disabled.
+const ensurePolling = () => {
+ currentPollDelayMs = pollingDelayMs;
+ lastProgressAt = Date.now(); // a fresh request restarts the give-up clock
+ if (!pollingActive && isFeatureEnabled(FeatureFlag.GlobalAsyncQueries)) {
+ pollingActive = true;
+ loadStatusChanges(pollingGeneration);
}
};
-const wsConnectMaxRetries = 6;
-const wsConnectErrorDelay = 2500;
-let wsConnectRetries = 0;
-let wsConnectTimeout: any;
-let ws: WebSocket;
+/**
+ * Handle a realtime message from the shared client.
+ *
+ * A per-principal chart-data payload is ``{task_id, status}``; because
delivery
+ * is scoped to this principal's own JWT-bound channel the status is
+ * authoritative enough to settle the waiter immediately (the ensuing
``refetch``
+ * reads the authorized per-query cache anyway). Other channels (e.g. the
+ * ``entity-changes:*`` list-view nudges) are not chart-data's concern.
+ */
+export const handleRealtimeMessage = (message: RealtimeMessage) => {
+ const { channel, payload } = message;
+ if (!channel.startsWith(REALTIME_CHANNEL_PREFIX)) return;
+ if (!payload || typeof payload !== 'object') return;
+ const { task_id: taskId, status } = payload as {
+ task_id?: unknown;
+ status?: unknown;
+ };
+ if (typeof taskId === 'string' && typeof status === 'string') {
+ applyStatus(taskId, status);
+ }
+};
-const wsConnect = (): void => {
- let url = config.GLOBAL_ASYNC_QUERIES_WEBSOCKET_URL;
- if (lastReceivedEventId) url += `?last_id=${lastReceivedEventId}`;
- ws = new WebSocket(url);
+// The handler reads the live waiter registry, so it stays correct across
init()
+// generations.
+subscribeRealtime(handleRealtimeMessage);
- ws.addEventListener('open', () => {
- logging.log('WebSocket connected');
- clearTimeout(wsConnectTimeout);
- wsConnectRetries = 0;
- });
+/**
+ * Await completion of an async chart-data job's query tasks, then re-issue the
+ * original request to read the now-cached results.
+ *
+ * Resolves with the fresh `QueryData[]` once every task has succeeded (the
+ * caller's `refetch` returns synchronously from the warm per-query cache);
+ * rejects if any task ends in a non-success terminal state, or with an
+ * AbortError if the caller aborts (which also cancels the outstanding tasks).
+ */
+export const waitForAsyncData = async <T = unknown[]>(
+ asyncJob: AsyncJob,
+ refetch: () => Promise<T>,
+ signal?: AbortSignal,
+): Promise<T> => {
+ const taskIds = asyncJob.task_ids ?? [];
- ws.addEventListener('close', () => {
- wsConnectTimeout = setTimeout(() => {
- wsConnectRetries += 1;
- if (wsConnectRetries <= wsConnectMaxRetries) {
- wsConnect();
- } else {
- logging.warn('WebSocket not available, falling back to async polling');
- loadEventsFromApi();
+ // Register the waiter synchronously, in the same tick the 202 was received —
+ // NOT after an await — so a completion socket event can't arrive before the
+ // waiter exists and be dropped.
+ await new Promise<void>((resolve, reject) => {
+ if (signal?.aborted) {
+ taskIds.forEach(cancelTask);
+ reject(new DOMException('Aborted', 'AbortError'));
+ return;
+ }
+ const waiter: Waiter = {
+ taskIds,
+ pending: new Set(taskIds),
+ failed: false,
+ resolve,
+ reject,
+ signal,
+ };
+ if (signal) {
+ waiter.onAbort = () => {
+ unregister(waiter);
+ taskIds.forEach(cancelTask);
Review Comment:
This cancels the shared task after removing only this local waiter. Two
charts for the same principal can join one shared task but map to one backend
subscriber, so aborting either chart makes the server see the last subscriber
and aborts work the other chart still needs. Could cancellation be deferred
while another local waiter remains?
##########
superset/config.py:
##########
@@ -2909,50 +2919,87 @@ def EMAIL_HEADER_MUTATOR( # pylint:
disable=invalid-name,unused-argument # noq
# Global async query config options.
-# Requires GLOBAL_ASYNC_QUERIES feature flag to be enabled.
-GLOBAL_ASYNC_QUERY_MANAGER_CLASS = (
- "superset.async_events.async_query_manager.AsyncQueryManager"
-)
-GLOBAL_ASYNC_QUERIES_REDIS_STREAM_PREFIX = "async-events-"
-GLOBAL_ASYNC_QUERIES_REDIS_STREAM_LIMIT = 1000
-GLOBAL_ASYNC_QUERIES_REDIS_STREAM_LIMIT_FIREHOSE = 1000000
-GLOBAL_ASYNC_QUERIES_REGISTER_REQUEST_HANDLERS = True
-GLOBAL_ASYNC_QUERIES_JWT_COOKIE_NAME = "async-token"
-GLOBAL_ASYNC_QUERIES_JWT_COOKIE_SECURE = False
-GLOBAL_ASYNC_QUERIES_JWT_COOKIE_SAMESITE: None | (Literal["None", "Lax",
"Strict"]) = (
- None
-)
-GLOBAL_ASYNC_QUERIES_JWT_COOKIE_DOMAIN = None
-GLOBAL_ASYNC_QUERIES_JWT_SECRET = CHANGE_ME_GLOBAL_ASYNC_QUERIES_JWT_SECRET
-# Lifetime of the async-query JWT, in seconds. After this period the token
-# expires and a fresh one is issued on the next request.
-GLOBAL_ASYNC_QUERIES_JWT_EXPIRATION_SECONDS =
int(timedelta(hours=1).total_seconds())
-GLOBAL_ASYNC_QUERIES_TRANSPORT: Literal["polling", "ws"] = "polling"
+# Requires the GLOBAL_ASYNC_QUERIES feature flag to be enabled. Async
chart-data
+# queries run on the Global Task Framework (one task per QueryObject) over
+# DISTRIBUTED_COORDINATION_CONFIG; the client polls
/api/v1/task/status_changes at
+# this interval (milliseconds) and re-issues its request once the tasks
succeed.
GLOBAL_ASYNC_QUERIES_POLLING_DELAY = int(
timedelta(milliseconds=500).total_seconds() * 1000
)
-GLOBAL_ASYNC_QUERIES_WEBSOCKET_URL = "ws://127.0.0.1:8080/"
-
-# Global async queries cache backend configuration options:
-# - Set 'CACHE_TYPE' to 'RedisCache' for RedisCacheBackend.
-# - Set 'CACHE_TYPE' to 'RedisSentinelCache' for RedisSentinelCacheBackend.
-GLOBAL_ASYNC_QUERIES_CACHE_BACKEND = {
- "CACHE_TYPE": "RedisCache",
- "CACHE_REDIS_HOST": "localhost",
- "CACHE_REDIS_PORT": 6379,
- "CACHE_REDIS_USER": "",
- "CACHE_REDIS_PASSWORD": "",
- "CACHE_REDIS_DB": 0,
- "CACHE_DEFAULT_TIMEOUT": 300,
- "CACHE_REDIS_SENTINELS": [("localhost", 26379)],
- "CACHE_REDIS_SENTINEL_MASTER": "mymaster",
- "CACHE_REDIS_SENTINEL_PASSWORD": None,
- "CACHE_REDIS_SSL": False, # True or False
- "CACHE_REDIS_SSL_CERTFILE": None,
- "CACHE_REDIS_SSL_KEYFILE": None,
- "CACHE_REDIS_SSL_CERT_REQS": "required",
- "CACHE_REDIS_SSL_CA_CERTS": None,
-}
+
+# Ceiling (milliseconds) for the status-poll interval. The client polls
eagerly at
+# GLOBAL_ASYNC_QUERIES_POLLING_DELAY, then backs off exponentially while the
tasks
+# it is awaiting stay quiet — up to this maximum — snapping back to eager the
moment
+# an awaited task changes.
+GLOBAL_ASYNC_QUERIES_POLLING_MAX_DELAY = int(
+ timedelta(seconds=30).total_seconds() * 1000
+)
+
+# How long (milliseconds) the client keeps polling with no progress on the
tasks it
+# is awaiting before it gives up and surfaces an error. Guards against a stuck
or
+# orphaned task (e.g. a worker killed mid-execution) keeping a chart spinning
— and
+# the poll running — forever. The clock resets whenever an awaited task
changes, so
+# steady progress is never interrupted.
+GLOBAL_ASYNC_QUERIES_POLLING_STALE_TIMEOUT = int(
+ timedelta(minutes=10).total_seconds() * 1000
+)
+
+# Minimum cache TTL (seconds) for chart-data results produced by an *async*
+# request. The async flow caches each query's result and the client then
+# re-issues the request to read it back, so a cache TTL shorter than the round
+# trip could evict the result before it is fetched, hanging the chart. When a
+# query runs async, its result-cache TTL is floored to this value (a longer
+# slice/dataset/deployment TTL is kept as-is, and 0 — "cache forever" — is left
+# untouched). This floor applies ONLY to async execution; synchronous
+# ``/chart/data`` requests are unaffected even when GLOBAL_ASYNC_QUERIES is on.
+GLOBAL_ASYNC_QUERIES_MIN_CACHE_TTL = int(timedelta(minutes=5).total_seconds())
+
+# Timeout (seconds) for an async chart-data query task. When reached, the task
is
+# aborted; on engines that support query cancellation the abort handler also
+# cancels the underlying warehouse query (over a fresh connection), so the task
+# ends promptly as TIMED_OUT. On engines without cancel support the query is
not
+# interrupted (the task is freed once the query returns on its own). Default
None
+# leaves async chart-data queries unbounded (matching prior behavior); set an
int
+# to enforce a ceiling.
+GLOBAL_ASYNC_QUERIES_QUERY_TIMEOUT: int | None = None
+
+# Deployment default for whether the UI runs chart-data queries asynchronously
when
+# GLOBAL_ASYNC_QUERIES is enabled. This is a FRONTEND-ONLY policy input: async
is
+# opt-in per request via an ``async_mode`` flag on ``/chart/data`` (an absent
flag is
+# treated as synchronous, so programmatic API clients keep the synchronous 200
flow),
+# and the frontend resolves the flag it sends via a policy chain —
per-dashboard
+# override → this default → the feature-flag gate. Default ``True`` preserves
the UI's
+# existing async behavior; set ``False`` to make the UI synchronous by default
and roll
+# async out per dashboard.
+GLOBAL_ASYNC_QUERIES_DEFAULT = True
+
+# Realtime websocket transport (the `superset-websocket` server) config.
+# When enabled, GTF task changes are pushed to the browser so charts and list
+# views update without waiting for the interval poll (which stays as the
+# fallback). Requires the superset-websocket server, a Redis coordination
+# backend (DISTRIBUTED_COORDINATION_CONFIG), and `can_read` on `Realtime`.
+# Two channel tiers:
+# - an authenticated broadcast per-entity-type pub/sub
+# (e.g. entity-changes:task) for lossy list-view activity (opaque entity
+# ids only), and
+# - targeted task-status pub/sub messages for the dashboard chart-data path,
+# fanned out by the websocket server to JWT-bound principal sockets.
+# The JWT authenticates the socket connection and binds it to its channel; the
+# server delivers targeted task-status events only to matching principal
+# sockets. Set a strong random WEBSOCKET_JWT_SECRET (>= 32 bytes) in
production.
+# The websocket server can be configured with a previous validation secret
+# during rotations; the Flask app always mints new cookies with the current
key.
+# The websocket server validates the signed token at connection time and
+# terminates sockets after JWT expiry, so post-mint permission revocation is
+# bounded by this lifetime plus the server ping interval.
+WEBSOCKET_ENABLE = False
Review Comment:
Helm still renders only the removed `GLOBAL_ASYNC_QUERIES_*` websocket
settings, while the new defaults use `WEBSOCKET_ENABLE`, `WEBSOCKET_URL`, and
`WEBSOCKET_JWT_*`. A Helm deployment with `supersetWebsockets.enabled`
therefore leaves the new feature disabled. Could the Helm template and its
coverage be migrated to the new configuration keys?
##########
UPDATING.md:
##########
@@ -24,6 +24,110 @@ assists people when migrating to a new version.
## Next
+### Global Async Queries re-platformed onto the Global Task Framework
(breaking)
+
+Global Async Queries (GAQ) no longer runs on its own bespoke async-events
+plumbing. Async chart data is now executed as Global Task Framework (GTF) tasks
+(one task per `QueryObject`), the browser learns of completion by polling
+`GET /api/v1/task/status_changes` (optionally accelerated by the WebSocket
+transport below) and re-issuing the original `/chart/data` request against the
+now-warm per-query cache, and the realtime WebSocket server is a generic,
+feature-agnostic task push transport rather than a GAQ-specific event tail.
+
+Breaking removals (no deprecation window):
+
+- The `/api/v1/async_event/` REST API, `AsyncQueryManager`, and the
+ `qc-<hash>` query-context descriptor replay endpoint
+ (`GET /api/v1/chart/data/<cache_key>`) are removed. Any client that consumed
a
+ `result_url` from a `202` response must move to the re-request model (the
+ built-in frontend already does).
+- The following config keys are removed: `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND`,
+ `GLOBAL_ASYNC_QUERIES_TRANSPORT`, `GLOBAL_ASYNC_QUERIES_WEBSOCKET_URL`,
+ `GLOBAL_ASYNC_QUERIES_REDIS_STREAM_PREFIX`,
+ `GLOBAL_ASYNC_QUERIES_REDIS_STREAM_LIMIT`,
+ `GLOBAL_ASYNC_QUERIES_REDIS_STREAM_LIMIT_FIREHOSE`,
+ `GLOBAL_ASYNC_QUERIES_REGISTER_REQUEST_HANDLERS`,
+ `GLOBAL_ASYNC_QUERIES_JWT_*`, and
+ `GLOBAL_ASYNC_QUERY_MANAGER_CLASS`. The coordinator (locks, GTF, and now GAQ)
+ uses `DISTRIBUTED_COORDINATION_CONFIG` exclusively.
+
+Enabling async chart data in the new flow:
+
+```python
+# feature flag: makes async chart data available (auto-enables
GLOBAL_TASK_FRAMEWORK)
+FEATURE_FLAGS = {"GLOBAL_ASYNC_QUERIES": True}
+
+# a Redis connection for distributed coordination (locks, GTF signalling,
+# and the realtime pub/sub); required for async execution in production
+DISTRIBUTED_COORDINATION_CONFIG = {
+ "CACHE_TYPE": "RedisCache",
+ "CACHE_REDIS_HOST": "localhost",
+ "CACHE_REDIS_PORT": 6379,
+ "CACHE_REDIS_DB": 0,
+}
+```
+
+Async is now **opt-in per request**: `GLOBAL_ASYNC_QUERIES` only makes async
+*available*; whether a given `/chart/data` request runs async is decided by an
+`async_mode` request flag (endpoint default `false`, so programmatic API
clients
+keep the synchronous `200` flow unless they opt in). The built-in frontend
+resolves the `async_mode` it sends from a policy chain — per-dashboard
override →
+deployment default `GLOBAL_ASYNC_QUERIES_DEFAULT` (default `true`) → the
feature
+flag — so the UI keeps its existing async behavior by default.
+
+Enabling the realtime WebSocket transport (optional; accelerates completion,
the
+`status_changes` interval poll remains the correctness backstop):
+
+> **Note:** the realtime WebSocket transport is opt-in (`WEBSOCKET_ENABLE`
+> defaults to `False`) and should be treated as experimental. Chart-data
+> completion correctness does not depend on it — the `status_changes` interval
+> poll is the source of truth — so it can be enabled or left off without
+> affecting async chart results.
+
+```python
+WEBSOCKET_ENABLE = True
+WEBSOCKET_URL = "ws://<same-host>:8080/"
+WEBSOCKET_JWT_SECRET = "<output of: openssl rand -base64 42>"
+```
+
+The built-in Gamma role receives `can_read Realtime`; grant that permission to
+custom roles that should receive websocket notifications.
+
+Run the `superset-websocket` Node server on the **same browser-visible host**
+(so its JWT channel cookie is shared) and point its `redis` config at the same
+instance as `DISTRIBUTED_COORDINATION_CONFIG`, plus `jwtSecret` /
+`jwtCookieName` matching the Flask config (`WEBSOCKET_JWT_SECRET` /
+`WEBSOCKET_JWT_COOKIE_NAME`, default `superset-ws-token`). During websocket JWT
+secret rotation, set the websocket server's `previousJwtSecret` /
+`PREVIOUS_JWT_SECRET` to the old key while Flask continues minting cookies with
+`WEBSOCKET_JWT_SECRET`. The server is bundled in the official Superset image
+and launched via an alternate entrypoint — no separate image is required:
+`docker run <superset-image> /app/docker/entrypoints/run-websocket.sh` (or the
+opt-in `websocket` profile in `docker compose`). It now consumes Redis Pub/Sub
Review Comment:
The server subscribes to Redis `entity-changes:*` and `task-status`;
`realtime:<channel_id>` is generated only for browser delivery after fanout.
Operators following this ACL guidance can omit `task-status`, which prevents
task updates from reaching clients. Could the documented Pub/Sub channels name
the server-side subscriptions instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
