You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by vi...@apache.org on 2026/08/23 17:26:09 UTC
(superset) branch villebro/gtf-websocket-wiring created (now 03c4765a357)
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a change to branch villebro/gtf-websocket-wiring
in repository https://gitbox.apache.org/repos/asf/superset.git
at 03c4765a357 docs(updating): document the GAQ→GTF cutover + websocket flow for 7.0
This branch includes the following new commits:
new 15c8a8587f9 fix(gtf): resolve tsc + mypy failures on the feature branch
new f89761cf2d2 feat(websocket): generic task push transport — per-principal chart-data + guest face pile
new 03c4765a357 docs(updating): document the GAQ→GTF cutover + websocket flow for 7.0
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(superset) 03/03: docs(updating): document the GAQ→GTF cutover + websocket flow for 7.0
Posted by vi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a commit to branch villebro/gtf-websocket-wiring
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 03c4765a357bea639163de5d107f0ce30eb0031e
Author: Ville Brofeldt <vi...@gmail.com>
AuthorDate: Sun Aug 23 10:25:36 2026 -0700
docs(updating): document the GAQ→GTF cutover + websocket flow for 7.0
Add a 7.0 (## Next) breaking-change entry for the Global Async Queries
re-platform onto GTF: the removed async_event API / AsyncQueryManager /
qc-<hash> replay + config keys, how to enable async chart data in the new flow
(GLOBAL_ASYNC_QUERIES flag + DISTRIBUTED_COORDINATION_CONFIG, per-request
async_mode opt-in), and how to enable the realtime websocket transport
(WEBSOCKET_* + the same-host superset-websocket server on Redis pub/sub).
Also drop the now-stale 'GLOBAL_ASYNC_QUERIES_CACHE_BACKEND is deprecated,
removed in 8.0' bullet: the key is fully removed this cycle (7.0), so the
deprecation-window narrative no longer applies.
---
UPDATING.md | 66 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 65 insertions(+), 1 deletion(-)
diff --git a/UPDATING.md b/UPDATING.md
index 3b7bc65ade4..d3cf3f47840 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -24,6 +24,71 @@ 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_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):
+
+```python
+WEBSOCKET_ENABLED = True
+WEBSOCKET_URL = "ws://<same-host>:8080/"
+WEBSOCKET_JWT_SECRET = "<output of: openssl rand -base64 42>"
+```
+
+Run the `superset-websocket` Node server on the **same 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`). The server now consumes Redis Pub/Sub channels
+(`entity-changes:*` broadcast nudges, `realtime:<channel_id>` per-principal
+messages) instead of the removed async-events streams; see
+`superset-websocket/README.md`.
+
- `SAMPLES_ROW_LIMIT` is now the default for `/datasource/samples` requests without a valid explicit `per_page`, rather than a hard per-request ceiling; explicit limits are honored up to the existing global row-limit ceiling, matching `/chart/data` SAMPLES requests.
### MCP tool results preserve stored string values
@@ -61,7 +126,6 @@ the old counter to use the outcome-specific replacements.
- [42393](https://github.com/apache/superset/pull/42393): Exported dataset YAML now carries a `uuid` for each metric and column so that custom folder assignments (which reference metrics/columns by UUID) survive an import into another workspace. This affects any export bundle that contains datasets, not just a dataset export: chart, dashboard, database and full-asset exports all embed the same dataset YAML, so a dashboard exported from this release also fails to import into an older one [...]
- [42300](https://github.com/apache/superset/pull/42300): Timeseries charts (line/area/bar) with a Y-axis bound in effect — either an explicit `yAxisBounds` or one derived from `truncateYAxis` — now clamp out-of-range data points to that bound instead of letting ECharts drop the point (and the line segments around it) entirely. Any existing chart with a configured Y-axis bound and data outside it will look different after upgrading: a gap becomes a point pinned to the boundary. The clamp [...]
- [42087](https://github.com/apache/superset/pull/42087): Stored calculated-column and metric expressions are validated when a query is built, under the same sub-query policy already applied to adhoc expressions. Previously only the dataset update path checked them on save, so expressions written by v1 import, by dataset duplication, or before that check existed were never validated. Since `ALLOW_ADHOC_SUBQUERY` defaults to `False` (see [19242](https://github.com/apache/superset/pull/192 [...]
-- The `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND` config key is **deprecated** in favor of the shared coordination backend `DISTRIBUTED_COORDINATION_CONFIG`, which powers a single Redis connection for distributed locks, pub/sub, and the async-events streams (the GAQ firehose). All parameters previously accepted by `GLOBAL_ASYNC_QUERIES_CACHE_BACKEND` are supported identically under `DISTRIBUTED_COORDINATION_CONFIG` (they use the same `RedisCache`/`RedisSentinelCache` backend). During the deprec [...]
### Selenium support removed — Playwright is now required for screenshots
(superset) 01/03: fix(gtf): resolve tsc + mypy failures on the feature branch
Posted by vi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a commit to branch villebro/gtf-websocket-wiring
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 15c8a8587f9b462f64a4002d50c891697aa8583e
Author: Ville Brofeldt <vi...@gmail.com>
AuthorDate: Sun Aug 23 09:52:27 2026 -0700
fix(gtf): resolve tsc + mypy failures on the feature branch
Two type errors surfaced on the gaq-to-gtf feature branch's CI (tsc and
mypy) that predate the 6b work; fold the fixes in here.
- dashboard/types.ts: declare the per-dashboard `async_mode` override
(step 5) on `DashboardInfo.metadata` so FilterValue/FiltersConfigForm's
`state.dashboardInfo?.metadata?.async_mode` reads type-check.
- tasks/async_queries.py: widen the chart-data orchestrator's
`depends_on` param/local to the `list[CoreTask | UUID | str] | None`
union that `TaskOptions.depends_on` accepts (list invariance: a
`list[Task]` is not assignable to the declared element union).
---
superset-frontend/src/dashboard/types.ts | 2 ++
superset/tasks/async_queries.py | 10 ++++++++--
2 files changed, 10 insertions(+), 2 deletions(-)
diff --git a/superset-frontend/src/dashboard/types.ts b/superset-frontend/src/dashboard/types.ts
index adbe07d0dfc..9ccbf8bc44e 100644
--- a/superset-frontend/src/dashboard/types.ts
+++ b/superset-frontend/src/dashboard/types.ts
@@ -35,6 +35,7 @@ import { chart } from 'src/components/Chart/chartReducer';
import componentTypes from 'src/dashboard/util/componentTypes';
import Database from 'src/types/Database';
import { UrlParamEntries } from 'src/utils/urlUtils';
+import { AsyncModeOverride } from 'src/utils/asyncMode';
import { ResourceStatus } from 'src/hooks/apiResources/apiResources';
import { UserWithPermissionsAndRoles } from 'src/types/bootstrapTypes';
import User from 'src/types/User';
@@ -197,6 +198,7 @@ export type DashboardInfo = {
refresh_frequency?: number;
positions?: JsonObject;
filter_scopes?: JsonObject;
+ async_mode?: AsyncModeOverride;
};
crossFiltersEnabled: boolean;
filterBarOrientation: FilterBarOrientation;
diff --git a/superset/tasks/async_queries.py b/superset/tasks/async_queries.py
index bc6abf81303..efba3fb1bde 100644
--- a/superset/tasks/async_queries.py
+++ b/superset/tasks/async_queries.py
@@ -18,6 +18,7 @@ from __future__ import annotations
import logging
from typing import Any, TYPE_CHECKING
+from uuid import UUID
from flask import current_app
from flask_appbuilder.security.sqla.models import User
@@ -37,6 +38,8 @@ from superset.tasks.decorators import task
from superset.utils.core import override_user
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
@@ -174,7 +177,10 @@ def submit_chart_data_query_tasks(
query_context.cache_values["queries"][totals_idx]["row_limit"] = None
totals_key = _query_task_cache_key(query_context, totals_idx)
- def _schedule(index: int, depends_on: list["Task"] | None = None) -> "Task":
+ def _schedule(
+ index: int,
+ depends_on: "list[CoreTask | UUID | str] | None" = None,
+ ) -> "Task":
return execute_chart_query.schedule(
serialize_query(query_context, index),
user_id,
@@ -192,7 +198,7 @@ def submit_chart_data_query_tasks(
tasks[totals_idx] = _schedule(totals_idx)
for index in range(len(queries)):
if index not in tasks:
- depends_on = (
+ depends_on: "list[CoreTask | UUID | str] | None" = (
[tasks[totals_idx]]
if index in needs_totals and totals_idx is not None
else None
(superset) 02/03: feat(websocket): generic task push transport — per-principal chart-data + guest face pile
Posted by vi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a commit to branch villebro/gtf-websocket-wiring
in repository https://gitbox.apache.org/repos/asf/superset.git
commit f89761cf2d2d20d6335b0589f96e18c255e98343
Author: Ville Brofeldt <vi...@gmail.com>
AuthorDate: Sun Aug 23 10:20:22 2026 -0700
feat(websocket): generic task push transport — per-principal chart-data + guest face pile
Wire up the realtime transport end to end (6b) on top of the 6a backend
foundation, and generalize superset-websocket into a feature-agnostic Task
push transport: the Node server routes purely by Redis channel name and never
interprets a message's contents, so any async task (chart data today;
thumbnails / reports / exports later) rides the same mechanism with its
specifics carried in the payload.
Slice C — per-principal (tier-2) publish:
- superset/websocket/channel.py: extract channel_id_for(user_id, guest_key) as
the single source of truth for the principal-channel format (user:<id> /
guest-<hmac>), shared by cookie minting and the task layer.
- TaskDAO.get_subscriber_channels(task_id): map a task's subscribers (users and
guests) to their per-principal channels.
- TaskManager.publish_task_status(task_uuid, status): best-effort fan-out of
{task_id, status} to realtime:<channel_id> for each subscriber channel;
called from publish_completion alongside the tier-1 nudge. A chart-data
client learns each terminal transition over the socket instead of re-polling.
Slice D — superset-websocket rewrite (generic transport):
- Replace the async-events Redis Streams firehose + result_url/per-job-channel
model with Redis Pub/Sub. Route by channel name only: entity-changes:*
(tier-1) broadcast to all sockets; realtime:<channel_id> (tier-2) delivered
to the JWT-bound principal. Forward a generic {channel, payload} envelope.
- Drop stream reconnection/last_id replay (Pub/Sub is lossy; the browser's
interval poll is the backstop). New config: entityChangesChannelPrefix /
perPrincipalChannelPrefix; cookie default superset-ws-token. README rewritten.
Slice E — frontend WS client (asyncEvent.ts):
- Connect to WEBSOCKET_URL (JWT channel cookie rides the handshake) when
WEBSOCKET_ENABLED; on a tier-2 message for an awaited task_id, apply the
status immediately (accelerates completion). Reconnect on close; the existing
status_changes interval poll stays the correctness backstop.
Guest subscribers in the Task List face pile:
- get_subscribers now returns embedded guests as anonymized is_guest entries
with stable G1/G2… labels (ordered by subscription time) instead of
nameless blanks; the face pile renders them as G1/G2 avatars.
Tests: TaskManager tier-2 publish, DAO get_subscriber_channels, channel_id_for,
schema guest serialization, Node pub/sub routing (broadcast + per-principal +
malformed), and asyncEvent WS acceleration. All green; mypy/ruff/pylint/tsc/
eslint clean on changed files.
---
superset-frontend/src/features/tasks/types.ts | 10 +-
.../src/middleware/asyncEvent.test.ts | 76 ++++
superset-frontend/src/middleware/asyncEvent.ts | 97 ++++++
superset-frontend/src/pages/TaskList/index.tsx | 25 +-
superset-websocket/README.md | 100 ++++--
superset-websocket/config.example.json | 5 +-
superset-websocket/config.test.json | 3 +-
superset-websocket/spec/index.test.ts | 388 +++++----------------
superset-websocket/src/config.ts | 23 +-
superset-websocket/src/index.ts | 261 +++++---------
superset/daos/tasks.py | 30 ++
superset/tasks/manager.py | 60 ++++
superset/tasks/schemas.py | 52 ++-
superset/websocket/channel.py | 24 +-
tests/unit_tests/daos/test_tasks.py | 33 ++
tests/unit_tests/tasks/test_manager.py | 71 ++++
tests/unit_tests/tasks/test_schemas.py | 36 ++
tests/unit_tests/websocket/test_channel.py | 12 +
18 files changed, 766 insertions(+), 540 deletions(-)
diff --git a/superset-frontend/src/features/tasks/types.ts b/superset-frontend/src/features/tasks/types.ts
index 1dcd5b62ac9..7386daef231 100644
--- a/superset-frontend/src/features/tasks/types.ts
+++ b/superset-frontend/src/features/tasks/types.ts
@@ -18,9 +18,13 @@
*/
export interface TaskSubscriber {
- user_id: number;
- first_name: string;
- last_name: string;
+ // Authenticated subscribers carry a user_id + profile; embedded guests have
+ // no ab_user, so they arrive as is_guest with an anonymized label (G1/G2/…).
+ user_id: number | null;
+ first_name?: string;
+ last_name?: string;
+ is_guest?: boolean;
+ label?: string;
subscribed_at: string;
}
diff --git a/superset-frontend/src/middleware/asyncEvent.test.ts b/superset-frontend/src/middleware/asyncEvent.test.ts
index 2de71a75a52..3223a75cdc3 100644
--- a/superset-frontend/src/middleware/asyncEvent.test.ts
+++ b/superset-frontend/src/middleware/asyncEvent.test.ts
@@ -157,3 +157,79 @@ test('settles every request awaiting a deduplicated shared task', async () => {
expect(a).toEqual([{ chart: 'a' }]);
expect(b).toEqual([{ chart: 'b' }]);
});
+
+describe('realtime WebSocket acceleration', () => {
+ const realtime = (taskId: string, status: string) =>
+ JSON.stringify({
+ channel: `realtime:user:1`,
+ payload: { task_id: taskId, status },
+ });
+
+ // waitForAsyncData registers its waiter only after the baseline cursor fetch
+ // resolves; wait a tick so a one-shot socket message isn't delivered before
+ // the task is being awaited.
+ const afterRegistered = () => new Promise(resolve => setTimeout(resolve, 50));
+
+ test('a tier-2 message settles a waiting chart without a poll', async () => {
+ // No status batches queued (only the baseline), so completion can ONLY come
+ // from the socket message — proving the WS path settles on its own.
+ queueStatuses();
+ asyncEvent.init(config);
+
+ const refetch = jest.fn().mockResolvedValue([{ rows: 1 }]);
+ const promise = asyncEvent.waitForAsyncData(
+ { task_ids: ['task-1'] },
+ refetch,
+ );
+
+ await afterRegistered();
+ asyncEvent.handleRealtimeMessage(realtime('task-1', 'success'));
+
+ expect(await promise).toEqual([{ rows: 1 }]);
+ expect(refetch).toHaveBeenCalledTimes(1);
+ });
+
+ test('a tier-2 failure message rejects the waiting chart', async () => {
+ queueStatuses();
+ asyncEvent.init(config);
+
+ const refetch = jest.fn();
+ const promise = asyncEvent.waitForAsyncData(
+ { task_ids: ['task-1'] },
+ refetch,
+ );
+
+ await afterRegistered();
+ asyncEvent.handleRealtimeMessage(realtime('task-1', 'failure'));
+
+ await expect(promise).rejects.toThrow();
+ expect(refetch).not.toHaveBeenCalled();
+ });
+
+ test('ignores non-realtime channels and malformed data', async () => {
+ queueStatuses();
+ asyncEvent.init(config);
+
+ const refetch = jest.fn().mockResolvedValue([{ rows: 1 }]);
+ const promise = asyncEvent.waitForAsyncData(
+ { task_ids: ['task-1'] },
+ refetch,
+ );
+
+ await afterRegistered();
+ // A public entity-change nudge carries no status and must not settle a chart.
+ asyncEvent.handleRealtimeMessage(
+ JSON.stringify({
+ channel: 'entity-changes:task',
+ payload: { entity_type: 'task', id: 'task-1' },
+ }),
+ );
+ // Malformed data must be swallowed, not thrown.
+ expect(() => asyncEvent.handleRealtimeMessage('not json')).not.toThrow();
+ expect(refetch).not.toHaveBeenCalled();
+
+ // The task is still pending; complete it so the test doesn't leak a waiter.
+ asyncEvent.handleRealtimeMessage(realtime('task-1', 'success'));
+ await promise;
+ });
+});
diff --git a/superset-frontend/src/middleware/asyncEvent.ts b/superset-frontend/src/middleware/asyncEvent.ts
index c64ec4ee622..33e1eb2e679 100644
--- a/superset-frontend/src/middleware/asyncEvent.ts
+++ b/superset-frontend/src/middleware/asyncEvent.ts
@@ -80,6 +80,19 @@ let baselineReady: Promise<void> | null;
// stop instead of scheduling a second loop or mutating fresh state.
let pollingGeneration = 0;
+// Optional realtime WebSocket transport (superset-websocket). It merely
+// accelerates completion: a per-principal tier-2 message carries a task's
+// terminal status so a waiting chart settles immediately instead of at the next
+// poll tick. The interval poll below is the correctness backstop, so the socket
+// is strictly best-effort — if it is down or a message is lost, nothing breaks.
+let ws: WebSocket | undefined;
+let wsReconnectTimeoutId: number;
+let wsReconnectDelayMs: number;
+// Per-principal channel prefix (mirrors superset-websocket routing and
+// TaskManager.REALTIME_CHANNEL_PREFIX). The browser socket is JWT-bound to its
+// own principal channel, so it only ever receives its own realtime messages.
+const REALTIME_CHANNEL_PREFIX = 'realtime:';
+
const fetchStatusChanges = makeApi<
{ cursor?: string | null; task_type: string },
StatusChangesResponse
@@ -161,6 +174,83 @@ const loadStatusChanges = async (generation: number) => {
);
};
+/**
+ * Apply a realtime message received over the WebSocket.
+ *
+ * The server forwards a generic ``{channel, payload}`` envelope. For a
+ * per-principal (tier-2) chart-data message the 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). Any other channel
+ * (e.g. the public ``entity-changes:*`` nudges) is ignored here — chart-data
+ * only cares about its own tasks. Strictly best-effort: malformed data is
+ * dropped, never thrown.
+ */
+export const handleRealtimeMessage = (rawData: string) => {
+ try {
+ const { channel, payload } = JSON.parse(rawData) ?? {};
+ if (typeof channel !== 'string') return;
+ if (!channel.startsWith(REALTIME_CHANNEL_PREFIX)) return;
+ const taskId = payload?.task_id;
+ const status = payload?.status;
+ if (typeof taskId === 'string' && typeof status === 'string') {
+ applyStatus(taskId, status);
+ }
+ } catch (err) {
+ logging.warn('Failed to handle realtime message', err);
+ }
+};
+
+const teardownWebSocket = () => {
+ if (wsReconnectTimeoutId) clearTimeout(wsReconnectTimeoutId);
+ if (ws) {
+ // Detach handlers first so the in-flight socket's close doesn't schedule a
+ // reconnect against the superseded generation.
+ ws.onmessage = null;
+ ws.onclose = null;
+ ws.onerror = null;
+ try {
+ ws.close();
+ } catch {
+ // ignore: closing an already-closed/broken socket is harmless
+ }
+ ws = undefined;
+ }
+};
+
+// Open the realtime WebSocket (if enabled) and route its messages. The JWT
+// channel cookie (set by the Flask app) rides the handshake automatically since
+// the server runs on the same host. On close we reconnect after a delay; the
+// interval poll keeps completions flowing meanwhile, so a socket outage only
+// costs latency, not correctness.
+const connectWebSocket = (generation: number) => {
+ if (generation !== pollingGeneration) return;
+ const url = config?.WEBSOCKET_URL;
+ if (!config?.WEBSOCKET_ENABLED || !url || typeof WebSocket === 'undefined') {
+ return;
+ }
+ try {
+ ws = new WebSocket(url);
+ } catch (err) {
+ logging.warn('Failed to open realtime WebSocket', err);
+ return;
+ }
+ ws.onmessage = (event: MessageEvent) => {
+ if (generation !== pollingGeneration) return;
+ handleRealtimeMessage(String(event.data));
+ };
+ ws.onclose = () => {
+ if (generation !== pollingGeneration) return;
+ wsReconnectTimeoutId = window.setTimeout(
+ () => connectWebSocket(generation),
+ wsReconnectDelayMs,
+ );
+ };
+ // Errors surface as a subsequent close; let onclose own the reconnect and
+ // avoid logging the (payload-free) error event on every transient blip.
+ ws.onerror = () => {};
+};
+
/**
* Await completion of an async chart-data job's query tasks, then re-issue the
* original request to read the now-cached results.
@@ -220,6 +310,7 @@ export const waitForAsyncData = async <T = unknown[]>(
export const init = (appConfig?: AppConfig) => {
pollingGeneration += 1;
if (pollingTimeoutId) clearTimeout(pollingTimeoutId);
+ teardownWebSocket();
if (!isFeatureEnabled(FeatureFlag.GlobalAsyncQueries)) return;
const generation = pollingGeneration;
@@ -228,6 +319,12 @@ export const init = (appConfig?: AppConfig) => {
config = appConfig || getBootstrapData().common.conf;
pollingDelayMs = config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY || 500;
+ wsReconnectDelayMs =
+ config.GLOBAL_ASYNC_QUERIES_WEBSOCKET_RECONNECT_DELAY || 5000;
+
+ // Open the realtime socket (no-op unless WEBSOCKET_ENABLED); it only
+ // accelerates the poll loop below.
+ connectWebSocket(generation);
// Establish a baseline cursor before any chart query is triggered, so tasks
// created afterwards are all caught by the changed-since poll, then start the
diff --git a/superset-frontend/src/pages/TaskList/index.tsx b/superset-frontend/src/pages/TaskList/index.tsx
index 22ae2f99f54..a1572240cec 100644
--- a/superset-frontend/src/pages/TaskList/index.tsx
+++ b/superset-frontend/src/pages/TaskList/index.tsx
@@ -348,12 +348,25 @@ function TaskList({ addDangerToast, addSuccessToast, user }: TaskListProps) {
return '-';
}
- // Convert subscribers to SubjectPile format
- const subjects = subscribers.map((sub: TaskSubscriber) => ({
- id: sub.user_id,
- label: `${sub.first_name} ${sub.last_name}`.trim(),
- type: 1,
- }));
+ // Convert subscribers to SubjectPile format. Authenticated users show
+ // their name initials; embedded guests (no profile) show their
+ // anonymized G1/G2 label under a synthetic negative id (there is no
+ // user_id to key on, and negatives can't collide with real user ids).
+ const subjects = subscribers.map(
+ (sub: TaskSubscriber, index: number) =>
+ sub.is_guest
+ ? {
+ id: -(index + 1),
+ label: sub.label ?? t('Guest'),
+ type: 1,
+ }
+ : {
+ id: sub.user_id as number,
+ label:
+ `${sub.first_name ?? ''} ${sub.last_name ?? ''}`.trim(),
+ type: 1,
+ },
+ );
return <SubjectPile subjects={subjects} maxCount={3} />;
},
diff --git a/superset-websocket/README.md b/superset-websocket/README.md
index de9cc746d94..0c771bd1ebe 100644
--- a/superset-websocket/README.md
+++ b/superset-websocket/README.md
@@ -19,40 +19,68 @@ under the License.
# Superset WebSocket Server
-A Node.js WebSocket server for sending async event data to the Superset web application frontend.
+A Node.js WebSocket server that pushes realtime events from the Superset backend
+to the web frontend. It is a **generic, feature-agnostic transport**: it routes
+messages by Redis channel name and never interprets their contents, so any
+Superset feature (async chart data, background tasks, and future producers such
+as thumbnails, reports, or exports) can push over it without a server change.
## Requirements
- Node.js 12+ (not tested with older versions)
- Redis 5+
-To use this feature, Superset needs to be configured to enable global async queries and to use WebSockets as the transport (see below).
+To use realtime push, enable it in the Superset backend (`WEBSOCKET_ENABLED`,
+`WEBSOCKET_URL`, `WEBSOCKET_JWT_SECRET`; see below) and run this server on the
+same host.
## Architecture
-This implementation is based on the architecture defined in [SIP-39](https://github.com/apache/superset/issues/9190).
+### Redis Pub/Sub channels and the two tiers
-### Streams
+Realtime events are published to Redis **Pub/Sub** by the Superset Flask app
+(`superset/tasks/manager.py`). Pub/Sub is intentionally lossy (fire-and-forget):
+a missed message is reconciled by the frontend's interval poll of the authorized
+REST API, so no message is authoritative and none needs to be replayed.
-Async events are pushed to [Redis Streams](https://redis.io/topics/streams-intro) from the [Superset Flask app](https://github.com/preset-io/superset/blob/master/superset/async_events/async_query_manager.py). An event for a particular user is published to two streams: 1) the global event stream that includes events for all users, and 2) a channel/session-specific stream only for the user. This approach provides a good balance of performance (reading off of a single global stream) and fau [...]
+The server tails two channel-prefix patterns and routes by name only:
-Note that Redis Stream [consumer groups](https://redis.io/topics/streams-intro#consumer-groups) are not used here due to the fact that each group receives a subset of the data for a stream, and WebSocket clients have a persistent connection to each app instance, requiring access to all data in a stream. Horizontal scaling of the WebSocket app requires having multiple WebSocket servers, each with full access to the Redis Stream data.
+1. **Tier 1 — public entity-change nudges** (`entity-changes:<type>`, e.g.
+ `entity-changes:task`). Broadcast to **every** connected socket. The payload
+ carries only opaque ids (`{entity_type, id}`) — no status or sensitive data —
+ so a list view can learn "an entity of this type changed" and re-fetch just
+ the affected rows through the authorized API. Each client filters to the ids
+ it renders.
+2. **Tier 2 — per-principal messages** (`realtime:<channel_id>`, where
+ `channel_id` is `user:<id>` or `guest-<hmac>`). Delivered **only** to sockets
+ whose JWT bound that principal channel. Because delivery is scoped to a
+ principal already entitled to see the entity, the payload may carry
+ feature-specific detail (e.g. a task's `{task_id, status}`).
-### Connection
-
-When a user's browser initially connects to the WebSocket server, it does so over HTTP, which includes the JWT authentication cookie, set by the Flask app, in the request. _Note that due to the cookie-based authentication method, the WebSocket server must be run on the same host as the web application._ The server validates the JWT token by using the shared secret (config: `jwtSecret`), and if valid, proceeds to upgrade the connection to a WebSocket. The user's session-based "channel" ID [...]
-
-A user may have multiple WebSocket connections under a single channel (session) ID. This would be the case if the user has multiple browser tabs open, for example. In this scenario, **all events received for a specific channel are sent to all connected sockets**, leaving it to the consumer to decide which events are relevant to the current application context.
+The server forwards each message to the browser in a generic envelope
+`{ channel, payload }` (the full Redis channel name plus the parsed payload), so
+the client routes on `channel` exactly as the server does.
-### Reconnection
+### Connection
-It is expected that a user's WebSocket connection may be dropped or interrupted due to fluctuating network conditions. The Superset frontend code keeps track of the last received async event ID, and attempts to reconnect to the WebSocket server with a `last_id` query parameter in the initial HTTP request. If a connection includes a valid `last_id` value, events that may have already been received and sent unsuccessfully are read from the channel-based Redis Stream and re-sent to the new [...]
+When a user's browser connects, it does so over HTTP, including the JWT
+authentication cookie set by the Flask app (`WEBSOCKET_JWT_COOKIE_NAME`, default
+`superset-ws-token`). _Because authentication is cookie-based, the WebSocket
+server must run on the same host as the web application._ The server verifies
+the JWT with the shared secret (`jwtSecret` / `WEBSOCKET_JWT_SECRET`) and binds
+the socket to the `channel` claim (the principal channel), which is how tier-2
+messages are routed to it. A principal may have multiple sockets (e.g. several
+browser tabs); all matching messages are sent to all of them.
### Connection Management
-The server utilizes the standard WebSocket [ping/pong functionality](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_servers#pings_and_pongs_the_heartbeat_of_websockets) to determine if active WebSocket connections are still alive. Active sockets are sent a _ping_ regularly (config: `pingSocketsIntervalMs`), and the internal _sockets_ registry is updated with a timestamp when a _pong_ response is received. If a _pong_ response has not been received befor [...]
-
-In addition to periodic socket connection cleanup, the internal _channels_ registry is regularly "cleaned" (config: `gcChannelsIntervalMs`) to remove stale references and prevent excessive memory consumption over time.
+The server uses standard WebSocket
+[ping/pong](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_servers#pings_and_pongs_the_heartbeat_of_websockets)
+to detect dead connections. Active sockets are pinged regularly (config:
+`pingSocketsIntervalMs`) and the internal registry records the last _pong_
+timestamp; a socket that has not responded within `socketResponseTimeoutMs` is
+terminated. The channel registry is periodically cleaned
+(`gcChannelsIntervalMs`) to release stale references.
## Install
@@ -89,35 +117,33 @@ the JWT cookie uses `SameSite=None`.
## Superset Configuration
-Configure the Superset Flask app to enable global async queries (in `superset_config.py`):
-
-Enable the `GLOBAL_ASYNC_QUERIES` feature flag:
+Enable realtime push in the Superset Flask app (in `superset_config.py`):
```python
-"GLOBAL_ASYNC_QUERIES": True
+WEBSOCKET_ENABLED = True
+WEBSOCKET_URL = "ws://<host>:<port>/"
+WEBSOCKET_JWT_SECRET = "<a strong random secret, >= 32 bytes>"
```
-Configure the following Superset values:
+Note that the WebSocket server must be run on the same hostname (different port)
+for the JWT cookie to be shared between the Flask app and the WebSocket server.
-```python
-GLOBAL_ASYNC_QUERIES_TRANSPORT = "ws"
-GLOBAL_ASYNC_QUERIES_WEBSOCKET_URL = "ws://<host>:<port>/"
-```
-
-Note that the WebSocket server must be run on the same hostname (different port) for cookies to be shared between the Flask app and the WebSocket server.
+Note also that `localhost` and `127.0.0.1` are not considered the same host. For
+example, if you're pointing your browser to `localhost:<port>` for Superset,
+then the WebSocket url will need to be configured as `localhost:<port>`.
-Note also that `localhost` and `127.0.0.1` are not considered the same host. For example, if you're pointing your browser to `localhost:<port>` for Superset, then the WebSocket url will need to be configured as `localhost:<port>`.
+The following values must match between the Flask app config and this server's
+`config.json` (or its environment-variable overrides):
-The following config values must contain the same values in both the Flask app config and `config.json`:
-
-```text
-GLOBAL_ASYNC_QUERIES_CACHE_BACKEND
-GLOBAL_ASYNC_QUERIES_REDIS_STREAM_PREFIX
-GLOBAL_ASYNC_QUERIES_JWT_COOKIE_NAME
-GLOBAL_ASYNC_QUERIES_JWT_SECRET
-```
+| Flask app config | WebSocket server config |
+| ----------------------------- | ----------------------------- |
+| `WEBSOCKET_JWT_SECRET` | `jwtSecret` / `JWT_SECRET` |
+| `WEBSOCKET_JWT_COOKIE_NAME` | `jwtCookieName` / `JWT_COOKIE_NAME` |
-More info on Superset configuration values for async queries: https://superset.apache.org/docs/contributing/misc#async-chart-queries
+The Redis connection (`redis` / `REDIS_*`) must point at the same Redis instance
+Superset publishes to (`DISTRIBUTED_COORDINATION_CONFIG`). The channel prefixes
+(`entityChangesChannelPrefix`, `perPrincipalChannelPrefix`) default to match the
+backend and rarely need changing.
## StatsD monitoring
diff --git a/superset-websocket/config.example.json b/superset-websocket/config.example.json
index 6ecd391c27d..63ece8aa97e 100644
--- a/superset-websocket/config.example.json
+++ b/superset-websocket/config.example.json
@@ -15,9 +15,10 @@
"db": 0,
"ssl": false
},
- "redisStreamPrefix": "async-events-",
+ "entityChangesChannelPrefix": "entity-changes:",
+ "perPrincipalChannelPrefix": "realtime:",
"jwtAlgorithms": ["HS256"],
"jwtSecret": "CHANGE-ME",
- "jwtCookieName": "async-token",
+ "jwtCookieName": "superset-ws-token",
"allowedOrigins": []
}
diff --git a/superset-websocket/config.test.json b/superset-websocket/config.test.json
index 85578465370..bc4fb01557c 100644
--- a/superset-websocket/config.test.json
+++ b/superset-websocket/config.test.json
@@ -8,7 +8,6 @@
"port": 8125,
"globalTags": []
},
- "redisStreamPrefix": "test-async-events-",
"jwtSecret": "test123-test123-test123-test123-test123-test123-test123",
- "jwtCookieName": "test-async-token"
+ "jwtCookieName": "test-ws-token"
}
diff --git a/superset-websocket/spec/index.test.ts b/superset-websocket/spec/index.test.ts
index 13e36ad31a6..f7bf5b02335 100644
--- a/superset-websocket/spec/index.test.ts
+++ b/superset-websocket/spec/index.test.ts
@@ -34,43 +34,26 @@ import { WebSocket } from 'ws';
import * as server from '../src/index';
import { statsd } from '../src/index';
-const { mockRedisXrange } = vi.hoisted(() => {
- return { mockRedisXrange: vi.fn() };
+const { mockPsubscribe, mockOn } = vi.hoisted(() => {
+ return { mockPsubscribe: vi.fn(), mockOn: vi.fn() };
});
vi.mock('ws');
vi.mock('ioredis', () => {
return {
Redis: vi.fn().mockImplementation(function () {
- return { xrange: mockRedisXrange, on: vi.fn() };
+ return { psubscribe: mockPsubscribe, on: mockOn };
}),
};
});
const wsMock = WebSocket as unknown as Mock<typeof WebSocket>;
-const channelId = 'bc9e040c-7b4a-4817-99b9-292832d97ec7';
-const streamReturnValue: server.StreamResult[] = [
- [
- '1615426152415-0',
- [
- 'data',
- `{"channel_id": "${channelId}", "job_id": "c9b99965-8f1e-4ce5-aa43-d6fc94d6a510", "user_id": "1", "status": "done", "errors": [], "result_url": "/superset/explore_json/data/ejr-37281682b1282cdb8f25e0de0339b386"}`,
- ],
- ],
- [
- '1615426152516-0',
- [
- 'data',
- `{"channel_id": "${channelId}", "job_id": "f1e5bb1f-f2f1-4f21-9b2f-c9b91dcc9b59", "user_id": "1", "status": "done", "errors": [], "result_url": "/api/v1/chart/data/qc-64e8452dc9907dd77746cb75a19202de"}`,
- ],
- ],
-];
+const channelId = 'user:5';
describe('server', () => {
let statsdIncrementMock: Mock<typeof statsd.increment>;
beforeEach(() => {
- mockRedisXrange.mockClear();
server.resetState();
statsdIncrementMock = vi.spyOn(statsd, 'increment').mockReturnValue();
});
@@ -139,53 +122,6 @@ describe('server', () => {
});
});
- describe('incrementId', () => {
- test('it increments a valid Redis stream ID', () => {
- expect(server.incrementId('1607477697866-0')).toEqual('1607477697866-1');
- });
-
- test('it handles an invalid Redis stream ID', () => {
- expect(server.incrementId('foo')).toEqual('foo');
- });
- });
-
- describe('getLastId', () => {
- const requestWithUrl = (url: string): http.IncomingMessage => {
- const request = new http.IncomingMessage(new net.Socket());
- request.url = url;
- return request;
- };
-
- test('returns null when last_id is absent', () => {
- expect(server.getLastId(requestWithUrl('http://localhost'))).toBeNull();
- });
-
- test('returns a well-formed Redis stream ID', () => {
- expect(
- server.getLastId(
- requestWithUrl('http://localhost?last_id=1607477697866-0'),
- ),
- ).toEqual('1607477697866-0');
- });
-
- test.each([
- 'abc-xyz',
- '1607477697866',
- '1607477697866-',
- '-0',
- '1607477697866-0; DROP TABLE',
- '1607477697866-0-0',
- ])('returns null for malformed last_id %p', value => {
- expect(
- server.getLastId(
- requestWithUrl(
- `http://localhost?last_id=${encodeURIComponent(value)}`,
- ),
- ),
- ).toBeNull();
- });
- });
-
describe('redisUrlFromConfig', () => {
test('it builds a valid Redis URL from defaults', () => {
expect(
@@ -239,127 +175,115 @@ describe('server', () => {
});
});
- describe('processStreamResults', () => {
- test('sends data to channel', async () => {
- const ws = new wsMock('localhost');
- const sendMock = vi.spyOn(ws, 'send');
- const socketInstance = { ws: ws, channel: channelId, pongTs: Date.now() };
+ describe('routeRedisMessage', () => {
+ test('routes a per-principal message to the matching channel only', () => {
+ // Two principals connected; a realtime:user:5 message reaches only user:5.
+ const wsA = new wsMock('localhost');
+ const sendA = vi.spyOn(wsA, 'send');
+ server.trackClient('user:5', {
+ ws: wsA,
+ channel: 'user:5',
+ pongTs: Date.now(),
+ });
- expect(statsdIncrementMock).toHaveBeenCalledTimes(0);
- server.trackClient(channelId, socketInstance);
- expect(statsdIncrementMock).toHaveBeenCalledTimes(1);
- expect(statsdIncrementMock).toHaveBeenNthCalledWith(
- 1,
- 'ws_connected_client',
- );
+ const wsB = new wsMock('localhost');
+ const sendB = vi.spyOn(wsB, 'send');
+ server.trackClient('user:9', {
+ ws: wsB,
+ channel: 'user:9',
+ pongTs: Date.now(),
+ });
- server.processStreamResults(streamReturnValue);
- expect(statsdIncrementMock).toHaveBeenCalledTimes(1);
+ const payload = { task_id: 'abc', status: 'success' };
+ server.routeRedisMessage('realtime:user:5', JSON.stringify(payload));
- const message1 = `{"id":"1615426152415-0","channel_id":"${channelId}","job_id":"c9b99965-8f1e-4ce5-aa43-d6fc94d6a510","user_id":"1","status":"done","errors":[],"result_url":"/superset/explore_json/data/ejr-37281682b1282cdb8f25e0de0339b386"}`;
- const message2 = `{"id":"1615426152516-0","channel_id":"${channelId}","job_id":"f1e5bb1f-f2f1-4f21-9b2f-c9b91dcc9b59","user_id":"1","status":"done","errors":[],"result_url":"/api/v1/chart/data/qc-64e8452dc9907dd77746cb75a19202de"}`;
- expect(sendMock).toHaveBeenCalledWith(message1);
- expect(sendMock).toHaveBeenCalledWith(message2);
+ expect(sendA).toHaveBeenCalledTimes(1);
+ expect(sendA).toHaveBeenCalledWith(
+ JSON.stringify({ channel: 'realtime:user:5', payload }),
+ );
+ expect(sendB).not.toHaveBeenCalled();
});
- test('channel not present', async () => {
+ test('routes a per-principal message to a guest channel', () => {
const ws = new wsMock('localhost');
- const sendMock = vi.spyOn(ws, 'send');
+ const send = vi.spyOn(ws, 'send');
+ server.trackClient('guest-abc', {
+ ws,
+ channel: 'guest-abc',
+ pongTs: Date.now(),
+ });
- expect(statsdIncrementMock).toHaveBeenCalledTimes(0);
- server.processStreamResults(streamReturnValue);
- expect(statsdIncrementMock).toHaveBeenCalledTimes(0);
+ const payload = { task_id: 'xyz', status: 'failure' };
+ server.routeRedisMessage('realtime:guest-abc', JSON.stringify(payload));
- expect(sendMock).not.toHaveBeenCalled();
+ expect(send).toHaveBeenCalledWith(
+ JSON.stringify({ channel: 'realtime:guest-abc', payload }),
+ );
});
- test('error sending data to client', async () => {
- const ws = new wsMock('localhost');
- const sendMock = vi.spyOn(ws, 'send').mockImplementation(() => {
- throw new Error();
+ test('broadcasts an entity-change nudge to every socket', () => {
+ const wsA = new wsMock('localhost');
+ const sendA = vi.spyOn(wsA, 'send');
+ server.trackClient('user:5', {
+ ws: wsA,
+ channel: 'user:5',
+ pongTs: Date.now(),
});
- const socketInstance = { ws: ws, channel: channelId, pongTs: Date.now() };
-
- expect(statsdIncrementMock).toHaveBeenCalledTimes(0);
- server.trackClient(channelId, socketInstance);
- expect(statsdIncrementMock).toHaveBeenCalledTimes(1);
- expect(statsdIncrementMock).toHaveBeenNthCalledWith(
- 1,
- 'ws_connected_client',
- );
-
- server.processStreamResults(streamReturnValue);
- expect(statsdIncrementMock).toHaveBeenCalledTimes(2);
- expect(statsdIncrementMock).toHaveBeenNthCalledWith(
- 2,
- 'ws_client_send_error',
- );
- expect(sendMock).toHaveBeenCalled();
- expect(Object.keys(server.channels)).toHaveLength(0);
- });
+ const wsB = new wsMock('localhost');
+ const sendB = vi.spyOn(wsB, 'send');
+ server.trackClient('guest-abc', {
+ ws: wsB,
+ channel: 'guest-abc',
+ pongTs: Date.now(),
+ });
- const makeItem = (i: number): server.StreamResult =>
- [
- `161542615${i}-0`,
- [
- 'data',
- JSON.stringify({
- channel_id: channelId,
- job_id: `job-${i}`,
- status: 'done',
- }),
- ],
- ] as server.StreamResult;
+ const payload = { entity_type: 'task', id: 'abc' };
+ server.routeRedisMessage('entity-changes:task', JSON.stringify(payload));
- afterEach(() => {
- server.opts.eventYieldBatchSize = 100;
+ const expected = JSON.stringify({
+ channel: 'entity-changes:task',
+ payload,
+ });
+ expect(sendA).toHaveBeenCalledWith(expected);
+ expect(sendB).toHaveBeenCalledWith(expected);
});
- test('yields to the event loop for large batches', async () => {
- server.opts.eventYieldBatchSize = 2;
+ test('ignores a message on an unrecognized channel', () => {
const ws = new wsMock('localhost');
- server.trackClient(channelId, {
+ const send = vi.spyOn(ws, 'send');
+ server.trackClient('user:5', {
ws,
- channel: channelId,
+ channel: 'user:5',
pongTs: Date.now(),
});
- const sendMock = vi.spyOn(ws, 'send');
- const setImmediateSpy = vi.spyOn(global, 'setImmediate');
- const results = [0, 1, 2, 3, 4].map(makeItem);
- await server.processStreamResults(results);
+ server.routeRedisMessage('some-other:channel', JSON.stringify({ a: 1 }));
- // every event is still delivered
- expect(sendMock).toHaveBeenCalledTimes(5);
- // and the loop yielded at least at indexes 2 and 4
- expect(setImmediateSpy.mock.calls.length).toBeGreaterThanOrEqual(2);
- setImmediateSpy.mockRestore();
+ expect(send).not.toHaveBeenCalled();
});
- test('processes the whole batch when yielding is disabled', async () => {
- server.opts.eventYieldBatchSize = 0;
+ test('swallows a malformed (non-JSON) message', () => {
const ws = new wsMock('localhost');
- server.trackClient(channelId, {
+ const send = vi.spyOn(ws, 'send');
+ server.trackClient('user:5', {
ws,
- channel: channelId,
+ channel: 'user:5',
pongTs: Date.now(),
});
- const sendMock = vi.spyOn(ws, 'send');
- const results = [0, 1, 2, 3, 4].map(makeItem);
- await server.processStreamResults(results);
-
- expect(sendMock).toHaveBeenCalledTimes(5);
+ // Must not throw; simply drops the unparseable message.
+ expect(() =>
+ server.routeRedisMessage('realtime:user:5', 'not json'),
+ ).not.toThrow();
+ expect(send).not.toHaveBeenCalled();
});
});
describe('backpressure', () => {
- const fakeEvent = {
- id: '1615426152415-0',
- channel_id: channelId,
- job_id: 'c9b99965-8f1e-4ce5-aa43-d6fc94d6a510',
- status: 'done',
+ const message: server.OutboundMessage = {
+ channel: 'realtime:user:5',
+ payload: { task_id: 'abc', status: 'success' },
};
afterEach(() => {
@@ -382,7 +306,7 @@ describe('server', () => {
pongTs: Date.now(),
});
- server.sendToChannel(channelId, fakeEvent);
+ server.sendToChannel(channelId, message);
expect(terminateMock).not.toHaveBeenCalled();
expect(sendMock).toHaveBeenCalled();
@@ -400,7 +324,7 @@ describe('server', () => {
pongTs: Date.now(),
});
- server.sendToChannel(channelId, fakeEvent);
+ server.sendToChannel(channelId, message);
expect(terminateMock).toHaveBeenCalled();
expect(sendMock).not.toHaveBeenCalled();
@@ -422,75 +346,28 @@ describe('server', () => {
pongTs: Date.now(),
});
- server.sendToChannel(channelId, fakeEvent);
+ server.sendToChannel(channelId, message);
expect(terminateMock).not.toHaveBeenCalled();
expect(sendMock).toHaveBeenCalled();
});
- });
- describe('fetchRangeFromStream', () => {
- beforeEach(() => {
- mockRedisXrange.mockClear();
- });
-
- test('success with results', async () => {
- mockRedisXrange.mockResolvedValueOnce(streamReturnValue);
- const cb = vi.fn() as Mock<
- (results: server.StreamResult[]) => void | Promise<void>
- >;
- await server.fetchRangeFromStream({
- sessionId: '123',
- startId: '-',
- endId: '+',
- listener: cb,
+ test('error sending data to client cleans the channel', () => {
+ const ws = new wsMock('localhost');
+ const sendMock = vi.spyOn(ws, 'send').mockImplementation(() => {
+ throw new Error();
});
-
- expect(mockRedisXrange).toHaveBeenCalledWith(
- 'test-async-events-123',
- '-',
- '+',
- );
- expect(cb).toHaveBeenCalledWith(streamReturnValue);
- });
-
- test('success no results', async () => {
- const cb = vi.fn() as Mock<
- (results: server.StreamResult[]) => void | Promise<void>
- >;
- await server.fetchRangeFromStream({
- sessionId: '123',
- startId: '-',
- endId: '+',
- listener: cb,
+ server.trackClient(channelId, {
+ ws,
+ channel: channelId,
+ pongTs: Date.now(),
});
- expect(mockRedisXrange).toHaveBeenCalledWith(
- 'test-async-events-123',
- '-',
- '+',
- );
- expect(cb).not.toHaveBeenCalled();
- });
-
- test('error', async () => {
- const cb = vi.fn() as Mock<
- (results: server.StreamResult[]) => void | Promise<void>
- >;
- mockRedisXrange.mockRejectedValueOnce(new Error());
- await server.fetchRangeFromStream({
- sessionId: '123',
- startId: '-',
- endId: '+',
- listener: cb,
- });
+ server.sendToChannel(channelId, message);
- expect(mockRedisXrange).toHaveBeenCalledWith(
- 'test-async-events-123',
- '-',
- '+',
- );
- expect(cb).not.toHaveBeenCalled();
+ expect(sendMock).toHaveBeenCalled();
+ expect(statsdIncrementMock).toHaveBeenCalledWith('ws_client_send_error');
+ expect(Object.keys(server.channels)).toHaveLength(0);
});
});
@@ -537,7 +414,7 @@ describe('server', () => {
}).toThrow();
});
- test('valid JWT, no lastId', async () => {
+ test('valid JWT binds the socket to its channel', async () => {
const validToken = jwt.sign({ channel: channelId }, config.jwtSecret);
const request = getRequest(validToken, 'http://localhost');
@@ -550,79 +427,6 @@ describe('server', () => {
expect(channelSockets.sockets).toHaveLength(1);
const socketId = channelSockets.sockets[0];
expect(server.sockets[socketId]).toEqual(socketInstanceExpected);
- expect(mockRedisXrange).not.toHaveBeenCalled();
- expect(wsEventMock).toHaveBeenCalledWith('pong', expect.any(Function));
- });
-
- test('valid JWT, malformed lastId is ignored', async () => {
- const validToken = jwt.sign({ channel: channelId }, config.jwtSecret);
- const request = getRequest(
- validToken,
- 'http://localhost?last_id=abc-xyz',
- );
-
- server.wsConnection(ws, request);
-
- const channelSockets = server.channels[channelId];
- expect(channelSockets).toEqual({
- sockets: expect.any(Array<string>),
- });
- expect(channelSockets.sockets).toHaveLength(1);
-
- // Malformed last_id must not trigger a stream range fetch.
- const socketId = channelSockets.sockets[0];
- expect(server.sockets[socketId]).toEqual(socketInstanceExpected);
- });
-
- test('valid JWT, with lastId', async () => {
- const validToken = jwt.sign({ channel: channelId }, config.jwtSecret);
- const lastId = '1615426152415-0';
- const request = getRequest(
- validToken,
- `http://localhost?last_id=${lastId}`,
- );
-
- server.wsConnection(ws, request);
-
- const channelSockets = server.channels[channelId];
- expect(channelSockets).toEqual({
- sockets: expect.any(Array<string>),
- });
- expect(channelSockets.sockets).toHaveLength(1);
- const socketId = channelSockets.sockets[0];
- expect(server.sockets[socketId]).toEqual(socketInstanceExpected);
- expect(mockRedisXrange).toHaveBeenCalledWith(
- expect.stringContaining(channelId),
- '1615426152415-1',
- '+',
- );
- expect(wsEventMock).toHaveBeenCalledWith('pong', expect.any(Function));
- });
-
- test('valid JWT, with lastId and lastFirehoseId', async () => {
- const validToken = jwt.sign({ channel: channelId }, config.jwtSecret);
- const lastId = '1615426152415-0';
- const lastFirehoseId = '1715426152415-0';
- const request = getRequest(
- validToken,
- `http://localhost?last_id=${lastId}`,
- );
-
- server.setLastFirehoseId(lastFirehoseId);
- server.wsConnection(ws, request);
-
- const channelSockets = server.channels[channelId];
- expect(channelSockets).toEqual({
- sockets: expect.any(Array<string>),
- });
- expect(channelSockets.sockets).toHaveLength(1);
- const socketId = channelSockets.sockets[0];
- expect(server.sockets[socketId]).toEqual(socketInstanceExpected);
- expect(mockRedisXrange).toHaveBeenCalledWith(
- expect.stringContaining(channelId),
- '1615426152415-1',
- lastFirehoseId,
- );
expect(wsEventMock).toHaveBeenCalledWith('pong', expect.any(Function));
});
diff --git a/superset-websocket/src/config.ts b/superset-websocket/src/config.ts
index 54a014c70e0..2a07ac4cdd4 100644
--- a/superset-websocket/src/config.ts
+++ b/superset-websocket/src/config.ts
@@ -42,9 +42,8 @@ type ConfigType = {
globalTags: Array<string>;
};
redis: RedisConfig;
- redisStreamPrefix: string;
- redisStreamReadCount: number;
- redisStreamReadBlockMs: number;
+ entityChangesChannelPrefix: string;
+ perPrincipalChannelPrefix: string;
jwtAlgorithms: string[];
jwtSecret: string;
jwtCookieName: string;
@@ -65,12 +64,13 @@ function defaultConfig(): ConfigType {
logLevel: 'info',
logToFile: false,
logFilename: 'app.log',
- redisStreamPrefix: 'async-events-',
- redisStreamReadCount: 100,
- redisStreamReadBlockMs: 5000,
+ // Redis Pub/Sub channel prefixes; must match the Superset Flask app's
+ // TaskManager (ENTITY_CHANGES_CHANNEL_PREFIX / REALTIME_CHANNEL_PREFIX).
+ entityChangesChannelPrefix: 'entity-changes:',
+ perPrincipalChannelPrefix: 'realtime:',
jwtAlgorithms: ['HS256'],
jwtSecret: '',
- jwtCookieName: 'async-token',
+ jwtCookieName: 'superset-ws-token',
jwtChannelIdKey: 'channel',
allowedOrigins: [],
socketResponseTimeoutMs: 60 * 1000,
@@ -144,11 +144,10 @@ function applyEnvOverrides(config: ConfigType): ConfigType {
LOG_LEVEL: val => (config.logLevel = val),
LOG_TO_FILE: val => (config.logToFile = toBoolean(val)),
LOG_FILENAME: val => (config.logFilename = val),
- REDIS_STREAM_PREFIX: val => (config.redisStreamPrefix = val),
- REDIS_STREAM_READ_COUNT: val =>
- (config.redisStreamReadCount = toNumber(val)),
- REDIS_STREAM_READ_BLOCK_MS: val =>
- (config.redisStreamReadBlockMs = toNumber(val)),
+ ENTITY_CHANGES_CHANNEL_PREFIX: val =>
+ (config.entityChangesChannelPrefix = val),
+ PER_PRINCIPAL_CHANNEL_PREFIX: val =>
+ (config.perPrincipalChannelPrefix = val),
JWT_SECRET: val => (config.jwtSecret = val),
JWT_COOKIE_NAME: val => (config.jwtCookieName = val),
ALLOWED_ORIGINS: val => (config.allowedOrigins = toStringArray(val)),
diff --git a/superset-websocket/src/index.ts b/superset-websocket/src/index.ts
index 4e206b5141d..ca60a738b4e 100644
--- a/superset-websocket/src/index.ts
+++ b/superset-websocket/src/index.ts
@@ -30,40 +30,30 @@ import { createLogger } from './logger.js';
import { buildConfig, RedisConfig } from './config.js';
import { checkServerIdentity, PeerCertificate } from 'tls';
-export type StreamResult = [
- recordId: string,
- record: [label: 'data', data: string],
-];
-
-// sync with superset-frontend/src/components/ErrorMessage/types
-export type ErrorLevel = 'info' | 'warning' | 'error';
-// eslint-disable-next-line @typescript-eslint/no-explicit-any
-export type SupersetError<ExtraType = Record<string, any> | null> = {
- error_type: string;
- extra: ExtraType;
- level: ErrorLevel;
- message: string;
-};
-
-type ListenerFunction = (results: StreamResult[]) => void | Promise<void>;
-interface EventValue {
- id: string;
- channel_id: string;
- job_id: string;
- user_id?: string;
- status: string;
- errors?: SupersetError[];
- result_url?: string;
-}
interface JwtPayload {
[key: string]: string;
}
-interface FetchRangeFromStreamParams {
- sessionId: string;
- startId: string;
- endId: string;
- listener: ListenerFunction;
+
+/**
+ * The generic, feature-agnostic message the server forwards to browsers. The
+ * server does not understand any feature's payload — it only routes by Redis
+ * `channel` name — so `payload` is passed through verbatim. A browser client
+ * routes on `channel`:
+ * - `entity-changes:<type>` — a lossy, public "an entity of this type
+ * changed" nudge, broadcast to every connected socket; `payload` carries
+ * opaque ids only (`{entity_type, id}`).
+ * - `<realtimeChannelPrefix><principalChannel>` — a per-principal message
+ * (e.g. a task's status), delivered only to sockets whose JWT bound that
+ * principal channel; `payload` is feature-defined (e.g. `{task_id, status}`).
+ * Any future feature (thumbnails, reports, exports, …) reuses this envelope
+ * without a server change: its specifics live entirely in `payload`.
+ */
+export interface OutboundMessage {
+ channel: string;
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
+ payload: any;
}
+
export interface SocketInstance {
ws: WebSocket;
channel: string;
@@ -139,9 +129,11 @@ export const buildRedisOpts = (baseConfig: RedisConfig) => {
return redisOpts;
};
-// initialize servers
-const redis = new Redis(buildRedisOpts(opts.redis));
-redis.on('error', (err: Error) => {
+// A Redis connection dedicated to Pub/Sub: once a connection enters subscriber
+// mode it can no longer issue ordinary commands, so this is kept separate from
+// any future command connection.
+const redisSubscriber = new Redis(buildRedisOpts(opts.redis));
+redisSubscriber.on('error', (err: Error) => {
logger.error(`Redis connection error: ${err.message}`);
});
const httpServer = http.createServer();
@@ -151,17 +143,18 @@ export const wss = new WebSocketServer({
});
const SOCKET_ACTIVE_STATES: number[] = [WebSocket.OPEN, WebSocket.CONNECTING];
-const GLOBAL_EVENT_STREAM_NAME = `${opts.redisStreamPrefix}full`;
-const DEFAULT_STREAM_LAST_ID = '$';
+
+// The Pub/Sub channel patterns the server tails. Tier-1 entity-change nudges
+// are public and broadcast to every socket; tier-2 per-principal channels are
+// routed to the matching principal only (see routeRedisMessage). Both are lossy
+// (Pub/Sub is fire-and-forget); the browser's interval poll is the correctness
+// backstop, so no stream replay / reconnection catch-up is needed.
+const ENTITY_CHANGES_PATTERN = `${opts.entityChangesChannelPrefix}*`;
+const PER_PRINCIPAL_PATTERN = `${opts.perPrincipalChannelPrefix}*`;
// initialize internal registries
export let channels: Record<string, ChannelValue> = {};
export let sockets: Record<string, SocketInstance> = {};
-let lastFirehoseId: string = DEFAULT_STREAM_LAST_ID;
-
-export const setLastFirehoseId = (id: string): void => {
- lastFirehoseId = id;
-};
// WebSocket close code used when a connection is refused because a configured
// connection limit has been reached (1013 = "Try Again Later").
@@ -246,12 +239,16 @@ export const trackClient = (
};
/**
- * Sends a single async event payload to a single channel.
- * A channel may have multiple connected sockets, this emits
- * the event to all connected sockets within a channel.
+ * Sends a single message to every socket registered on a channel.
+ * A channel may have multiple connected sockets (e.g. one browser tab each);
+ * this emits the message to all of them, leaving it to the client to decide
+ * which are relevant to its current context.
*/
-export const sendToChannel = (channel: string, value: EventValue): void => {
- const strData = JSON.stringify(value);
+export const sendToChannel = (
+ channel: string,
+ message: OutboundMessage,
+): void => {
+ const strData = JSON.stringify(message);
if (!channels[channel]) {
logger.debug(`channel ${channel} is unknown, skipping`);
return;
@@ -293,95 +290,62 @@ export const sendToChannel = (channel: string, value: EventValue): void => {
};
/**
- * Reads a range of events from a channel-specific Redis event stream.
- * Invoked in the client re-connection flow.
+ * Sends a message to every connected socket, regardless of channel. Used for
+ * the public, lossy tier-1 entity-change nudges (`entity-changes:*`), which
+ * carry only opaque ids — each client filters to the ids it renders. Reuses
+ * `sendToChannel` per channel so backpressure and cleanup apply uniformly.
*/
-export const fetchRangeFromStream = async ({
- sessionId,
- startId,
- endId,
- listener,
-}: FetchRangeFromStreamParams) => {
- const streamName = `${opts.redisStreamPrefix}${sessionId}`;
- try {
- const reply = await redis.xrange(streamName, startId, endId);
- if (!reply || !reply.length) return;
- await listener(reply as StreamResult[]);
- } catch (e) {
- logger.error(e);
+export const broadcastToAll = (message: OutboundMessage): void => {
+ for (const channel in channels) {
+ sendToChannel(channel, message);
}
};
/**
- * Reads from the global Redis event stream continuously.
- * Utilizes a blocking connection to Redis to wait for data to
- * be returned from the stream.
+ * Routes a raw Redis Pub/Sub message to the appropriate sockets.
+ *
+ * Per-principal messages (`<perPrincipalChannelPrefix><principalChannel>`, e.g.
+ * `realtime:user:5`) are delivered only to sockets whose JWT bound that
+ * principal channel. Entity-change messages (`entity-changes:<type>`) are
+ * broadcast to all sockets. The full Redis channel name is forwarded to the
+ * browser in the envelope so the client can route by it too.
*/
-export const subscribeToGlobalStream = async (
- stream: string,
- listener: ListenerFunction,
-) => {
- /*eslint no-constant-condition: ["error", { "checkLoops": false }]*/
- while (true) {
- try {
- const reply = await redis.xread(
- 'COUNT',
- opts.redisStreamReadCount,
- 'BLOCK',
- opts.redisStreamReadBlockMs,
- 'STREAMS',
- stream,
- lastFirehoseId,
- );
- if (!reply) {
- continue;
- }
- const results = reply[0][1];
- const { length } = results;
- if (!results.length) {
- continue;
- }
- // Await the listener before advancing so that batches are processed
- // sequentially. processStreamResults yields to the event loop mid-batch
- // for large bursts; without awaiting here a subsequent xread could start
- // a concurrent batch and interleave out-of-order sends to clients.
- await listener(results as StreamResult[]);
- setLastFirehoseId(results[length - 1][0]);
- } catch (e) {
- logger.error(e);
- continue;
- }
+export const routeRedisMessage = (channel: string, rawMessage: string): void => {
+ let payload: unknown;
+ try {
+ payload = JSON.parse(rawMessage);
+ } catch (err) {
+ logger.error(`Failed to parse message on channel ${channel}: ${err}`);
+ return;
+ }
+ const message: OutboundMessage = { channel, payload };
+
+ const { perPrincipalChannelPrefix, entityChangesChannelPrefix } = opts;
+ if (channel.startsWith(perPrincipalChannelPrefix)) {
+ const principalChannel = channel.slice(perPrincipalChannelPrefix.length);
+ sendToChannel(principalChannel, message);
+ } else if (channel.startsWith(entityChangesChannelPrefix)) {
+ broadcastToAll(message);
+ } else {
+ logger.debug(`Ignoring message on unrecognized channel ${channel}`);
}
};
/**
- * Callback function to process events received from a Redis Stream.
- *
- * For large batches the loop periodically yields to the Node.js event loop
- * (via setImmediate) so that connection management, health checks and
- * ping/pong handling are not starved while a burst of events is processed.
- * The yield cadence is controlled by `eventYieldBatchSize` (0 disables it).
+ * Subscribes the Redis connection to the tier-1 (broadcast) and tier-2
+ * (per-principal) channel patterns and routes each received message.
*/
-export const processStreamResults = async (
- results: StreamResult[],
-): Promise<void> => {
- // Log only the batch size, not the raw payloads, which carry user and
- // job identifiers.
- logger.debug(`events received: count=${results.length}`);
- const { eventYieldBatchSize } = opts;
- for (let i = 0; i < results.length; i += 1) {
- if (eventYieldBatchSize > 0 && i > 0 && i % eventYieldBatchSize === 0) {
- await new Promise(resolve => setImmediate(resolve));
- }
- try {
- const item = results[i];
- const id = item[0];
- const data = JSON.parse(item[1][1]);
- sendToChannel(data.channel_id, { id, ...data });
- } catch (err) {
- logger.error(err);
- }
- }
+export const subscribeToChannels = async (): Promise<void> => {
+ redisSubscriber.on(
+ 'pmessage',
+ (_pattern: string, channel: string, message: string) => {
+ routeRedisMessage(channel, message);
+ },
+ );
+ await redisSubscriber.psubscribe(ENTITY_CHANGES_PATTERN, PER_PRINCIPAL_PATTERN);
+ logger.info(
+ `Subscribed to Redis channels: ${ENTITY_CHANGES_PATTERN}, ${PER_PRINCIPAL_PATTERN}`,
+ );
};
/**
@@ -405,36 +369,6 @@ const readChannelId = (request: http.IncomingMessage): string => {
return channelId;
};
-// Redis stream IDs have the form '<millisecondsTime>-<sequenceNumber>',
-// e.g. '1607477697866-0'.
-const REDIS_STREAM_ID_REGEX = /^\d{1,15}-\d{1,10}$/;
-
-/**
- * Extracts the `last_id` query param value from an HTTP request, returning it
- * only when it is a well-formed Redis stream ID. Malformed values are ignored
- * (returns null) rather than being passed through to incrementId / Redis.
- */
-export const getLastId = (request: http.IncomingMessage): string | null => {
- const url = new URL(String(request.url), 'http://0.0.0.0');
- const lastId = url.searchParams.get('last_id');
- if (lastId === null) return null;
- if (!REDIS_STREAM_ID_REGEX.test(lastId)) {
- logger.warn(`Ignoring malformed last_id query param: ${lastId}`);
- return null;
- }
- return lastId;
-};
-
-/**
- * Increments a Redis Stream ID
- */
-export const incrementId = (id: string): string => {
- // redis stream IDs are in this format: '1607477697866-0'
- const parts = id.split('-');
- if (parts.length < 2) return id;
- return parts[0] + '-' + (Number(parts[1]) + 1);
-};
-
/**
* WebSocket `connection` event handler, called via wss
*/
@@ -457,20 +391,8 @@ export const wsConnection = (ws: WebSocket, request: http.IncomingMessage) => {
const socketId = trackClient(channel, socketInstance);
logger.debug(`socket ${socketId} connected on channel ${channel}`);
- // reconnection logic
- const lastId = getLastId(request);
- if (lastId) {
- // fetch range of events from lastId to most recent event received on
- // via global event stream
- const endId =
- lastFirehoseId === DEFAULT_STREAM_LAST_ID ? '+' : lastFirehoseId;
- fetchRangeFromStream({
- sessionId: channel,
- startId: incrementId(lastId), // inclusive
- endId, // inclusive
- listener: processStreamResults,
- });
- }
+ // Pub/Sub is lossy and not replayable, so there is no server-side catch-up on
+ // reconnect: the browser's interval poll reconciles any missed nudges.
// init event handler for `pong` events (connection management)
ws.on('pong', function pong(data: Buffer) {
@@ -670,8 +592,8 @@ if (startServer) {
httpServer.listen(opts.port);
logger.info(`Server started on port ${opts.port}`);
- // start reading from event stream
- subscribeToGlobalStream(GLOBAL_EVENT_STREAM_NAME, processStreamResults);
+ // start receiving realtime messages from Redis Pub/Sub
+ subscribeToChannels();
// init garbage collection routines
setInterval(checkSockets, opts.pingSocketsIntervalMs);
@@ -688,5 +610,4 @@ if (startServer) {
export const resetState = () => {
channels = {};
sockets = {};
- lastFirehoseId = DEFAULT_STREAM_LAST_ID;
};
diff --git a/superset/daos/tasks.py b/superset/daos/tasks.py
index 59cc587b387..2a68b82de16 100644
--- a/superset/daos/tasks.py
+++ b/superset/daos/tasks.py
@@ -314,6 +314,36 @@ class TaskDAO(BaseDAO[Task]):
# Subscription management methods
+ @classmethod
+ def get_subscriber_channels(cls, task_id: int) -> list[str]:
+ """Return the per-principal realtime channels of a task's subscribers.
+
+ Maps each subscriber row (a user or an embedded guest) to its channel id
+ via the shared :func:`superset.websocket.channel.channel_id_for`
+ formatter, so a tier-2 completion event published to one of these
+ channels reaches exactly the socket whose cookie bound the same channel
+ (see ``TaskManager.publish_completion``). Skips the base filter: this is
+ internal plumbing for a task the executor already owns, not a user-facing
+ listing.
+
+ :param task_id: internal id of the task
+ :returns: distinct channel ids (``user:<id>`` / ``guest-<hmac>``); empty
+ when the task has no resolvable subscribers
+ """
+ from superset.websocket.channel import channel_id_for
+
+ rows = (
+ db.session.query(TaskSubscriber.user_id, TaskSubscriber.guest_key)
+ .filter(TaskSubscriber.task_id == task_id)
+ .all()
+ )
+ channels: list[str] = []
+ for user_id, guest_key in rows:
+ channel = channel_id_for(user_id, guest_key)
+ if channel is not None and channel not in channels:
+ channels.append(channel)
+ return channels
+
@classmethod
def add_subscriber(cls, task_id: int, user_id: int) -> bool:
"""
diff --git a/superset/tasks/manager.py b/superset/tasks/manager.py
index 86c5cce3f43..569cf88bcfe 100644
--- a/superset/tasks/manager.py
+++ b/superset/tasks/manager.py
@@ -123,6 +123,62 @@ class TaskManager:
)
return False
+ # Per-principal (tier-2) realtime channels. Unlike the public tier-1
+ # entity-change nudge above, a tier-2 message is fanned out to each of a
+ # task's *subscribers'* channels and MAY carry task-specific detail
+ # (``status``) because delivery is scoped to principals already entitled to
+ # see the task. The channel id is the principal's own channel
+ # (``user:<id>`` / ``guest-<hmac>``, see ``superset.websocket.channel``);
+ # the superset-websocket server routes a ``realtime:<channel_id>`` message
+ # to the socket whose JWT cookie bound that channel. This lets the chart-data
+ # client learn each status transition without re-polling the REST API on
+ # every intermediate change.
+ REALTIME_CHANNEL_PREFIX = "realtime:"
+
+ @classmethod
+ def publish_task_status(cls, task_uuid: UUID, status: str) -> bool:
+ """Publish a task's ``status`` to each subscriber's per-principal channel.
+
+ Best-effort tier-2 delivery: for every distinct subscriber channel of the
+ task (see ``TaskDAO.get_subscriber_channels``), publish
+ ``{task_id, status}`` to ``realtime:<channel_id>``. The superset-websocket
+ server forwards it to sockets whose JWT cookie bound that channel, so the
+ submitter (and any SHARED-dedup joiners) sees the transition immediately.
+ No-op when no coordination backend is configured or the task has no
+ resolvable subscribers; callers must never let a failure here disrupt
+ completion signalling (the interval poll is the backstop).
+
+ :param task_uuid: public UUID of the task (also carried in the payload)
+ :param status: the task's current status
+ :returns: True if at least one message was published, False otherwise
+ """
+ from superset.coordination.base import CoordinationService
+ from superset.daos.tasks import TaskDAO
+ from superset.utils import json
+
+ if not CoordinationService.is_backend_defined():
+ return False
+ try:
+ task = TaskDAO.find_one_or_none(uuid=task_uuid, skip_base_filter=True)
+ if task is None:
+ return False
+ channels = TaskDAO.get_subscriber_channels(task.id)
+ if not channels:
+ return False
+ message = json.dumps({"task_id": str(task_uuid), "status": status})
+ for channel in channels:
+ CoordinationService.publish(
+ f"{cls.REALTIME_CHANNEL_PREFIX}{channel}", message
+ )
+ return True
+ except Exception as ex: # noqa: BLE001 pylint: disable=broad-except
+ # Strictly best-effort (mirrors publish_entity_change): a Redis or
+ # serialization hiccup must never disrupt completion signalling.
+ logger.warning(
+ "Failed to publish task status for task %s: %s", task_uuid, ex
+ )
+ return False
+
@classmethod
def get_abort_channel(cls, task_uuid: UUID) -> str:
"""
@@ -199,6 +255,10 @@ class TaskManager:
# Also emit a lossy opaque nudge for realtime UI transports (separate
# from the guaranteed completion signal above).
cls.publish_entity_change(task_uuid)
+ # And push the terminal status to each subscriber's per-principal
+ # channel (tier-2), so a chart-data client learns completion detail
+ # over the socket instead of re-polling the REST API.
+ cls.publish_task_status(task_uuid, status)
return True
except redis.RedisError as ex:
# Best-effort: waiters fall back to polling, so a transient Redis
diff --git a/superset/tasks/schemas.py b/superset/tasks/schemas.py
index 6bd011cdb90..184355374f3 100644
--- a/superset/tasks/schemas.py
+++ b/superset/tasks/schemas.py
@@ -16,6 +16,8 @@
# under the License.
"""Task API schemas"""
+from datetime import datetime
+
from marshmallow import fields, Schema
from marshmallow.fields import Method
@@ -160,19 +162,45 @@ class TaskResponseSchema(Schema):
return obj.subscriber_count # type: ignore[attr-defined]
def get_subscribers(self, obj: object) -> list[dict[str, object]]:
- """Get list of subscribers with user info"""
+ """Get list of subscribers with user info.
+
+ Authenticated subscribers are returned with their user profile. Embedded
+ guests have no ``ab_user`` profile, so they are returned as anonymized
+ entries (``is_guest`` with a stable per-task ``label`` ``G1``/``G2``/…,
+ ordered by subscription time) — the Task List renders them as ``G1``/``G2``
+ avatars rather than nameless blanks.
+ """
+ all_subs = list(obj.subscribers) # type: ignore[attr-defined]
+ # Assign stable G-ordinals to guest subscribers, ordered by subscription
+ # time (then id) so the labels don't shuffle between requests.
+ guest_subs = sorted(
+ (s for s in all_subs if s.user_id is None),
+ key=lambda s: (s.subscribed_at or datetime.min, s.id),
+ )
+ guest_ordinal = {s.id: i + 1 for i, s in enumerate(guest_subs)}
+
subscribers = []
- for sub in obj.subscribers: # type: ignore[attr-defined]
- subscribers.append(
- {
- "user_id": sub.user_id,
- "first_name": sub.user.first_name if sub.user else None,
- "last_name": sub.user.last_name if sub.user else None,
- "subscribed_at": sub.subscribed_at.isoformat()
- if sub.subscribed_at
- else None,
- }
- )
+ for sub in all_subs:
+ subscribed_at = sub.subscribed_at.isoformat() if sub.subscribed_at else None
+ if sub.user_id is not None:
+ subscribers.append(
+ {
+ "user_id": sub.user_id,
+ "is_guest": False,
+ "first_name": sub.user.first_name if sub.user else None,
+ "last_name": sub.user.last_name if sub.user else None,
+ "subscribed_at": subscribed_at,
+ }
+ )
+ else:
+ subscribers.append(
+ {
+ "user_id": None,
+ "is_guest": True,
+ "label": f"G{guest_ordinal[sub.id]}",
+ "subscribed_at": subscribed_at,
+ }
+ )
return subscribers
def get_depends_on(self, obj: object) -> list[dict[str, object]]:
diff --git a/superset/websocket/channel.py b/superset/websocket/channel.py
index 02204c74ce8..1ee254cedef 100644
--- a/superset/websocket/channel.py
+++ b/superset/websocket/channel.py
@@ -44,6 +44,24 @@ from superset.utils.core import get_user_id
logger = logging.getLogger(__name__)
+def channel_id_for(user_id: int | None, guest_key: str | None) -> str | None:
+ """Derive a principal's realtime channel from its identity, or ``None``.
+
+ The single source of truth for the channel-id format, shared by the
+ request-scoped :func:`get_channel_id` (cookie minting) and the task layer's
+ per-subscriber publish (``TaskDAO.get_subscriber_channels``) so a message
+ published to a subscriber's channel lands on the socket whose cookie bound
+ that same channel. ``user:<id>`` for an authenticated user; the guest's
+ token-derived key (already namespaced ``guest-…``) for an embedded guest;
+ ``None`` when neither identifies a principal.
+ """
+ if user_id is not None:
+ return f"user:{user_id}"
+ if guest_key:
+ return guest_key
+ return None
+
+
def get_channel_id() -> str | None:
"""Return the realtime channel for the current request principal, or ``None``.
@@ -52,10 +70,8 @@ def get_channel_id() -> str | None:
and ``None`` for an anonymous request (no channel, no cookie).
"""
if security_manager.get_current_guest_user_if_guest():
- return get_current_guest_subscriber_key()
- if (user_id := get_user_id()) is not None:
- return f"user:{user_id}"
- return None
+ return channel_id_for(None, get_current_guest_subscriber_key())
+ return channel_id_for(get_user_id(), None)
def mint_channel_token(channel_id: str) -> str:
diff --git a/tests/unit_tests/daos/test_tasks.py b/tests/unit_tests/daos/test_tasks.py
index f35597ed174..ca10970e643 100644
--- a/tests/unit_tests/daos/test_tasks.py
+++ b/tests/unit_tests/daos/test_tasks.py
@@ -359,6 +359,39 @@ def test_add_subscriber_idempotent(session_with_task: Session) -> None:
assert len(task.subscribers) == 1
+def test_get_subscriber_channels(session_with_task: Session) -> None:
+ """get_subscriber_channels maps each subscriber to its per-principal channel."""
+ from superset.daos.tasks import TaskDAO
+
+ task = create_task(
+ session_with_task,
+ task_key="shared-channels",
+ scope=TaskScope.SHARED,
+ user_id=None,
+ )
+ TaskDAO.add_subscriber(task.id, user_id=7)
+ TaskDAO.add_guest_subscriber(task.id, guest_key="guest-abc")
+
+ channels = TaskDAO.get_subscriber_channels(task.id)
+
+ # A user subscriber → user:<id>; a guest subscriber → its token-derived key.
+ assert set(channels) == {"user:7", "guest-abc"}
+
+
+def test_get_subscriber_channels_empty(session_with_task: Session) -> None:
+ """get_subscriber_channels returns [] for a task with no subscribers."""
+ from superset.daos.tasks import TaskDAO
+
+ task = create_task(
+ session_with_task,
+ task_key="no-subs",
+ scope=TaskScope.SHARED,
+ user_id=None,
+ )
+
+ assert TaskDAO.get_subscriber_channels(task.id) == []
+
+
def test_remove_subscriber(session_with_task: Session) -> None:
"""Test removing a subscriber from a task"""
from superset.daos.tasks import TaskDAO
diff --git a/tests/unit_tests/tasks/test_manager.py b/tests/unit_tests/tasks/test_manager.py
index 8a3aa5c4dc7..23e45459368 100644
--- a/tests/unit_tests/tasks/test_manager.py
+++ b/tests/unit_tests/tasks/test_manager.py
@@ -196,6 +196,77 @@ class TestTaskManagerEntityChange:
assert TaskManager.publish_entity_change(uuid.uuid4()) is False
+class TestTaskManagerPublishTaskStatus:
+ """publish_task_status fans a task's status out to subscriber channels (tier-2)."""
+
+ def setup_method(self):
+ _reset_prefixes()
+
+ def teardown_method(self):
+ _reset_prefixes()
+
+ IS_DEFINED = "superset.coordination.base.CoordinationService.is_backend_defined"
+ PUBLISH = "superset.coordination.base.CoordinationService.publish"
+ DAO = "superset.daos.tasks.TaskDAO"
+
+ @patch(IS_DEFINED, return_value=False)
+ def test_no_backend(self, mock_defined):
+ assert TaskManager.publish_task_status(uuid.uuid4(), "success") is False
+
+ @patch(DAO)
+ @patch(PUBLISH)
+ @patch(IS_DEFINED, return_value=True)
+ def test_publishes_status_to_each_subscriber_channel(
+ self, mock_defined, mock_publish, mock_dao
+ ):
+ task_uuid = uuid.uuid4()
+ mock_dao.find_one_or_none.return_value = MagicMock(id=7)
+ mock_dao.get_subscriber_channels.return_value = ["user:1", "guest-abc"]
+
+ assert TaskManager.publish_task_status(task_uuid, "success") is True
+
+ mock_dao.get_subscriber_channels.assert_called_once_with(7)
+ # One publish per distinct subscriber channel, each on realtime:<channel>
+ # carrying the task uuid + status (delivery is per-principal, so status is
+ # allowed here — unlike the public tier-1 nudge).
+ assert mock_publish.call_count == 2
+ channels = {call.args[0] for call in mock_publish.call_args_list}
+ assert channels == {"realtime:user:1", "realtime:guest-abc"}
+ for call in mock_publish.call_args_list:
+ assert json.loads(call.args[1]) == {
+ "task_id": str(task_uuid),
+ "status": "success",
+ }
+
+ @patch(DAO)
+ @patch(PUBLISH)
+ @patch(IS_DEFINED, return_value=True)
+ def test_no_subscribers_is_noop(self, mock_defined, mock_publish, mock_dao):
+ mock_dao.find_one_or_none.return_value = MagicMock(id=7)
+ mock_dao.get_subscriber_channels.return_value = []
+
+ assert TaskManager.publish_task_status(uuid.uuid4(), "success") is False
+ mock_publish.assert_not_called()
+
+ @patch(DAO)
+ @patch(PUBLISH)
+ @patch(IS_DEFINED, return_value=True)
+ def test_task_not_found_is_noop(self, mock_defined, mock_publish, mock_dao):
+ mock_dao.find_one_or_none.return_value = None
+
+ assert TaskManager.publish_task_status(uuid.uuid4(), "success") is False
+ mock_publish.assert_not_called()
+
+ @patch(DAO)
+ @patch(PUBLISH, side_effect=redis.RedisError("boom"))
+ @patch(IS_DEFINED, return_value=True)
+ def test_redis_error_is_swallowed(self, mock_defined, mock_publish, mock_dao):
+ mock_dao.find_one_or_none.return_value = MagicMock(id=7)
+ mock_dao.get_subscriber_channels.return_value = ["user:1"]
+
+ assert TaskManager.publish_task_status(uuid.uuid4(), "success") is False
+
+
class TestTaskManagerListenForAbort:
"""listen_for_abort delegates to CoordinationService.listen_for_signal()."""
diff --git a/tests/unit_tests/tasks/test_schemas.py b/tests/unit_tests/tasks/test_schemas.py
index 1e90e6d57a0..c220c6327c4 100644
--- a/tests/unit_tests/tasks/test_schemas.py
+++ b/tests/unit_tests/tasks/test_schemas.py
@@ -67,3 +67,39 @@ def test_get_properties_exposes_debug_fields_when_show_stacktrace(
assert properties["exception_type"] == "KeyError"
assert str(properties["stack_trace"]).startswith("Traceback")
+
+
+def test_get_subscribers_mixes_users_and_anonymized_guests() -> None:
+ """User subscribers keep their profile; guests get anonymized G1/G2 labels."""
+ from datetime import datetime
+
+ def _sub(id_, user_id, user, subscribed_at):
+ return SimpleNamespace(
+ id=id_, user_id=user_id, user=user, subscribed_at=subscribed_at
+ )
+
+ alice = SimpleNamespace(first_name="Alice", last_name="Smith")
+ task = SimpleNamespace(
+ subscribers=[
+ _sub(1, 7, alice, datetime(2020, 1, 1, 0, 0, 0)),
+ # Two guests, deliberately out of subscription order to prove the
+ # G-ordinals are assigned by subscribed_at, not list order.
+ _sub(2, None, None, datetime(2020, 1, 1, 0, 0, 2)),
+ _sub(3, None, None, datetime(2020, 1, 1, 0, 0, 1)),
+ ]
+ )
+
+ result = TaskResponseSchema().get_subscribers(task)
+
+ assert result[0] == {
+ "user_id": 7,
+ "is_guest": False,
+ "first_name": "Alice",
+ "last_name": "Smith",
+ "subscribed_at": "2020-01-01T00:00:00",
+ }
+ # Guest ordinals follow subscription time: id=3 (earlier) → G1, id=2 → G2.
+ guests = {r["subscribed_at"]: r for r in result if r["is_guest"]}
+ assert guests["2020-01-01T00:00:01"]["label"] == "G1"
+ assert guests["2020-01-01T00:00:02"]["label"] == "G2"
+ assert all(g["user_id"] is None for g in guests.values())
diff --git a/tests/unit_tests/websocket/test_channel.py b/tests/unit_tests/websocket/test_channel.py
index 67412e89bfc..4ba43149872 100644
--- a/tests/unit_tests/websocket/test_channel.py
+++ b/tests/unit_tests/websocket/test_channel.py
@@ -18,6 +18,18 @@ import jwt
from pytest_mock import MockerFixture
+def test_channel_id_for_maps_principal_identity() -> None:
+ from superset.websocket.channel import channel_id_for
+
+ # A user id wins and maps to user:<id>; a guest key is returned verbatim
+ # (already namespaced); neither identity → None (anonymous).
+ assert channel_id_for(7, None) == "user:7"
+ assert channel_id_for(None, "guest-abc") == "guest-abc"
+ assert channel_id_for(None, None) is None
+ # A user id takes precedence over any stray guest key.
+ assert channel_id_for(7, "guest-abc") == "user:7"
+
+
def test_channel_id_for_logged_in_user(app_context, mocker: MockerFixture) -> None:
from superset.websocket import channel