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/24 20:02:08 UTC
(superset) branch villebro/gtf-poll-backoff-duration created (now c2968dd4373)
This is an automated email from the ASF dual-hosted git repository.
villebro pushed a change to branch villebro/gtf-poll-backoff-duration
in repository https://gitbox.apache.org/repos/asf/superset.git
at c2968dd4373 fix(gtf): stop async poll when idle + back off; second-granularity durations
This branch includes the following new commits:
new c2968dd4373 fix(gtf): stop async poll when idle + back off; second-granularity durations
The 1 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) 01/01: fix(gtf): stop async poll when idle + back off; second-granularity durations
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-poll-backoff-duration
in repository https://gitbox.apache.org/repos/asf/superset.git
commit c2968dd4373ecf81ec59b27585758caa08924167
Author: Ville Brofeldt <vi...@gmail.com>
AuthorDate: Mon Aug 24 13:01:24 2026 -0700
fix(gtf): stop async poll when idle + back off; second-granularity durations
Task-list/polling polish surfaced while testing on Postgres:
- Stop the chart-data status poll entirely once every awaited task settles
(no perpetual idle heartbeat), and restart it — eager again — when a new
waiter registers.
- Exponential backoff: the poll starts at the eager interval, doubles (capped at
8× the base) while awaited tasks are quiet, and snaps back to eager the moment
an awaited task changes or a new waiter registers. Keeps a long-running task
responsive without hammering the API while it's quiet.
- Duration display: render at second granularity — sub-second precision
(283ms 200μs) is noise for task tracking, so anything under a second shows
"<1s" and larger durations round to whole seconds.
- Don't show a duration for pending tasks: a task that hasn't started has no
execution time (queue time is not duration), so duration_seconds returns None
until it starts.
---
.../src/features/tasks/timeUtils.test.ts | 7 +-
superset-frontend/src/features/tasks/timeUtils.ts | 24 +++----
.../src/middleware/asyncEvent.test.ts | 43 +++++++++++++
superset-frontend/src/middleware/asyncEvent.ts | 74 ++++++++++++++++++----
superset/models/tasks.py | 20 ++----
tests/unit_tests/tasks/test_handlers.py | 6 +-
6 files changed, 132 insertions(+), 42 deletions(-)
diff --git a/superset-frontend/src/features/tasks/timeUtils.test.ts b/superset-frontend/src/features/tasks/timeUtils.test.ts
index 0514b426852..68b6e412b37 100644
--- a/superset-frontend/src/features/tasks/timeUtils.test.ts
+++ b/superset-frontend/src/features/tasks/timeUtils.test.ts
@@ -30,8 +30,13 @@ test('formatDuration returns null for invalid inputs', () => {
expect(formatDuration(-5)).toBeNull();
});
+test('formatDuration collapses sub-second durations to <1s', () => {
+ expect(formatDuration(0.283)).toBe('<1s');
+ expect(formatDuration(0.999)).toBe('<1s');
+});
+
test('formatDuration formats seconds correctly', () => {
- expect(formatDuration(37.5)).toBe('37s 500ms');
+ expect(formatDuration(37.5)).toBe('38s'); // rounded to whole seconds
expect(formatDuration(1)).toBe('1s');
expect(formatDuration(30)).toBe('30s');
});
diff --git a/superset-frontend/src/features/tasks/timeUtils.ts b/superset-frontend/src/features/tasks/timeUtils.ts
index bf4e2e13e85..b608a0637d7 100644
--- a/superset-frontend/src/features/tasks/timeUtils.ts
+++ b/superset-frontend/src/features/tasks/timeUtils.ts
@@ -52,9 +52,12 @@ function getCachedDurationFormatter(locale?: string): Intl.DurationFormat {
/**
* Format a duration in seconds to a human-readable string.
*
+ * Rendered at second granularity — sub-second precision (ms/μs/ns) is noise for
+ * task tracking, so anything under a second collapses to "<1s".
+ *
* @param seconds - Duration in seconds
* @param locale - Current locale
- * @returns Formatted string like "1m 30s" or "2h 15m", or null if invalid
+ * @returns Formatted string like "1m 30s" or "2h 15m", "<1s", or null if invalid
*/
export function formatDuration(
seconds: number | null | undefined,
@@ -64,17 +67,14 @@ export function formatDuration(
return null;
}
- const durObject = parseMilliseconds(seconds * 1000);
- const unitOrder = [
- 'years',
- 'days',
- 'hours',
- 'minutes',
- 'seconds',
- 'milliseconds',
- 'microseconds',
- 'nanoseconds',
- ] as const;
+ // Below a second there's nothing useful to show at second granularity.
+ if (seconds < 1) {
+ return '<1s';
+ }
+
+ // Round to whole seconds so no ms/μs/ns units are ever emitted.
+ const durObject = parseMilliseconds(Math.round(seconds) * 1000);
+ const unitOrder = ['years', 'days', 'hours', 'minutes', 'seconds'] as const;
const nonZeroUnits = unitOrder
.filter(unit => durObject[unit] > 0)
.slice(0, 2)
diff --git a/superset-frontend/src/middleware/asyncEvent.test.ts b/superset-frontend/src/middleware/asyncEvent.test.ts
index ad631099476..724bd42b040 100644
--- a/superset-frontend/src/middleware/asyncEvent.test.ts
+++ b/superset-frontend/src/middleware/asyncEvent.test.ts
@@ -177,6 +177,49 @@ test('polls from the pre-task cursor returned in the 202', async () => {
expect(polled).toBe(true);
});
+test('stops polling once all awaited tasks settle', async () => {
+ queueStatuses({ 'task-1': { status: 'success' } });
+ asyncEvent.init(config);
+
+ await asyncEvent.waitForAsyncData(
+ { task_ids: ['task-1'] },
+ jest.fn().mockResolvedValue([]),
+ );
+
+ // The loop must go fully idle — no heartbeat polling after everything settled.
+ const pollsAtSettle = fetchMock.callHistory.calls(
+ STATUS_CHANGES_ENDPOINT,
+ ).length;
+ await new Promise(resolve => {
+ setTimeout(resolve, config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY * 12);
+ });
+ expect(fetchMock.callHistory.calls(STATUS_CHANGES_ENDPOINT).length).toBe(
+ pollsAtSettle,
+ );
+});
+
+test('restarts polling for a new job after going idle', async () => {
+ queueStatuses({ 'task-1': { status: 'success' } });
+ asyncEvent.init(config);
+
+ await asyncEvent.waitForAsyncData(
+ { task_ids: ['task-1'] },
+ jest.fn().mockResolvedValue([]),
+ );
+ // Loop is idle now. A second job must wake it and resolve.
+ statusResponses.push({
+ statuses: { 'task-2': { status: 'success' } },
+ cursor: '2020-01-01T00:00:09',
+ });
+ const refetch = jest.fn().mockResolvedValue([{ rows: 9 }]);
+ const result = await asyncEvent.waitForAsyncData(
+ { task_ids: ['task-2'] },
+ refetch,
+ );
+
+ expect(result).toEqual([{ rows: 9 }]);
+});
+
describe('realtime WebSocket acceleration', () => {
const realtime = (taskId: string, status: string) => ({
channel: `realtime:user:1`,
diff --git a/superset-frontend/src/middleware/asyncEvent.ts b/superset-frontend/src/middleware/asyncEvent.ts
index d39bfc0d522..e42bc6fc89a 100644
--- a/superset-frontend/src/middleware/asyncEvent.ts
+++ b/superset-frontend/src/middleware/asyncEvent.ts
@@ -72,7 +72,16 @@ type Waiter = {
let config: AppConfig;
let pollingDelayMs: number;
+// Backoff state: the poll starts eager (`pollingDelayMs`), degrades — doubling up
+// to `pollBackoffMaxMs` — while awaited tasks are quiet, and snaps back to eager
+// the moment an awaited task changes or a new waiter registers. Capped so a very
+// slow task is still noticed within `pollBackoffMaxMs`.
+let currentPollDelayMs: number;
+let pollBackoffMaxMs: number;
let pollingTimeoutId: number;
+// Whether the poll loop is running. It stops entirely when no waiters remain (no
+// idle heartbeat) and is restarted by ``ensurePolling`` when a waiter registers.
+let pollingActive = false;
// Registry of in-flight waiters keyed by every task uuid they await, so a single
// shared poll loop fans status changes out to whichever requests are awaiting them.
// A SHARED task can be deduplicated across concurrent chart requests, so each task
@@ -155,30 +164,66 @@ const applyStatus = (taskId: string, status: string) => {
};
const loadStatusChanges = async (generation: number) => {
- if (generation !== pollingGeneration) return;
+ if (generation !== pollingGeneration) {
+ pollingActive = false;
+ return;
+ }
if (waitersByTaskId.size) {
try {
const { statuses, cursor: next } = await fetchStatusChanges({
cursor,
task_type: CHART_QUERY_TASK_TYPE,
});
- if (generation !== pollingGeneration) return;
+ if (generation !== pollingGeneration) {
+ pollingActive = false;
+ return;
+ }
cursor = next;
- Object.entries(statuses).forEach(([taskId, { status }]) =>
- applyStatus(taskId, status),
- );
+ // "Progress" = the batch carried a change for a task we're awaiting; check
+ // membership before applyStatus, which deletes a settled task's waiters.
+ let progressed = false;
+ Object.entries(statuses).forEach(([taskId, { status }]) => {
+ if (waitersByTaskId.has(taskId)) progressed = true;
+ applyStatus(taskId, status);
+ });
+ // Reset to eager on an awaited-task change; otherwise back off (bounded).
+ currentPollDelayMs = progressed
+ ? pollingDelayMs
+ : Math.min(currentPollDelayMs * 2, pollBackoffMaxMs);
} catch (err) {
- if (generation !== pollingGeneration) return;
+ if (generation !== pollingGeneration) {
+ pollingActive = false;
+ return;
+ }
logging.warn(err);
}
}
// Reschedule from the tail so a slow request never overlaps the next tick.
+ // Nothing left to await → stop the loop entirely (no idle heartbeat); the next
+ // waiter restarts it from the eager interval via ensurePolling.
+ if (!waitersByTaskId.size) {
+ pollingActive = false;
+ currentPollDelayMs = pollingDelayMs;
+ return;
+ }
+ pollingActive = true;
pollingTimeoutId = window.setTimeout(
() => loadStatusChanges(generation),
- pollingDelayMs,
+ currentPollDelayMs,
);
};
+// Start (or wake) the poll loop for a freshly registered waiter: poll eagerly
+// again, and kick the loop if it had gone idle. Idempotent — a no-op while the
+// loop is already running or when async queries are disabled.
+const ensurePolling = () => {
+ currentPollDelayMs = pollingDelayMs;
+ if (!pollingActive && isFeatureEnabled(FeatureFlag.GlobalAsyncQueries)) {
+ pollingActive = true;
+ loadStatusChanges(pollingGeneration);
+ }
+};
+
/**
* Handle a realtime message from the shared client (src/middleware/realtime.ts).
*
@@ -269,6 +314,8 @@ export const waitForAsyncData = async <T = unknown[]>(
}
waiters.add(waiter);
});
+ // Wake the poll loop (eager again) now that there's something to await.
+ ensurePolling();
});
return refetch();
@@ -291,12 +338,18 @@ export const init = (appConfig?: AppConfig) => {
const generation = pollingGeneration;
waitersByTaskId = new Map();
cursor = null;
+ pollingActive = false;
pollingDelayMs = config.GLOBAL_ASYNC_QUERIES_POLLING_DELAY || 500;
+ // Cap the backoff at 8× the eager interval so even a very slow task is noticed
+ // within that bound (the socket, when enabled, accelerates it further).
+ pollBackoffMaxMs = pollingDelayMs * 8;
+ currentPollDelayMs = pollingDelayMs;
// 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
- // shared poll loop from that watermark.
+ // created afterwards are all caught by the changed-since poll. The loop itself
+ // stays idle until the first waiter registers (ensurePolling), then stops again
+ // once every awaited task settles.
fetchStatusChanges({ task_type: CHART_QUERY_TASK_TYPE })
.then(({ cursor: baseline }) => {
// Seed the initial cursor only; never overwrite a value a waiter already
@@ -308,9 +361,6 @@ export const init = (appConfig?: AppConfig) => {
// A missing baseline just means the first poll starts from "everything
// changed so far"; the >= cursor semantics still catch our tasks.
logging.warn('Failed to fetch async baseline cursor', err);
- })
- .finally(() => {
- loadStatusChanges(generation);
});
};
diff --git a/superset/models/tasks.py b/superset/models/tasks.py
index 5707373d1e1..1facfa2a166 100644
--- a/superset/models/tasks.py
+++ b/superset/models/tasks.py
@@ -268,14 +268,15 @@ class Task(CoreTask, AuditMixinNullable, Model):
@property
def duration_seconds(self) -> float | None:
"""
- Get task duration in seconds.
+ Get task duration in seconds (execution time), or ``None`` before a task
+ has started.
- Finished tasks: Time from started_at to ended_at (None if never started)
- Running/aborting tasks: Time from started_at to now
- - Pending tasks: Time from created_on to now (queue time)
+ - Pending tasks: None — a task that hasn't started has no duration to show
+ (queue time is not execution time)
- Note: started_at/ended_at are stored in UTC, but created_on from
- AuditMixinNullable is stored as naive local time. We handle both cases.
+ Note: started_at/ended_at are stored in UTC.
"""
if self.is_finished:
# Task has completed - use fixed timestamps, never increment
@@ -295,16 +296,7 @@ class Task(CoreTask, AuditMixinNullable, Model):
else self.started_at
)
return (now - started).total_seconds()
- elif self.created_on:
- # Pending - created_on is naive LOCAL time (from AuditMixinNullable)
- # Use naive local time for comparison
- now = datetime.now() # Local time, no timezone
- created = (
- self.created_on.replace(tzinfo=None)
- if self.created_on.tzinfo is not None
- else self.created_on
- )
- return (now - created).total_seconds()
+ # Pending (not yet started) - no duration.
return None
# Scope-related properties
diff --git a/tests/unit_tests/tasks/test_handlers.py b/tests/unit_tests/tasks/test_handlers.py
index 0105c55ce96..0d696b08bfa 100644
--- a/tests/unit_tests/tasks/test_handlers.py
+++ b/tests/unit_tests/tasks/test_handlers.py
@@ -240,7 +240,7 @@ class TestTaskDuration:
@freeze_time("2024-01-01 10:00:15")
def test_duration_seconds_pending_task(self):
- """Test duration for pending task returns queue time."""
+ """A pending (not-yet-started) task has no duration to show."""
from superset.models.tasks import Task
task = Task()
@@ -248,8 +248,8 @@ class TestTaskDuration:
task.started_at = None
task.ended_at = None
- # 15 seconds since creation
- assert task.duration_seconds == 15.0
+ # Not started yet → None (queue time is not execution time).
+ assert task.duration_seconds is None
def test_duration_seconds_no_timestamps(self):
"""Test duration returns None when no timestamps available."""