(superset) 01/01: fix(gtf): close async-chart-data waiter race + widen guest_key column

[vi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

villebro pushed a commit to branch villebro/gtf-review-fixes
in repository https://gitbox.apache.org/repos/asf/superset.git

commit 0528372174856cacc64e54c7ac2ffbde67e53e02
Author: Ville Brofeldt <vi...@gmail.com>
AuthorDate: Mon Aug 24 07:23:02 2026 -0700

    fix(gtf): close async-chart-data waiter race + widen guest_key column
    
    Two correctness fixes from the PR #43407 review:
    
    - P1 (async chart-data can hang): waitForAsyncData awaited baselineReady before
      registering the task waiter. The 202 is returned when tasks are *scheduled*,
      not finished, so at that moment the shared poll cursor is <= now < any task's
      future terminal — but awaiting between the 202 and registration opened a gap
      in which a fast task could finish and a concurrent chart's poll advance the
      cursor past its terminal update (and the socket event is dropped with no
      waiter yet), hanging the chart/native filter forever. Register the waiter
      synchronously in the same tick as the 202 (no await first); the poll then
      always observes the completion. Removed the now-unused baselineReady handle.
    
    - P1 (guest subscriber key too long): superset.tasks.guest returns "guest-" +
      a 64-char SHA256 hex digest = 70 chars, but task_subscribers.guest_key was
      String(64), so a guest-created async task's subscription would error on
      Postgres / truncate on MySQL. Widen the column to String(128) in the model
      and the (feature-branch) migration.
    
    Tests: async-registration is exercised by delivering the completion in the same
    tick with no wait; a full-length (70-char) guest key round-trips through the DAO.
---
 .../src/middleware/asyncEvent.test.ts              | 12 ++++--------
 superset-frontend/src/middleware/asyncEvent.ts     | 12 +++++++++---
 ..._7e2c9a4f1b83_create_task_dependencies_table.py |  2 +-
 superset/models/task_subscribers.py                |  4 +++-
 tests/unit_tests/daos/test_tasks.py                | 22 ++++++++++++++++++++++
 5 files changed, 39 insertions(+), 13 deletions(-)

diff --git a/superset-frontend/src/middleware/asyncEvent.test.ts b/superset-frontend/src/middleware/asyncEvent.test.ts
index 75507e85080..070cb2c0279 100644
--- a/superset-frontend/src/middleware/asyncEvent.test.ts
+++ b/superset-frontend/src/middleware/asyncEvent.test.ts
@@ -164,11 +164,6 @@ describe('realtime WebSocket acceleration', () => {
     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.
@@ -181,7 +176,10 @@ describe('realtime WebSocket acceleration', () => {
       refetch,
     );
 
-    await afterRegistered();
+    // Deliver the completion in the SAME tick, with no wait: waitForAsyncData
+    // registers its waiter synchronously (before any await), so a fast task
+    // whose event arrives immediately after the 202 is never missed. (Regression
+    // guard: registration must not sit behind an awaited baseline fetch.)
     asyncEvent.handleRealtimeMessage(realtime('task-1', 'success'));
 
     expect(await promise).toEqual([{ rows: 1 }]);
@@ -198,7 +196,6 @@ describe('realtime WebSocket acceleration', () => {
       refetch,
     );
 
-    await afterRegistered();
     asyncEvent.handleRealtimeMessage(realtime('task-1', 'failure'));
 
     await expect(promise).rejects.toThrow();
@@ -215,7 +212,6 @@ describe('realtime WebSocket acceleration', () => {
       refetch,
     );
 
-    await afterRegistered();
     // A public entity-change nudge carries no status and must not settle a chart.
     asyncEvent.handleRealtimeMessage({
       channel: 'entity-changes:task',
diff --git a/superset-frontend/src/middleware/asyncEvent.ts b/superset-frontend/src/middleware/asyncEvent.ts
index 3c62b1ed00a..4c9d0dd2e08 100644
--- a/superset-frontend/src/middleware/asyncEvent.ts
+++ b/superset-frontend/src/middleware/asyncEvent.ts
@@ -84,7 +84,6 @@ let waitersByTaskId: Map<string, Set<Waiter>> = new Map();
 // is triggered) so no task created afterwards is missed, then advanced by each
 // poll. Always the server's own clock, never the browser's.
 let cursor: string | null;
-let baselineReady: Promise<void> | null;
 // Incremented on every init() so an in-flight poll can detect it is stale and
 // stop instead of scheduling a second loop or mutating fresh state.
 let pollingGeneration = 0;
@@ -217,8 +216,15 @@ export const waitForAsyncData = async <T = unknown[]>(
   signal?: AbortSignal,
 ): Promise<T> => {
   const taskIds = asyncJob.task_ids ?? [];
-  if (baselineReady) await baselineReady;
 
+  // Register the waiter synchronously, in the same tick the 202 was received —
+  // NOT after an await. The 202 is returned when the tasks are *scheduled*, not
+  // finished, so at this point the shared poll cursor is <= now < any task's
+  // future terminal transition; the poll is therefore guaranteed to observe the
+  // completion. Awaiting anything here (e.g. the init baseline) would open a gap
+  // in which a fast task could finish and a concurrent chart's poll advance the
+  // cursor past its terminal update, and the socket event (no waiter yet) would
+  // be dropped — hanging the request.
   await new Promise<void>((resolve, reject) => {
     if (signal?.aborted) {
       taskIds.forEach(cancelTask);
@@ -281,7 +287,7 @@ export const init = (appConfig?: AppConfig) => {
   // 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.
-  baselineReady = fetchStatusChanges({ task_type: CHART_QUERY_TASK_TYPE })
+  fetchStatusChanges({ task_type: CHART_QUERY_TASK_TYPE })
     .then(({ cursor: baseline }) => {
       if (generation === pollingGeneration) cursor = baseline;
     })
diff --git a/superset/migrations/versions/2026-08-21_12-00_7e2c9a4f1b83_create_task_dependencies_table.py b/superset/migrations/versions/2026-08-21_12-00_7e2c9a4f1b83_create_task_dependencies_table.py
index 9c70c3aac2c..6a0fe43b045 100644
--- a/superset/migrations/versions/2026-08-21_12-00_7e2c9a4f1b83_create_task_dependencies_table.py
+++ b/superset/migrations/versions/2026-08-21_12-00_7e2c9a4f1b83_create_task_dependencies_table.py
@@ -139,7 +139,7 @@ def upgrade():
     # not collide.)
     add_columns(
         TASK_SUBSCRIBERS_TABLE,
-        Column("guest_key", sa.String(length=64), nullable=True),
+        Column("guest_key", sa.String(length=128), nullable=True),
     )
     with op.batch_alter_table(TASK_SUBSCRIBERS_TABLE) as batch_op:
         batch_op.alter_column("user_id", existing_type=sa.Integer(), nullable=True)
diff --git a/superset/models/task_subscribers.py b/superset/models/task_subscribers.py
index f56f4621447..475f655fc05 100644
--- a/superset/models/task_subscribers.py
+++ b/superset/models/task_subscribers.py
@@ -63,7 +63,9 @@ class TaskSubscriber(CoreTaskSubscriber, AuditMixinNullable, Model):
     user_id = Column(
         Integer, ForeignKey("ab_user.id", ondelete="CASCADE"), nullable=True
     )
-    guest_key = Column(String(64), nullable=True, index=True)
+    # A guest key is ``guest-`` + a 64-char SHA256 hex digest (70 chars); the
+    # column is sized with headroom (see superset.tasks.guest).
+    guest_key = Column(String(128), nullable=True, index=True)
     subscribed_at = Column(DateTime, nullable=False, default=datetime.now(timezone.utc))
 
     # Relationships
diff --git a/tests/unit_tests/daos/test_tasks.py b/tests/unit_tests/daos/test_tasks.py
index ca10970e643..dde92a17874 100644
--- a/tests/unit_tests/daos/test_tasks.py
+++ b/tests/unit_tests/daos/test_tasks.py
@@ -378,6 +378,28 @@ def test_get_subscriber_channels(session_with_task: Session) -> None:
     assert set(channels) == {"user:7", "guest-abc"}
 
 
+def test_add_guest_subscriber_full_length_key(session_with_task: Session) -> None:
+    """A realistic guest key (``guest-`` + 64-char SHA256 hex = 70 chars) persists.
+
+    Regression guard for the guest_key column width: the real key from
+    superset.tasks.guest is 70 chars, so a String(64) column would error/truncate.
+    """
+    from superset.daos.tasks import TaskDAO
+
+    task = create_task(
+        session_with_task,
+        task_key="shared-guest-len",
+        scope=TaskScope.SHARED,
+        user_id=None,
+    )
+    guest_key = "guest-" + "a" * 64  # mirrors superset.tasks.guest (70 chars)
+
+    assert TaskDAO.add_guest_subscriber(task.id, guest_key=guest_key) is True
+
+    session_with_task.refresh(task)
+    assert task.subscribers[0].guest_key == guest_key
+
+
 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