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 18:23:56 UTC
(superset) 01/01: feat(listview): realtime list updates baked into useListViewResource
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a commit to branch villebro/gtf-realtime-listviews
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 0f132ec6e2791a770b7133d09d659c9b6b7acd41
Author: Ville Brofeldt <vi...@gmail.com>
AuthorDate: Sun Aug 23 11:23:14 2026 -0700
feat(listview): realtime list updates baked into useListViewResource
Step 7 of the GAQ→GTF epic — realtime list views, the headline of the new
websocket transport. A list backed by useListViewResource now live-patches its
on-screen rows as the underlying entities change, with no manual refresh.
- Shared realtime client (src/middleware/realtime.ts): owns the single browser
socket and fans the generic {channel, payload} envelope out to any number of
subscribers, so features share one connection. Extracted the socket ownership
out of asyncEvent.ts, which now just subscribes for its tier-2 chart-data
handler. Payload-agnostic and best-effort (reconnect on close).
- useListViewResource gains opt-in realtime (enableRealtime, realtimeIdField):
it subscribes to entity-changes:<resource>, ignores nudges for rows not
currently displayed, debounce-collects the rest (~500ms) and issues ONE
batched fetch of just those rows through the normal authorized list endpoint
(col:<idField> op:in), merging them in place by id. Update-only (no new-row
insertion), so authz/RLS are unchanged and a burst can't hammer the backend.
- Task List is the first surface: realtime enabled with idField 'uuid' (tasks
are UUID-facing); tasks already emit entity-change nudges at completion (6a).
Added 'uuid' to the task API search_columns so the batched refetch can filter.
Tests: shared client (dispatch/subscribe/reconnect), asyncEvent refactored onto
it, useListViewResource live-patch + ignore-offscreen, all green; tsc/eslint/
ruff/pylint clean.
---
.../src/middleware/asyncEvent.test.ts | 23 ++-
superset-frontend/src/middleware/asyncEvent.ts | 120 ++++----------
superset-frontend/src/middleware/realtime.test.ts | 168 +++++++++++++++++++
superset-frontend/src/middleware/realtime.ts | 179 +++++++++++++++++++++
superset-frontend/src/pages/TaskList/index.tsx | 13 +-
superset-frontend/src/views/CRUD/hooks.test.tsx | 110 +++++++++++++
superset-frontend/src/views/CRUD/hooks.ts | 86 ++++++++++
superset/tasks/api.py | 1 +
8 files changed, 599 insertions(+), 101 deletions(-)
diff --git a/superset-frontend/src/middleware/asyncEvent.test.ts b/superset-frontend/src/middleware/asyncEvent.test.ts
index 3223a75cdc3..a8a74dea9b2 100644
--- a/superset-frontend/src/middleware/asyncEvent.test.ts
+++ b/superset-frontend/src/middleware/asyncEvent.test.ts
@@ -159,11 +159,10 @@ test('settles every request awaiting a deduplicated shared task', async () => {
});
describe('realtime WebSocket acceleration', () => {
- const realtime = (taskId: string, status: string) =>
- JSON.stringify({
- channel: `realtime:user:1`,
- payload: { task_id: taskId, status },
- });
+ const realtime = (taskId: string, status: string) => ({
+ 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
@@ -206,7 +205,7 @@ describe('realtime WebSocket acceleration', () => {
expect(refetch).not.toHaveBeenCalled();
});
- test('ignores non-realtime channels and malformed data', async () => {
+ test('ignores non-realtime channels', async () => {
queueStatuses();
asyncEvent.init(config);
@@ -218,14 +217,10 @@ describe('realtime WebSocket acceleration', () => {
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();
+ asyncEvent.handleRealtimeMessage({
+ channel: 'entity-changes:task',
+ payload: { entity_type: 'task', id: 'task-1' },
+ });
expect(refetch).not.toHaveBeenCalled();
// The task is still pending; complete it so the test doesn't leak a waiter.
diff --git a/superset-frontend/src/middleware/asyncEvent.ts b/superset-frontend/src/middleware/asyncEvent.ts
index 33e1eb2e679..793e79a5b9e 100644
--- a/superset-frontend/src/middleware/asyncEvent.ts
+++ b/superset-frontend/src/middleware/asyncEvent.ts
@@ -24,6 +24,11 @@ import {
} from '@superset-ui/core';
import { logging } from '@apache-superset/core/utils';
import getBootstrapData from 'src/utils/getBootstrapData';
+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
@@ -80,17 +85,12 @@ 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.
+// The shared realtime client (src/middleware/realtime.ts) owns the socket; here
+// we only consume the tier-2 messages relevant to chart-data completion. The
+// socket is best-effort — the interval poll below is the correctness backstop.
const REALTIME_CHANNEL_PREFIX = 'realtime:';
const fetchStatusChanges = makeApi<
@@ -175,81 +175,28 @@ const loadStatusChanges = async (generation: number) => {
};
/**
- * Apply a realtime message received over the WebSocket.
+ * Handle a realtime message from the shared client (src/middleware/realtime.ts).
*
- * 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.
+ * 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:*`` list-view
+ * nudges) is ignored here — chart-data only cares about its own tasks.
*/
-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);
+export const handleRealtimeMessage = (message: RealtimeMessage) => {
+ const { channel, payload } = message;
+ 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);
}
};
-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 = () => {};
-};
+// Consume the shared realtime socket once at module load. The handler reads the
+// live waiter registry, so it stays correct across init() generations.
+subscribeRealtime(handleRealtimeMessage);
/**
* Await completion of an async chart-data job's query tasks, then re-issue the
@@ -310,21 +257,22 @@ export const waitForAsyncData = async <T = unknown[]>(
export const init = (appConfig?: AppConfig) => {
pollingGeneration += 1;
if (pollingTimeoutId) clearTimeout(pollingTimeoutId);
- teardownWebSocket();
+
+ config = appConfig || getBootstrapData().common.conf;
+
+ // (Re)connect the shared realtime socket whenever the websocket transport is
+ // enabled — independent of GLOBAL_ASYNC_QUERIES, since realtime list views
+ // (tier-1 entity-change nudges) ride the same socket. Idempotent: a no-op when
+ // WEBSOCKET_ENABLED is false, and supersedes any prior socket otherwise.
+ connectRealtime(config);
+
if (!isFeatureEnabled(FeatureFlag.GlobalAsyncQueries)) return;
const generation = pollingGeneration;
waitersByTaskId = new Map();
cursor = null;
- 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/middleware/realtime.test.ts b/superset-frontend/src/middleware/realtime.test.ts
new file mode 100644
index 00000000000..d6912d9e357
--- /dev/null
+++ b/superset-frontend/src/middleware/realtime.test.ts
@@ -0,0 +1,168 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import {
+ connectRealtime,
+ dispatchRealtimeMessage,
+ resetRealtimeForTests,
+ subscribeRealtime,
+ type RealtimeMessage,
+} from 'src/middleware/realtime';
+
+// A minimal fake WebSocket so connectRealtime can "open" a socket without a
+// real network connection. Instances register themselves so a test can drive
+// onmessage/onclose.
+class FakeWebSocket {
+ static instances: FakeWebSocket[] = [];
+
+ onmessage: ((event: { data: string }) => void) | null = null;
+
+ onclose: (() => void) | null = null;
+
+ onerror: (() => void) | null = null;
+
+ close = jest.fn();
+
+ constructor(public url: string) {
+ FakeWebSocket.instances.push(this);
+ }
+}
+
+const originalWebSocket = global.WebSocket;
+
+beforeEach(() => {
+ FakeWebSocket.instances = [];
+ (global as any).WebSocket = FakeWebSocket;
+});
+
+afterEach(() => {
+ resetRealtimeForTests();
+ (global as any).WebSocket = originalWebSocket;
+});
+
+const ENABLED = {
+ WEBSOCKET_ENABLED: true,
+ WEBSOCKET_URL: 'ws://localhost:8080/',
+};
+
+test('dispatches a parsed envelope to every subscribed handler', () => {
+ const a = jest.fn();
+ const b = jest.fn();
+ subscribeRealtime(a);
+ subscribeRealtime(b);
+
+ dispatchRealtimeMessage(
+ JSON.stringify({ channel: 'entity-changes:task', payload: { id: '7' } }),
+ );
+
+ const expected: RealtimeMessage = {
+ channel: 'entity-changes:task',
+ payload: { id: '7' },
+ };
+ expect(a).toHaveBeenCalledWith(expected);
+ expect(b).toHaveBeenCalledWith(expected);
+});
+
+test('unsubscribe stops a handler from receiving further messages', () => {
+ const handler = jest.fn();
+ const unsubscribe = subscribeRealtime(handler);
+
+ unsubscribe();
+ dispatchRealtimeMessage(
+ JSON.stringify({ channel: 'entity-changes:task', payload: {} }),
+ );
+
+ expect(handler).not.toHaveBeenCalled();
+});
+
+test('ignores malformed data and messages without a channel', () => {
+ const handler = jest.fn();
+ subscribeRealtime(handler);
+
+ expect(() => dispatchRealtimeMessage('not json')).not.toThrow();
+ dispatchRealtimeMessage(JSON.stringify({ payload: { id: '7' } })); // no channel
+
+ expect(handler).not.toHaveBeenCalled();
+});
+
+test('a throwing handler does not break other handlers', () => {
+ const bad = jest.fn(() => {
+ throw new Error('boom');
+ });
+ const good = jest.fn();
+ subscribeRealtime(bad);
+ subscribeRealtime(good);
+
+ expect(() =>
+ dispatchRealtimeMessage(
+ JSON.stringify({ channel: 'entity-changes:task', payload: {} }),
+ ),
+ ).not.toThrow();
+ expect(good).toHaveBeenCalledTimes(1);
+});
+
+test('does not open a socket when disabled', () => {
+ connectRealtime({ WEBSOCKET_ENABLED: false, WEBSOCKET_URL: 'ws://x/' });
+ expect(FakeWebSocket.instances).toHaveLength(0);
+});
+
+test('opens a socket when enabled and routes its messages to handlers', () => {
+ const handler = jest.fn();
+ subscribeRealtime(handler);
+ connectRealtime(ENABLED);
+
+ expect(FakeWebSocket.instances).toHaveLength(1);
+ const ws = FakeWebSocket.instances[0];
+ expect(ws.url).toBe(ENABLED.WEBSOCKET_URL);
+
+ ws.onmessage?.({
+ data: JSON.stringify({ channel: 'realtime:user:1', payload: { ok: true } }),
+ });
+
+ expect(handler).toHaveBeenCalledWith({
+ channel: 'realtime:user:1',
+ payload: { ok: true },
+ });
+});
+
+test('reconnects after the socket closes', () => {
+ jest.useFakeTimers();
+ connectRealtime(ENABLED);
+ expect(FakeWebSocket.instances).toHaveLength(1);
+
+ FakeWebSocket.instances[0].onclose?.();
+ jest.advanceTimersByTime(5000);
+
+ expect(FakeWebSocket.instances).toHaveLength(2);
+ jest.useRealTimers();
+});
+
+test('disconnect closes the socket and stops reconnecting', () => {
+ jest.useFakeTimers();
+ connectRealtime(ENABLED);
+ const ws = FakeWebSocket.instances[0];
+
+ // disconnect via a superseding reset; the prior socket must not reconnect.
+ resetRealtimeForTests();
+ expect(ws.close).toHaveBeenCalled();
+
+ ws.onclose?.(); // a late close from the torn-down socket
+ jest.advanceTimersByTime(5000);
+ expect(FakeWebSocket.instances).toHaveLength(1); // no new socket
+ jest.useRealTimers();
+});
diff --git a/superset-frontend/src/middleware/realtime.ts b/superset-frontend/src/middleware/realtime.ts
new file mode 100644
index 00000000000..5e960e2f7c1
--- /dev/null
+++ b/superset-frontend/src/middleware/realtime.ts
@@ -0,0 +1,179 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Shared client for the realtime WebSocket transport (`superset-websocket`).
+ *
+ * Owns the single browser socket and fans every message out to registered
+ * handlers, so multiple features share one connection: async chart-data
+ * completion (per-principal `realtime:*` messages, see `asyncEvent.ts`) and
+ * realtime list views (public `entity-changes:*` nudges, see
+ * `useListViewResource`). The server forwards a generic `{channel, payload}`
+ * envelope and this client is deliberately payload-agnostic — each handler
+ * routes on `channel` and interprets `payload` for its own feature.
+ *
+ * The transport is strictly best-effort: it is only an acceleration layer over
+ * each feature's own authorized fetch/poll, so a socket outage costs latency,
+ * not correctness. Connection auth is the `superset-ws-token` JWT cookie, which
+ * rides the handshake automatically (same-host requirement).
+ */
+import { logging } from '@apache-superset/core/utils';
+import getBootstrapData from 'src/utils/getBootstrapData';
+
+/** The generic envelope the server forwards to the browser. */
+export interface RealtimeMessage {
+ channel: string;
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
+ payload: any;
+}
+
+type RealtimeHandler = (message: RealtimeMessage) => void;
+
+type RealtimeConfig = {
+ WEBSOCKET_ENABLED?: boolean;
+ WEBSOCKET_URL?: string;
+};
+
+// Backoff before reconnecting after the socket closes (mirrors the Node
+// server's own subscribe retry). The socket is best-effort, so a generous
+// delay is fine — each feature's poll/fetch covers the gap.
+const RECONNECT_DELAY_MS = 5000;
+
+let socket: WebSocket | undefined;
+let reconnectTimeoutId: number;
+let enabled = false;
+let url: string | undefined;
+let started = false;
+// Bumped on every (re)configuration so an in-flight reconnect scheduled against
+// a superseded socket detects it is stale and stops.
+let generation = 0;
+const handlers = new Set<RealtimeHandler>();
+
+/**
+ * Parse a raw socket message and dispatch the `{channel, payload}` envelope to
+ * every handler. Exported for tests. Strictly best-effort: malformed data and a
+ * throwing handler are logged and swallowed, never propagated.
+ */
+export const dispatchRealtimeMessage = (rawData: string): void => {
+ let message: RealtimeMessage;
+ try {
+ const parsed = JSON.parse(rawData) ?? {};
+ if (typeof parsed.channel !== 'string') return;
+ message = { channel: parsed.channel, payload: parsed.payload };
+ } catch (err) {
+ logging.warn('Failed to parse realtime message', err);
+ return;
+ }
+ handlers.forEach(handler => {
+ try {
+ handler(message);
+ } catch (err) {
+ logging.warn('Realtime handler error', err);
+ }
+ });
+};
+
+const openSocket = (thisGeneration: number): void => {
+ if (thisGeneration !== generation) return;
+ if (!enabled || !url || typeof WebSocket === 'undefined') return;
+ let ws: WebSocket;
+ try {
+ ws = new WebSocket(url);
+ } catch (err) {
+ logging.warn('Failed to open realtime WebSocket', err);
+ return;
+ }
+ socket = ws;
+ ws.onmessage = (event: MessageEvent) => {
+ if (thisGeneration === generation)
+ dispatchRealtimeMessage(String(event.data));
+ };
+ ws.onclose = () => {
+ if (thisGeneration !== generation) return;
+ reconnectTimeoutId = window.setTimeout(
+ () => openSocket(thisGeneration),
+ RECONNECT_DELAY_MS,
+ );
+ };
+ // 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 = () => {};
+};
+
+const teardownSocket = (): void => {
+ if (reconnectTimeoutId) clearTimeout(reconnectTimeoutId);
+ if (socket) {
+ // Detach handlers first so the closing socket can't schedule a reconnect
+ // against the superseded generation.
+ socket.onmessage = null;
+ socket.onclose = null;
+ socket.onerror = null;
+ try {
+ socket.close();
+ } catch {
+ // ignore: closing an already-closed/broken socket is harmless
+ }
+ socket = undefined;
+ }
+};
+
+/**
+ * (Re)configure and (re)connect the shared socket. Idempotent and safe to call
+ * repeatedly (e.g. from app bootstrap): it supersedes any prior socket. Reads
+ * `WEBSOCKET_ENABLED` / `WEBSOCKET_URL` from bootstrap config when none is
+ * passed. A no-op (and tears down any existing socket) when disabled.
+ */
+export const connectRealtime = (config?: RealtimeConfig): void => {
+ const conf = config ?? getBootstrapData().common.conf;
+ teardownSocket();
+ generation += 1;
+ started = true;
+ enabled = Boolean(conf?.WEBSOCKET_ENABLED);
+ url = conf?.WEBSOCKET_URL;
+ openSocket(generation);
+};
+
+/** Tear down the socket and stop reconnecting. */
+export const disconnectRealtime = (): void => {
+ generation += 1;
+ teardownSocket();
+};
+
+/**
+ * Register a handler for every realtime message; returns an unsubscribe
+ * function. On the first subscription this lazily connects the socket from
+ * bootstrap config (unless `connectRealtime` was already called), so a feature
+ * can opt into realtime purely by subscribing.
+ */
+export const subscribeRealtime = (handler: RealtimeHandler): (() => void) => {
+ handlers.add(handler);
+ if (!started) connectRealtime();
+ return () => {
+ handlers.delete(handler);
+ };
+};
+
+// Test-only: reset module state between cases.
+export const resetRealtimeForTests = (): void => {
+ disconnectRealtime();
+ handlers.clear();
+ started = false;
+ enabled = false;
+ url = undefined;
+};
diff --git a/superset-frontend/src/pages/TaskList/index.tsx b/superset-frontend/src/pages/TaskList/index.tsx
index a1572240cec..0dfc81358f4 100644
--- a/superset-frontend/src/pages/TaskList/index.tsx
+++ b/superset-frontend/src/pages/TaskList/index.tsx
@@ -118,7 +118,18 @@ function TaskList({ addDangerToast, addSuccessToast, user }: TaskListProps) {
state: { loading, resourceCount: tasksCount, resourceCollection: tasks },
fetchData,
refreshData,
- } = useListViewResource<Task>('task', t('task'), addDangerToast);
+ } = useListViewResource<Task>(
+ 'task',
+ t('task'),
+ addDangerToast,
+ true, // infoEnable
+ [], // defaultCollectionValue
+ undefined, // baseFilters
+ true, // initialLoadingState
+ undefined, // selectColumns
+ true, // enableRealtime
+ 'uuid', // realtimeIdField (tasks are keyed by uuid)
+ );
// Get full user with roles to check admin status
const bootstrapData = getBootstrapData();
diff --git a/superset-frontend/src/views/CRUD/hooks.test.tsx b/superset-frontend/src/views/CRUD/hooks.test.tsx
index a48bc48fa36..becbdbd9ff0 100644
--- a/superset-frontend/src/views/CRUD/hooks.test.tsx
+++ b/superset-frontend/src/views/CRUD/hooks.test.tsx
@@ -27,6 +27,10 @@ import {
useChartEditModal,
} from './hooks';
import type Chart from 'src/types/Chart';
+import {
+ dispatchRealtimeMessage,
+ resetRealtimeForTests,
+} from 'src/middleware/realtime';
/** Find the endpoint string from a spy's mock calls that matches a substring. */
function findEndpoint(spy: jest.SpyInstance, substring: string): string {
@@ -47,6 +51,12 @@ beforeEach(() => {
jest.restoreAllMocks();
});
+afterEach(() => {
+ // Clear the shared realtime client's module-level handlers/socket so a
+ // realtime-enabled hook from one test can't leak into the next.
+ resetRealtimeForTests();
+});
+
// useListViewResource
test('useListViewResource: initial state has loading true and empty collection', () => {
jest.spyOn(SupersetClient, 'get').mockResolvedValue({
@@ -395,6 +405,106 @@ test('useListViewResource: uses desc sort direction when desc is true', async ()
expect(endpoint).toContain('order_direction:desc');
});
+test('useListViewResource: realtime nudge live-patches a displayed row in place', async () => {
+ const initial = [
+ { id: 1, name: 'A' },
+ { id: 2, name: 'B' },
+ ];
+ const getSpy = jest
+ .spyOn(SupersetClient, 'get')
+ .mockResolvedValueOnce({
+ json: { permissions: [] },
+ } as unknown as JsonResponse) // _info
+ .mockResolvedValueOnce({
+ json: { result: initial, count: 2 },
+ } as unknown as JsonResponse) // fetchData
+ .mockResolvedValueOnce({
+ json: { result: [{ id: 1, name: 'A-updated' }] },
+ } as unknown as JsonResponse); // batched realtime refetch
+
+ const { result } = renderHook(() =>
+ useListViewResource(
+ 'chart',
+ 'Charts',
+ jest.fn(),
+ true,
+ [],
+ undefined,
+ true,
+ undefined,
+ true, // enableRealtime
+ ),
+ );
+
+ await act(async () => {
+ await result.current.fetchData({
+ pageIndex: 0,
+ pageSize: 25,
+ sortBy: [{ id: 'id', desc: false }],
+ filters: [],
+ });
+ });
+ expect(result.current.state.resourceCollection).toEqual(initial);
+
+ // A nudge for a displayed row triggers one batched refetch (col:id op:in),
+ // and the fetched row is merged in place; the untouched row is unchanged.
+ act(() => {
+ dispatchRealtimeMessage(
+ JSON.stringify({ channel: 'entity-changes:chart', payload: { id: 1 } }),
+ );
+ });
+
+ await waitFor(() => {
+ expect(result.current.state.resourceCollection).toEqual([
+ { id: 1, name: 'A-updated' },
+ { id: 2, name: 'B' },
+ ]);
+ });
+ const endpoint = findEndpoint(getSpy, 'col:id');
+ expect(endpoint).toContain('opr:in');
+});
+
+test('useListViewResource: realtime ignores nudges for rows not on screen', async () => {
+ const initial = [{ id: 1, name: 'A' }];
+ const getSpy = jest.spyOn(SupersetClient, 'get').mockResolvedValue({
+ json: { result: initial, count: 1 },
+ } as unknown as JsonResponse);
+
+ const { result } = renderHook(() =>
+ useListViewResource(
+ 'chart',
+ 'Charts',
+ jest.fn(),
+ true,
+ [],
+ undefined,
+ true,
+ undefined,
+ true,
+ ),
+ );
+
+ await act(async () => {
+ await result.current.fetchData({
+ pageIndex: 0,
+ pageSize: 25,
+ sortBy: [{ id: 'id', desc: false }],
+ filters: [],
+ });
+ });
+
+ const callsBefore = getSpy.mock.calls.length;
+ act(() => {
+ dispatchRealtimeMessage(
+ JSON.stringify({ channel: 'entity-changes:chart', payload: { id: 999 } }),
+ );
+ });
+
+ // No id=999 row is displayed, so no batched refetch is scheduled.
+ await new Promise(resolve => setTimeout(resolve, 600));
+ expect(getSpy.mock.calls.length).toBe(callsBefore);
+});
+
// useSingleViewResource
test('useSingleViewResource: initial state has loading false and null resource', () => {
const { result } = renderHook(() =>
diff --git a/superset-frontend/src/views/CRUD/hooks.ts b/superset-frontend/src/views/CRUD/hooks.ts
index 6969f3d7cfb..6cccaad5819 100644
--- a/superset-frontend/src/views/CRUD/hooks.ts
+++ b/superset-frontend/src/views/CRUD/hooks.ts
@@ -45,12 +45,20 @@ import copyTextToClipboard from 'src/utils/copy';
import { getShareableUrl } from 'src/utils/navigationUtils';
import SupersetText from 'src/utils/textUtils';
import { DatabaseObject } from 'src/features/databases/types';
+import {
+ subscribeRealtime,
+ type RealtimeMessage,
+} from 'src/middleware/realtime';
import {
FavoriteStatus,
FileEncryptedExtraFields,
ImportResourceName,
} from './types';
+// Realtime list views coalesce a burst of entity-change nudges into at most one
+// batched row fetch per this window, bounding backend load under heavy churn.
+const REALTIME_REFETCH_DEBOUNCE_MS = 500;
+
interface ListViewResourceState<D extends object = any> {
loading: boolean;
collection: D[];
@@ -86,6 +94,13 @@ export function useListViewResource<D extends object = any>(
baseFilters?: FilterValue[], // must be memoized
initialLoadingState = true,
selectColumns?: string[],
+ // Realtime: when enabled, the list subscribes to the entity-change nudge
+ // channel and live-patches only its currently-displayed rows (see the
+ // realtime effect below). `realtimeIdField` is the row identity used both to
+ // match nudges against the collection and to filter the batched refetch
+ // (default the integer `id`; UUID-facing resources like tasks pass `uuid`).
+ enableRealtime = false,
+ realtimeIdField = 'id',
) {
const [state, setState] = useState<ListViewResourceState<D>>({
count: 0,
@@ -239,6 +254,77 @@ export function useListViewResource<D extends object = any>(
[fetchData],
);
+ // Realtime list updates: on an entity-change nudge for a row currently on
+ // screen, debounce-collect the ids and batch-refetch just those rows through
+ // the normal authorized list endpoint (authz/RLS still apply), merging them in
+ // place. Update-only — no new-row insertion — and best-effort, so a downed
+ // socket just defers the update to the next normal fetch.
+ const collectionRef = useRef(state.collection);
+ collectionRef.current = state.collection;
+
+ useEffect(() => {
+ if (!enableRealtime) return undefined;
+
+ const channel = `entity-changes:${resource}`;
+ const pendingIds = new Set<string>();
+ let debounceId: ReturnType<typeof setTimeout> | undefined;
+
+ const rowId = (row: D): string =>
+ String((row as Record<string, unknown>)[realtimeIdField]);
+ const isDisplayed = (id: string): boolean =>
+ collectionRef.current.some(row => rowId(row) === id);
+
+ const flush = () => {
+ debounceId = undefined;
+ // Re-check membership: the collection may have changed (paged/filtered)
+ // since the nudges arrived, so only fetch rows still on screen.
+ const ids = [...pendingIds].filter(isDisplayed);
+ pendingIds.clear();
+ if (!ids.length) return;
+
+ const query = rison.encode_uri({
+ filters: [{ col: realtimeIdField, opr: 'in', value: ids }],
+ page: 0,
+ page_size: ids.length,
+ });
+ SupersetClient.get({ endpoint: `/api/v1/${resource}/?q=${query}` })
+ .then(({ json = {} }) => {
+ const fetched: D[] = json.result ?? [];
+ if (!fetched.length) return;
+ const byId = new Map(fetched.map(row => [rowId(row), row]));
+ // Replace matched rows in place; keep order, count, and every
+ // untouched row reference (so React only re-renders changed rows).
+ setState(current => ({
+ ...current,
+ collection: current.collection.map(
+ row => byId.get(rowId(row)) ?? row,
+ ),
+ }));
+ })
+ .catch(() => {
+ // Best-effort: a failed patch just leaves the stale rows until the
+ // next normal fetch; never surface a toast for a background refresh.
+ });
+ };
+
+ const unsubscribe = subscribeRealtime((message: RealtimeMessage) => {
+ if (message.channel !== channel) return;
+ const id = message.payload?.id;
+ if (id == null) return;
+ const idStr = String(id);
+ if (!isDisplayed(idStr)) return;
+ pendingIds.add(idStr);
+ if (debounceId === undefined) {
+ debounceId = setTimeout(flush, REALTIME_REFETCH_DEBOUNCE_MS);
+ }
+ });
+
+ return () => {
+ unsubscribe();
+ if (debounceId !== undefined) clearTimeout(debounceId);
+ };
+ }, [enableRealtime, resource, realtimeIdField]);
+
return {
state: {
loading: state.loading,
diff --git a/superset/tasks/api.py b/superset/tasks/api.py
index cd092379efe..ae2916f1c11 100644
--- a/superset/tasks/api.py
+++ b/superset/tasks/api.py
@@ -131,6 +131,7 @@ class TaskRestApi(BaseSupersetModelRestApi):
"status",
"created_by",
"created_on",
+ "uuid",
]
base_order = ("created_on", "desc")