villebro commented on code in PR #43407:
URL: https://github.com/apache/superset/pull/43407#discussion_r3886947817
##########
superset/tasks/async_queries.py:
##########
@@ -16,126 +16,312 @@
# under the License.
from __future__ import annotations
-import dataclasses
import logging
-from typing import Any, TYPE_CHECKING
+from contextlib import contextmanager
+from datetime import datetime
+from typing import Any, Iterator, TYPE_CHECKING
+from uuid import UUID
-from celery.exceptions import SoftTimeLimitExceeded
from flask import current_app
from flask_appbuilder.security.sqla.models import User
-from marshmallow import ValidationError
+from superset_core.tasks.types import TaskOptions, TaskScope
-from superset.charts.data.form_data import set_form_data
-from superset.charts.schemas import ChartDataQueryContextSchema
-from superset.exceptions import (
- SupersetErrorException,
- SupersetErrorsException,
+from superset.common.query_serialization import (
+ load_serialized_query,
+ serialize_query,
+ SerializedQuery,
)
+from superset.constants import CacheRegion
+from superset.exceptions import SupersetException
from superset.extensions import (
- async_query_manager,
- celery_app,
security_manager,
)
+from superset.tasks.ambient_context import get_context
+from superset.tasks.decorators import task
+from superset.tasks.query_cancel import (
+ cancel_chart_query,
+ capture_cancel_id,
+ capture_cancel_query_id,
+)
from superset.utils.core import override_user
-from superset.utils.error_sanitization import sanitize_error_dicts
if TYPE_CHECKING:
+ from superset_core.tasks.models import Task as CoreTask
+
from superset.common.query_context import QueryContext
+ from superset.common.query_object import QueryObject
+ from superset.models.tasks import Task
+ from superset.security.guest_token import GuestToken
logger = logging.getLogger(__name__)
-query_timeout = current_app.config[
- "SQLLAB_ASYNC_TIME_LIMIT_SEC"
-] # TODO: new config key
+# GTF task type for the chart-data fan-out. Each QueryObject runs as its own
SHARED
+# task keyed by its query_cache_key (safe cross-user dedup — the key encodes
+# RLS/impersonation). The client polls /api/v1/task/status_changes (filtered
to this
+# type) and aggregates the tasks' statuses itself; GTF owns completion
emission (there
+# is no coordinator task). The atomic unit is a QueryObject, not
chart-specific, so the
+# type is versioned to allow the serialization/execution contract to evolve.
+CHART_QUERY_TASK = "superset.query_object_v1"
+CACHE_KEY_PAYLOAD_KEY = "cache_key"
-def _create_query_context_from_form(form_data: dict[str, Any]) -> QueryContext:
- """
- Create the query context from the form data.
- :param form_data: The task form data
- :returns: The query context
- :raises ValidationError: If the request is incorrect
+def _resolve_user(user_id: int | None, guest_token: "GuestToken | None") ->
User:
+ """Resolve the acting user for an async chart-data task.
+
+ The GTF executor does not impersonate on its own, so each task establishes
the
+ request user itself, which is what RLS and database impersonation key off.
"""
+ if user_id:
+ return security_manager.get_user_by_id(user_id)
+ if guest_token:
+ return security_manager.get_guest_user_from_token(guest_token)
+ return security_manager.get_anonymous_user()
+
- try:
- return ChartDataQueryContextSchema().load(form_data)
- except KeyError as ex:
- raise ValidationError("Request is incorrect") from ex
+def _inject_contribution_totals(
+ query_obj: "QueryObject", totals_cache_key: str
+) -> None:
+ """Inject ``contribution_totals`` from the cached totals query into
``query_obj``.
+ A contribution query normalizes its metrics against column sums from a
separate
+ "totals" query. In the per-query task model the totals query runs as its
own
+ task (a ``depends_on`` prerequisite) and caches its dataframe; here we
read that
+ cached dataframe and inject the sums into this query's contribution
+ post-processing before it runs — the same result the synchronous
+ ``ensure_totals_available`` produces, but reading the cache the
prerequisite
+ populated instead of re-running the totals query. ``contribution_totals``
is
+ stripped from the cache key, so this affects only the result, not the key.
+ """
+ from superset.common.utils.query_cache_manager import QueryCacheManager
-def _load_user_from_job_metadata(job_metadata: dict[str, Any]) -> User:
- if user_id := job_metadata.get("user_id"):
- # logged in user
- user = security_manager.get_user_by_id(user_id)
- elif guest_token := job_metadata.get("guest_token"):
- # embedded guest user
- user = security_manager.get_guest_user_from_token(guest_token)
- del job_metadata["guest_token"]
- else:
- # default to anonymous user if no user is found
- user = security_manager.get_anonymous_user()
- return user
+ cache = QueryCacheManager.get(key=totals_cache_key,
region=CacheRegion.DATA)
+ if not cache.is_loaded or cache.df is None:
+ # The depends_on prerequisite guarantees the totals task succeeded and
wrote
+ # this cache entry, so a miss is unexpected (e.g. it was evicted
between the
+ # totals task finishing and this task reading). Fail loudly rather than
+ # caching a silently un-normalized result the client would then
re-request:
+ # this task's single query cannot reproduce the synchronous path's
+ # ensure_totals_available (it has no totals query to run).
+ raise SupersetException(
+ f"Contribution totals not found in cache under {totals_cache_key}"
+ )
+ df = cache.df
+ totals = {col: df[col].sum() for col in df.columns if df[col].dtype.kind
in "biufc"}
+ for post_processing in query_obj.post_processing or []:
+ if post_processing.get("operation") == "contribution":
+ post_processing.setdefault("options", {})["contribution_totals"] =
totals
-def _handle_soft_time_limit(
- job_metadata: dict[str, Any], ex: Exception, activity: str
-) -> None:
+def _get_dependency_cache_key() -> str:
+ """
+ Return the cache key published by a prerequisite chart-data task.
+
+ A dependent task reaches this point only after the scheduler's all-success
+ dependency gate has passed, so prerequisite payloads are expected to
contain
+ any output metadata their task body committed.
"""
- SoftTimeLimitExceeded is raised both by a genuine timeout and by a
- user-initiated cancel (revoke sends SIGUSR1). The cancel endpoint has
- already emitted the terminal event for the latter - it has to, since a task
- revoked while still queued never reaches this handler - so only a timeout
- is reported here, and without one the client would wait forever.
+ for payload in get_context().get_dependency_payloads():
+ cache_key = payload.get(CACHE_KEY_PAYLOAD_KEY)
+ if isinstance(cache_key, str):
+ return cache_key
+ raise SupersetException("Prerequisite task did not publish a cache key")
+
+
+@contextmanager
+def _capture_query_cancellation(query_context: "QueryContext") ->
Iterator[None]:
+ """Enable engine-level cancellation of this task's warehouse query.
+
+ Captures the engine cancel id off the live cursor (for engines that expose
+ one before execution) and registers an abort handler that kills the backend
+ session over a fresh connection, unblocking the task's ``get_df``. Engines
+ without cancel support capture nothing and the task stays non-abortable, so
+ an abort/timeout simply frees the task without killing the (uncancellable)
+ query — matching the pre-cancellation behavior for those engines.
"""
- if async_query_manager.is_job_cancelled(job_metadata["job_id"]):
- logger.info("Cancelled by the user while %s", activity)
+ database = getattr(query_context.datasource, "database", None)
+ if database is None:
+ yield
return
- logger.warning("A timeout occurred while %s, error: %s", activity, ex)
- async_query_manager.update_job(
- job_metadata,
- async_query_manager.STATUS_ERROR,
- errors=[{"message": f"A timeout occurred while {activity}"}],
- )
+ ctx = get_context()
+ app = current_app._get_current_object() # noqa: SLF001
+ captured = False
+
+ def _sink(cursor: Any) -> None:
+ nonlocal captured
+ if captured:
+ return
+ cancel_id = capture_cancel_query_id(database, cursor)
+ if cancel_id is None:
+ return
+ captured = True
+
+ # Persist the handle so the orphan reaper can cancel the query if this
+ # worker dies. Set it before on_abort: registering the first handler
+ # flushes the whole property cache, persisting the handle in that
write.
+ ctx.set_cancellation(database.id, cancel_id)
+ # Registering the first abort handler marks the task abortable and
starts
+ # the abort listener; on abort it cancels the query on a fresh
connection.
+ def _cancel() -> None:
+ cancel_chart_query(database, cancel_id, app)
-@celery_app.task(name="load_chart_data_into_cache",
soft_time_limit=query_timeout)
-def load_chart_data_into_cache(
- job_metadata: dict[str, Any],
- form_data: dict[str, Any],
+ ctx.on_abort(_cancel)
+
+ with capture_cancel_id(_sink):
+ yield
+
+
+@task(name=CHART_QUERY_TASK, scope=TaskScope.SHARED)
+def execute_chart_query(
+ serialized_query: SerializedQuery,
+ user_id: int | None = None,
+ guest_token: "GuestToken | None" = None,
+ requires_totals: bool = False,
) -> None:
- # pylint: disable=import-outside-toplevel
- from superset.commands.chart.data.get_data_command import ChartDataCommand
-
- with override_user(_load_user_from_job_metadata(job_metadata),
force=False):
- try:
- set_form_data(form_data)
- query_context = _create_query_context_from_form(form_data)
- command = ChartDataCommand(query_context)
- result = command.run(cache=True)
- cache_key = result["cache_key"]
- result_url = f"/api/v1/chart/data/{cache_key}"
- async_query_manager.update_job(
- job_metadata,
- async_query_manager.STATUS_DONE,
- result_url=result_url,
- )
- except SoftTimeLimitExceeded as ex:
- _handle_soft_time_limit(job_metadata, ex, "loading chart data")
- raise
- except Exception as ex:
- # Extract SIP-40 style errors when available
- if isinstance(ex, SupersetErrorException):
- errors = [dataclasses.asdict(ex.error)]
- elif isinstance(ex, SupersetErrorsException):
- errors = [dataclasses.asdict(error) for error in ex.errors]
- else:
- # Fallback for non-Superset exceptions
- error = str(ex.message if hasattr(ex, "message") else ex)
- errors = [{"message": error}]
- async_query_manager.update_job(
- job_metadata,
- async_query_manager.STATUS_ERROR,
- errors=sanitize_error_dicts(errors),
+ """Execute a single chart-data query and cache it under its
query_cache_key.
+
+ The atomic async unit: reconstruct the one query (canonical serialization),
+ optionally inject contribution totals from a prerequisite totals task,
then run
+ the existing per-query execution/caching path so a re-request reads the
same
+ DATA-cache entry. The query runs under ``_capture_query_cancellation`` so
an
+ abort/timeout can cancel it on engines that support query cancellation.
+ """
+ with override_user(_resolve_user(user_id, guest_token), force=False):
+ query_context = load_serialized_query(serialized_query)
+ # Floor the result-cache TTL: async caches the result for a follow-up
+ # request to read back (see get_cache_timeout).
+ query_context.is_async_execution = True
+ query_obj = query_context.queries[0]
+ if requires_totals:
+ _inject_contribution_totals(query_obj, _get_dependency_cache_key())
+ # Executes on cache miss and writes CacheRegion.DATA under
query_cache_key.
+ with _capture_query_cancellation(query_context):
+ result = query_context.get_df_payload_result(query_obj)
+ if cache_key := result.payload.get(CACHE_KEY_PAYLOAD_KEY):
+ # Write synchronously: a dependent contribution query reads this
+ # cache key via get_dependency_payloads once the DAG gate releases,
+ # so it must not sit in the throttle buffer.
+ get_context().update_task(
+ payload={CACHE_KEY_PAYLOAD_KEY: cache_key}, immediate=True
)
- raise
+
+
+def _query_task_cache_key(query_context: "QueryContext", index: int) -> str |
None:
+ """Compute a query's cache key exactly as its task will.
+
+ ``execute_chart_query`` validates each query before keying (see
+ ``get_df_payload_result``), so validate here too — otherwise the SHARED
task's
+ ``task_key`` (used for cross-user dedup) could diverge from the key the
task
+ actually caches under.
+ """
+ query_obj = query_context.queries[index]
+ query_obj.validate()
+ return query_context.query_cache_key(query_obj)
+
+
+def submit_chart_data_query_tasks(
+ query_context: "QueryContext",
+ user_id: int | None,
+) -> dict[str, Any]:
+ """Fan a chart-data request out into one GTF task per ``QueryObject``.
+
+ Each ``QueryObject`` runs as its own SHARED task keyed by its
``query_cache_key``
+ (safe cross-user dedup — the key encodes RLS/impersonation), writing the
per-query
+ DATA cache a later re-request reads back. A contribution query
``depends_on`` the
+ totals query's task and reads its cached result to normalize.
+
+ There is no coordinator task: the client polls
``/api/v1/task/status_changes`` and
+ aggregates the query tasks' own honest statuses itself (all ``SUCCESS`` →
re-issue
+ the request, now served entirely from the per-query cache; any terminal
non-success
+ → error). Completion is emitted per task by GTF (via the coordination
service), so
+ that is also what the websocket transport subscribes to.
+
+ Returns the HTTP 202 body ``{"task_ids": [...], "cursor": "..."}`` — the
query
+ tasks' UUIDs, in query order, plus the server-issued polling cursor. The
client
+ uses those values to poll through the GTF task API. Client aborts may
unsubscribe
+ from shared work or abort pending work; engine-level query cancellation is
outside
+ this chart async path.
+ """
+ guest_user = security_manager.get_current_guest_user_if_guest()
+ guest_token = guest_user.guest_token if guest_user else None
+
+ # Capture a status-poll cursor BEFORE any task is created. Because it
+ # predates every task's creation (and therefore every terminal transition),
+ # the client can poll `status_changes` from it and is guaranteed to observe
+ # each task's completion — closing the race where a task finishes before
the
+ # client's waiter/poll is established. Returned in the 202 (one small
value,
+ # unlike echoing every task id back).
+ poll_cursor = datetime.now()
Review Comment:
Good catch — this is fixed on `gaq-to-gtf` (commit `0500cfb8a7`): the 202
cursor is now floored to whole seconds via a shared `floored_status_cursor()`
helper used by both the handshake and `get_statuses_changed_since`, so a
same-second `changed_on` under the `>=` bound can no longer be skipped on
second-precision metastores (MySQL). Precision test added.
##########
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:
Real bug — thanks. Fixed in #43685: cancellation is deferred to the last
local waiter. A new `cancelUnwaitedTasks()` only cancels a task id when no
local waiter still awaits it (checked after unregistering the aborting waiter,
or before registering on an already-aborted signal), so aborting one chart no
longer cancels a deduplicated SHARED task another chart for the same principal
still needs. Regression test added.
##########
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:
You're right that the Helm template still renders the removed
`GLOBAL_ASYNC_QUERIES_*` websocket keys. We are intentionally **not** migrating
the Helm chart here: it is being deprecated as of 7.0 (official deprecation
notice landing in the chart shortly), and first-class support for the
realtime/websocket feature will be added to the official Superset Kubernetes
operator (https://github.com/apache/superset-kubernetes-operator) rather than
Helm. Flagging so we don't invest in Helm plumbing that's on its way out.
##########
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:
Fixed in #43685. You're correct — the server **subscribes** to
`entity-changes:*` and `task-status`; `realtime:<channel_id>` is only the
browser-delivery channel after fanout. Both `UPDATING.md` and the websocket
README now name the server-side subscriptions and explicitly note a Redis ACL
must allow `entity-changes:*` and `task-status` (omitting `task-status`
silently drops task updates).
--
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]
